diff --git a/test/test_plugin_fcm.py b/test/test_plugin_fcm.py index 34b6e61f..d5487ab2 100644 --- a/test/test_plugin_fcm.py +++ b/test/test_plugin_fcm.py @@ -59,6 +59,8 @@ logging.disable(logging.CRITICAL) # Test files for KeyFile Directory PRIVATE_KEYFILE_DIR = os.path.join(os.path.dirname(__file__), 'var', 'fcm') +FCM_KEYFILE = os.path.join(PRIVATE_KEYFILE_DIR, 'service_account.json') + # Our Testing URLs apprise_url_tests = ( @@ -135,13 +137,13 @@ apprise_url_tests = ( 'instance': TypeError, }), ('fcm://project_id?to=device&keyfile=/invalid/path', { - # Test to= and auto detection of oauth mode + # Test to= and auto-detection of OAuth mode 'instance': NotifyFCM, # we'll fail to send our notification as a result 'response': False, }), ('fcm://?to=device&project=project_id&keyfile=/invalid/path', { - # Test project= & to= and auto detection of oauth mode + # Test project= & to= and auto detection of OAuth mode 'instance': NotifyFCM, # we'll fail to send our notification as a result 'response': False, @@ -152,21 +154,21 @@ apprise_url_tests = ( }), ('fcm://project_id?to=device&mode=oauth2&keyfile=/invalid/path', { # Same test as above except we explicitly set our oauth2 mode - # Test to= and auto detection of oauth mode + # Test to= and auto-detection of OAuth mode 'instance': NotifyFCM, # we'll fail to send our notification as a result 'response': False, }), ('fcm://apikey/#topic1/device/?mode=legacy', { 'instance': NotifyFCM, - # throw a bizzare code forcing us to fail to look it up + # throw a bizarre code forcing us to fail to look it up 'response': False, 'requests_response_code': 999, }), ('fcm://apikey/#topic1/device/?mode=legacy', { 'instance': NotifyFCM, # Throws a series of connection and transfer exceptions when this flag - # is set and tests that we gracfully handle them + # is set and tests that we gracefully handle them 'test_requests_exceptions': True, }), ('fcm://project/#topic1/device/?mode=oauth2&keyfile=file://{}'.format( @@ -174,7 +176,7 @@ apprise_url_tests = ( os.path.dirname(__file__), 'var', 'fcm', 'service_account.json')), { 'instance': NotifyFCM, - # throw a bizzare code forcing us to fail to look it up + # throw a bizarre code forcing us to fail to look it up 'response': False, 'requests_response_code': 999, }), @@ -184,12 +186,47 @@ apprise_url_tests = ( 'service_account.json')), { 'instance': NotifyFCM, # Throws a series of connection and transfer exceptions when - # this flag is set and tests that we gracfully handle them + # this flag is set and tests that we gracefully handle them 'test_requests_exceptions': True, }), ) +@pytest.fixture +def mock_post(mocker): + """ + Prepare a good OAuth mock response. + """ + + mock_thing = mocker.patch("requests.post") + + response = mock.Mock() + response.content = json.dumps({ + "access_token": "ya29.c.abcd", + "expires_in": 3599, + "token_type": "Bearer", + }) + response.status_code = requests.codes.ok + mock_thing.return_value = response + + return mock_thing + + +@pytest.fixture +def mock_post_legacy(mocker): + """ + Prepare a good legacy mock response. + """ + + mock_thing = mocker.patch("requests.post") + + response = mock.Mock() + response.status_code = requests.codes.ok + mock_thing.return_value = response + + return mock_thing + + @pytest.mark.skipif( 'cryptography' not in sys.modules, reason="Requires cryptography") def test_plugin_fcm_urls(): @@ -204,20 +241,11 @@ def test_plugin_fcm_urls(): @pytest.mark.skipif( 'cryptography' not in sys.modules, reason="Requires cryptography") -@pytest.mark.skipif( - hasattr(sys, "pypy_version_info"), reason="Does not work reliably on PyPy") -@mock.patch('requests.post') -def test_plugin_fcm_general_legacy(mock_post): +def test_plugin_fcm_legacy_default(mock_post_legacy): """ - NotifyFCM() General Legacy/APIKey Checks - + NotifyFCM() Legacy/APIKey default checks. """ - # Prepare a good response - response = mock.Mock() - response.status_code = requests.codes.ok - mock_post.return_value = response - # A valid Legacy URL obj = Apprise.instantiate( 'fcm://abc123/device/' @@ -228,11 +256,11 @@ def test_plugin_fcm_general_legacy(mock_post): assert obj.notify("test") is True # Test our call count - assert mock_post.call_count == 1 - assert mock_post.call_args_list[0][0][0] == \ + assert mock_post_legacy.call_count == 1 + assert mock_post_legacy.call_args_list[0][0][0] == \ 'https://fcm.googleapis.com/fcm/send' - payload = mock_post.mock_calls[0][2] + payload = mock_post_legacy.mock_calls[0][2] data = json.loads(payload['data']) assert 'data' in data assert isinstance(data, dict) @@ -249,23 +277,27 @@ def test_plugin_fcm_general_legacy(mock_post): assert data['notification']['notification']['image'] == \ 'https://example.com/interesting.png' - # - # Test priorities - # - mock_post.reset_mock() + +@pytest.mark.skipif( + 'cryptography' not in sys.modules, reason="Requires cryptography") +def test_plugin_fcm_legacy_priorities(mock_post_legacy): + """ + NotifyFCM() Legacy/APIKey priorities checks. + """ + obj = Apprise.instantiate( 'fcm://abc123/device/?priority=low') - assert mock_post.call_count == 0 + assert mock_post_legacy.call_count == 0 # Send our notification assert obj.notify(title="title", body="body") is True # Test our call count - assert mock_post.call_count == 1 - assert mock_post.call_args_list[0][0][0] == \ + assert mock_post_legacy.call_count == 1 + assert mock_post_legacy.call_args_list[0][0][0] == \ 'https://fcm.googleapis.com/fcm/send' - payload = mock_post.mock_calls[0][2] + payload = mock_post_legacy.mock_calls[0][2] data = json.loads(payload['data']) assert 'data' not in data assert 'notification' in data @@ -278,23 +310,27 @@ def test_plugin_fcm_general_legacy(mock_post): # legacy can only switch between high/low assert data['priority'] == "normal" - # - # Test colors - # - mock_post.reset_mock() + +@pytest.mark.skipif( + 'cryptography' not in sys.modules, reason="Requires cryptography") +def test_plugin_fcm_legacy_no_colors(mock_post_legacy): + """ + NotifyFCM() Legacy/APIKey `color=no` checks. + """ + obj = Apprise.instantiate( 'fcm://abc123/device/?color=no') - assert mock_post.call_count == 0 + assert mock_post_legacy.call_count == 0 # Send our notification assert obj.notify(title="title", body="body") is True # Test our call count - assert mock_post.call_count == 1 - assert mock_post.call_args_list[0][0][0] == \ + assert mock_post_legacy.call_count == 1 + assert mock_post_legacy.call_args_list[0][0][0] == \ 'https://fcm.googleapis.com/fcm/send' - payload = mock_post.mock_calls[0][2] + payload = mock_post_legacy.mock_calls[0][2] data = json.loads(payload['data']) assert 'data' not in data assert 'notification' in data @@ -304,20 +340,27 @@ def test_plugin_fcm_general_legacy(mock_post): assert 'image' not in data['notification']['notification'] assert 'color' not in data['notification']['notification'] - mock_post.reset_mock() + +@pytest.mark.skipif( + 'cryptography' not in sys.modules, reason="Requires cryptography") +def test_plugin_fcm_legacy_colors(mock_post_legacy): + """ + NotifyFCM() Legacy/APIKey colors checks. + """ + obj = Apprise.instantiate( 'fcm://abc123/device/?color=AA001b') - assert mock_post.call_count == 0 + assert mock_post_legacy.call_count == 0 # Send our notification assert obj.notify(title="title", body="body") is True # Test our call count - assert mock_post.call_count == 1 - assert mock_post.call_args_list[0][0][0] == \ + assert mock_post_legacy.call_count == 1 + assert mock_post_legacy.call_args_list[0][0][0] == \ 'https://fcm.googleapis.com/fcm/send' - payload = mock_post.mock_calls[0][2] + payload = mock_post_legacy.mock_calls[0][2] data = json.loads(payload['data']) assert 'data' not in data assert 'notification' in data @@ -331,38 +374,56 @@ def test_plugin_fcm_general_legacy(mock_post): @pytest.mark.skipif( 'cryptography' not in sys.modules, reason="Requires cryptography") -@mock.patch('requests.post') -def test_plugin_fcm_general_oauth(mock_post): +def test_plugin_fcm_oauth_default(mock_post): """ - NotifyFCM() General OAuth Checks - + NotifyFCM() general OAuth checks - success. + Test using a valid Project ID and key file. """ - # Valid Keyfile - path = os.path.join(PRIVATE_KEYFILE_DIR, 'service_account.json') + obj = Apprise.instantiate( + f'fcm://mock-project-id/device/#topic/?keyfile={FCM_KEYFILE}') - # Prepare a good response - response = mock.Mock() - response.content = json.dumps({ - "access_token": "ya29.c.abcd", - "expires_in": 3599, - "token_type": "Bearer", - }) - response.status_code = requests.codes.ok - mock_post.return_value = response + # send our notification + assert obj.notify("test") is True - # Test having a valid keyfile, but not a valid project id match + # Test our call count + assert mock_post.call_count == 3 + assert mock_post.call_args_list[0][0][0] == \ + 'https://accounts.google.com/o/oauth2/token' + assert mock_post.call_args_list[1][0][0] == \ + 'https://fcm.googleapis.com/v1/projects/mock-project-id/messages:send' + assert mock_post.call_args_list[2][0][0] == \ + 'https://fcm.googleapis.com/v1/projects/mock-project-id/messages:send' + + +@pytest.mark.skipif( + 'cryptography' not in sys.modules, reason="Requires cryptography") +def test_plugin_fcm_oauth_invalid_project_id(mock_post): + """ + NotifyFCM() OAuth checks, with invalid project id. + """ + + # Test having a valid keyfile, but not a valid project id match. obj = Apprise.instantiate( - 'fcm://invalid_project_id/device/?keyfile={}'.format(str(path))) + f'fcm://invalid_project_id/device/?keyfile={FCM_KEYFILE}') + # we'll fail as a result assert obj.notify("test") is False # Test our call count assert mock_post.call_count == 0 + +@pytest.mark.skipif( + 'cryptography' not in sys.modules, reason="Requires cryptography") +def test_plugin_fcm_oauth_keyfile_error(mock_post): + """ + NotifyFCM() OAuth checks, while unable to read key file. + """ + # Now we test using a valid Project ID but we can't open our file obj = Apprise.instantiate( - 'fcm://mock-project-id/device/?keyfile={}'.format(str(path))) + f'fcm://mock-project-id/device/?keyfile={FCM_KEYFILE}') with mock.patch('builtins.open', side_effect=OSError): # we'll fail as a result @@ -371,28 +432,19 @@ def test_plugin_fcm_general_oauth(mock_post): # Test our call count assert mock_post.call_count == 0 - # Now we test using a valid Project ID - obj = Apprise.instantiate( - 'fcm://mock-project-id/device/#topic/?keyfile={}'.format(str(path))) - # send our notification - assert obj.notify("test") is True - - # Test our call count - assert mock_post.call_count == 3 - assert mock_post.call_args_list[0][0][0] == \ - 'https://accounts.google.com/o/oauth2/token' - assert mock_post.call_args_list[1][0][0] == \ - 'https://fcm.googleapis.com/v1/projects/mock-project-id/messages:send' - assert mock_post.call_args_list[2][0][0] == \ - 'https://fcm.googleapis.com/v1/projects/mock-project-id/messages:send' +@pytest.mark.skipif( + 'cryptography' not in sys.modules, reason="Requires cryptography") +def test_plugin_fcm_oauth_data_parameters(mock_post): + """ + NotifyFCM() OAuth checks, success. + Test using a valid Project ID and data parameters. + """ - mock_post.reset_mock() - # Now we test using a valid Project ID and data parameters obj = Apprise.instantiate( - 'fcm://mock-project-id/device/#topic/?keyfile={}' + f'fcm://mock-project-id/device/#topic/?keyfile={FCM_KEYFILE}' '&+key=value&+key2=value2' - '&image_url=https://example.com/interesting.png'.format(str(path))) + '&image_url=https://example.com/interesting.png') assert mock_post.call_count == 0 # send our notification @@ -442,13 +494,17 @@ def test_plugin_fcm_general_oauth(mock_post): assert data['message']['notification']['image'] == \ 'https://example.com/interesting.png' - # - # Test priorities - # - mock_post.reset_mock() + +@pytest.mark.skipif( + 'cryptography' not in sys.modules, reason="Requires cryptography") +def test_plugin_fcm_oauth_priorities(mock_post): + """ + Verify priorities work as intended. + """ + obj = Apprise.instantiate( - 'fcm://mock-project-id/device/?keyfile={}' - '&priority=high'.format(str(path))) + f'fcm://mock-project-id/device/?keyfile={FCM_KEYFILE}' + '&priority=high') assert mock_post.call_count == 0 # Send our notification @@ -473,13 +529,17 @@ def test_plugin_fcm_general_oauth(mock_post): assert data['message']['webpush']['headers']['Urgency'] == "high" assert data['message']['android']['priority'] == "HIGH" - # - # Test colors - # - mock_post.reset_mock() + +@pytest.mark.skipif( + 'cryptography' not in sys.modules, reason="Requires cryptography") +def test_plugin_fcm_oauth_no_colors(mock_post): + """ + Verify `color=no` work as intended. + """ + obj = Apprise.instantiate( - 'fcm://mock-project-id/device/?keyfile={}' - '&color=no'.format(str(path))) + f'fcm://mock-project-id/device/?keyfile={FCM_KEYFILE}' + '&color=no') assert mock_post.call_count == 0 # Send our notification @@ -501,10 +561,17 @@ def test_plugin_fcm_general_oauth(mock_post): assert isinstance(data['message']['notification'], dict) assert 'color' not in data['message']['notification'] - mock_post.reset_mock() + +@pytest.mark.skipif( + 'cryptography' not in sys.modules, reason="Requires cryptography") +def test_plugin_fcm_oauth_colors(mock_post): + """ + Verify colors work as intended. + """ + obj = Apprise.instantiate( - 'fcm://mock-project-id/device/?keyfile={}' - '&color=#12AAbb'.format(str(path))) + f'fcm://mock-project-id/device/?keyfile={FCM_KEYFILE}' + '&color=#12AAbb') assert mock_post.call_count == 0 # Send our notification @@ -530,29 +597,17 @@ def test_plugin_fcm_general_oauth(mock_post): @pytest.mark.skipif( 'cryptography' not in sys.modules, reason="Requires cryptography") -@mock.patch('requests.post') -def test_plugin_fcm_keyfile_parse(mock_post): +def test_plugin_fcm_keyfile_parse_default(mock_post): """ NotifyFCM() KeyFile Tests """ - # Prepare a good response - response = mock.Mock() - response.content = json.dumps({ - "access_token": "ya29.c.abcd", - "expires_in": 3599, - "token_type": "Bearer", - }) - response.status_code = requests.codes.ok - mock_post.return_value = response - - path = os.path.join(PRIVATE_KEYFILE_DIR, 'service_account.json') oauth = GoogleOAuth() # We can not get an Access Token without content loaded assert oauth.access_token is None # Load our content - assert oauth.load(path) is True + assert oauth.load(FCM_KEYFILE) is True assert oauth.access_token is not None # Test our call count @@ -561,19 +616,26 @@ def test_plugin_fcm_keyfile_parse(mock_post): 'https://accounts.google.com/o/oauth2/token' mock_post.reset_mock() + # a second call uses cache since our token hasn't expired yet assert oauth.access_token is not None assert mock_post.call_count == 0 - # Same test case without expires_in entry - mock_post.reset_mock() - response.content = json.dumps({ + +@pytest.mark.skipif( + 'cryptography' not in sys.modules, reason="Requires cryptography") +def test_plugin_fcm_keyfile_parse_no_expiry(mock_post): + """ + Test case without `expires_in` entry. + """ + + mock_post.return_value.content = json.dumps({ "access_token": "ya29.c.abcd", "token_type": "Bearer", }) oauth = GoogleOAuth() - assert oauth.load(path) is True + assert oauth.load(FCM_KEYFILE) is True assert oauth.access_token is not None # Test our call count @@ -581,35 +643,42 @@ def test_plugin_fcm_keyfile_parse(mock_post): assert mock_post.call_args_list[0][0][0] == \ 'https://accounts.google.com/o/oauth2/token' - # Test user-agent override - mock_post.reset_mock() + +@pytest.mark.skipif( + 'cryptography' not in sys.modules, reason="Requires cryptography") +def test_plugin_fcm_keyfile_parse_user_agent(mock_post): + """ + Test case with `user-agent` override. + """ + oauth = GoogleOAuth(user_agent="test-agent-override") - assert oauth.load(path) is True + assert oauth.load(FCM_KEYFILE) is True assert oauth.access_token is not None assert mock_post.call_count == 1 assert mock_post.call_args_list[0][0][0] == \ 'https://accounts.google.com/o/oauth2/token' - # - # Test some errors that can get thrown when trying to handle - # the service_account.json file - # - # Reset our object - mock_post.reset_mock() +@pytest.mark.skipif( + 'cryptography' not in sys.modules, reason="Requires cryptography") +def test_plugin_fcm_keyfile_parse_keyfile_failures(mock_post: mock.Mock): + """ + Test some errors that can get thrown when trying to handle + the `service_account.json` file. + """ # Now we test a case where we can't access the file we've been pointed to: oauth = GoogleOAuth() with mock.patch('builtins.open', side_effect=OSError): # We will fail to retrieve our Access Token - assert oauth.load(path) is False + assert oauth.load(FCM_KEYFILE) is False assert oauth.access_token is None oauth = GoogleOAuth() with mock.patch('json.loads', side_effect=([], )): # We will fail to retrieve our Access Token since we did not parse # a dictionary - assert oauth.load(path) is False + assert oauth.load(FCM_KEYFILE) is False assert oauth.access_token is None # Case where we can't load the PEM key: @@ -618,7 +687,7 @@ def test_plugin_fcm_keyfile_parse(mock_post): 'cryptography.hazmat.primitives.serialization' '.load_pem_private_key', side_effect=ValueError("")): - assert oauth.load(path) is False + assert oauth.load(FCM_KEYFILE) is False assert oauth.access_token is None # Case where we can't load the PEM key: @@ -627,7 +696,7 @@ def test_plugin_fcm_keyfile_parse(mock_post): 'cryptography.hazmat.primitives.serialization' '.load_pem_private_key', side_effect=TypeError("")): - assert oauth.load(path) is False + assert oauth.load(FCM_KEYFILE) is False assert oauth.access_token is None # Case where we can't load the PEM key: @@ -637,27 +706,31 @@ def test_plugin_fcm_keyfile_parse(mock_post): '.load_pem_private_key', side_effect=UnsupportedAlgorithm("")): # Note: This test should be te - assert oauth.load(path) is False + assert oauth.load(FCM_KEYFILE) is False assert oauth.access_token is None - # Not one call was made to the web - assert mock_post.call_count == 0 + # Verify that not a single call to the web escaped the test harness. + assert mock_post.mock_calls == [] - # - # Test some web errors that can occur when speaking upstream - # with Google to get our token generated - # - response.status_code = requests.codes.internal_server_error - mock_post.reset_mock() +@pytest.mark.skipif( + 'cryptography' not in sys.modules, reason="Requires cryptography") +def test_plugin_fcm_keyfile_parse_token_failures(mock_post): + """ + Test some web errors that can occur when speaking upstream + with Google to get our token generated. + """ + + mock_post.return_value.status_code = requests.codes.internal_server_error + oauth = GoogleOAuth() - assert oauth.load(path) is True + assert oauth.load(FCM_KEYFILE) is True # We'll fail due to an bad web response assert oauth.access_token is None # Return our status code to how it was - response.status_code = requests.codes.ok + mock_post.return_value.status_code = requests.codes.ok # No access token bad_response_1 = mock.Mock() @@ -678,7 +751,7 @@ def test_plugin_fcm_keyfile_parse(mock_post): # Test all of our bad side effects oauth = GoogleOAuth() - assert oauth.load(path) is True + assert oauth.load(FCM_KEYFILE) is True # We'll fail due to an bad web response assert oauth.access_token is None @@ -738,7 +811,7 @@ def test_plugin_fcm_keyfile_missing_entries_parse(tmpdir): @pytest.mark.skipif( 'cryptography' not in sys.modules, reason="Requires cryptography") -def test_plugin_fcm_priorities(): +def test_plugin_fcm_priority_manager(): """ NotifyFCM() FCMPriorityManager() Testing """ @@ -773,7 +846,7 @@ def test_plugin_fcm_priorities(): @pytest.mark.skipif( 'cryptography' not in sys.modules, reason="Requires cryptography") -def test_plugin_fcm_colors(): +def test_plugin_fcm_color_manager(): """ NotifyFCM() FCMColorManager() Testing """ diff --git a/test/test_plugin_msteams.py b/test/test_plugin_msteams.py index 8d2d5857..181d290e 100644 --- a/test/test_plugin_msteams.py +++ b/test/test_plugin_msteams.py @@ -25,7 +25,6 @@ from unittest import mock -import sys import json import requests import pytest @@ -175,24 +174,51 @@ def test_plugin_msteams_urls(): AppriseURLTester(tests=apprise_url_tests).run_all() -@pytest.mark.skipif( - hasattr(sys, "pypy_version_info") and sys.version_info < (3, 7), - reason="Does not work or is flaky on PyPy 3.6") -@mock.patch('requests.post') -def test_plugin_msteams_templating(mock_post, tmpdir): - """ - NotifyMSTeams() Templating +@pytest.fixture +def msteams_url(): + return 'msteams://{}@{}/{}/{}'.format(UUID4, UUID4, 'a' * 32, UUID4) - """ - # Prepare Mock +@pytest.fixture +def request_mock(mocker): + """ + Prepare requests mock. + """ + mock_post = mocker.patch("requests.post") mock_post.return_value = requests.Request() mock_post.return_value.status_code = requests.codes.ok + return mock_post - uuid4 = '8b799edf-6f98-4d3a-9be7-2862fb4e5752' - url = 'msteams://{}@{}/{}/{}'.format(uuid4, uuid4, 'a' * 32, uuid4) - # Test cases where our URL is invalid +@pytest.fixture +def simple_template(tmpdir): + # Test cases where our URL is invalid. + template = tmpdir.join("simple.json") + template.write(""" + { + "@type": "MessageCard", + "@context": "https://schema.org/extensions", + "summary": "{{name}}", + "themeColor": "{{app_color}}", + "sections": [ + { + "activityImage": null, + "activityTitle": "{{title}}", + "text": "{{body}}" + } + ] + } + """) + return template + + +def test_plugin_msteams_templating_basic_success( + request_mock, msteams_url, tmpdir): + """ + NotifyMSTeams() Templating - success. + Test cases where URL and JSON is valid. + """ + template = tmpdir.join("simple.json") template.write(""" { @@ -212,7 +238,7 @@ def test_plugin_msteams_templating(mock_post, tmpdir): # Instantiate our URL obj = Apprise.instantiate('{url}/?template={template}&{kwargs}'.format( - url=url, + url=msteams_url, template=str(template), kwargs=':key1=token&:key2=token', )) @@ -222,27 +248,31 @@ def test_plugin_msteams_templating(mock_post, tmpdir): body="body", title='title', notify_type=NotifyType.INFO) is True - assert mock_post.called is True - assert mock_post.call_args_list[0][0][0].startswith( + assert request_mock.called is True + assert request_mock.call_args_list[0][0][0].startswith( 'https://outlook.office.com/webhook/') # Our Posted JSON Object - posted_json = json.loads(mock_post.call_args_list[0][1]['data']) + posted_json = json.loads(request_mock.call_args_list[0][1]['data']) assert 'summary' in posted_json assert posted_json['summary'] == 'Apprise' assert posted_json['themeColor'] == '#3AA3E3' assert posted_json['sections'][0]['activityTitle'] == 'title' assert posted_json['sections'][0]['text'] == 'body' - # Test invalid JSON - # Test cases where our URL is invalid +def test_plugin_msteams_templating_invalid_json( + request_mock, msteams_url, tmpdir): + """ + NotifyMSTeams() Templating - invalid JSON. + """ + template = tmpdir.join("invalid.json") template.write("}") # Instantiate our URL obj = Apprise.instantiate('{url}/?template={template}&{kwargs}'.format( - url=url, + url=msteams_url, template=str(template), kwargs=':key1=token&:key2=token', )) @@ -253,7 +283,14 @@ def test_plugin_msteams_templating(mock_post, tmpdir): body="body", title='title', notify_type=NotifyType.INFO) is False - # Test cases where we're missing the @type part of the URL + +def test_plugin_msteams_templating_json_missing_type( + request_mock, msteams_url, tmpdir): + """ + NotifyMSTeams() Templating - invalid JSON. + Test case where we're missing the @type part of the URL. + """ + template = tmpdir.join("missing_type.json") template.write(""" { @@ -272,7 +309,7 @@ def test_plugin_msteams_templating(mock_post, tmpdir): # Instantiate our URL obj = Apprise.instantiate('{url}/?template={template}&{kwargs}'.format( - url=url, + url=msteams_url, template=str(template), kwargs=':key1=token&:key2=token', )) @@ -284,7 +321,14 @@ def test_plugin_msteams_templating(mock_post, tmpdir): body="body", title='title', notify_type=NotifyType.INFO) is False - # Test cases where we're missing the @context part of the URL + +def test_plugin_msteams_templating_json_missing_context( + request_mock, msteams_url, tmpdir): + """ + NotifyMSTeams() Templating - invalid JSON. + Test cases where we're missing the @context part of the URL. + """ + template = tmpdir.join("missing_context.json") template.write(""" { @@ -303,18 +347,33 @@ def test_plugin_msteams_templating(mock_post, tmpdir): # Instantiate our URL obj = Apprise.instantiate('{url}/?template={template}&{kwargs}'.format( - url=url, + url=msteams_url, template=str(template), kwargs=':key1=token&:key2=token', )) - assert isinstance(obj, NotifyMSTeams) + # We can not load the file because we're missing the @context entry assert obj.notify( body="body", title='title', notify_type=NotifyType.INFO) is False - # Test a case where we can not access the file: + +def test_plugin_msteams_templating_load_json_failure( + request_mock, msteams_url, tmpdir): + """ + NotifyMSTeams() Templating - template loading failure. + Test a case where we can not access the file. + """ + + template = tmpdir.join("empty.json") + template.write("") + + obj = Apprise.instantiate('{url}/?template={template}'.format( + url=msteams_url, + template=str(template), + )) + with mock.patch('json.loads', side_effect=OSError): # we fail, but this time it's because we couldn't # access the cached file contents for reading @@ -322,8 +381,14 @@ def test_plugin_msteams_templating(mock_post, tmpdir): body="body", title='title', notify_type=NotifyType.INFO) is False - # A more complicated example; uses a target - mock_post.reset_mock() + +def test_plugin_msteams_templating_target_success( + request_mock, msteams_url, tmpdir): + """ + NotifyMSTeams() Templating - success with target. + A more complicated example; uses a target. + """ + template = tmpdir.join("more_complicated_example.json") template.write(""" { @@ -358,7 +423,7 @@ def test_plugin_msteams_templating(mock_post, tmpdir): # Instantiate our URL obj = Apprise.instantiate('{url}/?template={template}&{kwargs}'.format( - url=url, + url=msteams_url, template=str(template), kwargs=':key1=token&:key2=token&:target=http://localhost', )) @@ -368,12 +433,12 @@ def test_plugin_msteams_templating(mock_post, tmpdir): body="body", title='title', notify_type=NotifyType.INFO) is True - assert mock_post.called is True - assert mock_post.call_args_list[0][0][0].startswith( + assert request_mock.called is True + assert request_mock.call_args_list[0][0][0].startswith( 'https://outlook.office.com/webhook/') # Our Posted JSON Object - posted_json = json.loads(mock_post.call_args_list[0][1]['data']) + posted_json = json.loads(request_mock.call_args_list[0][1]['data']) assert 'summary' in posted_json assert posted_json['summary'] == 'Apprise Notifications' assert posted_json['themeColor'] == '#3AA3E3' @@ -385,41 +450,12 @@ def test_plugin_msteams_templating(mock_post, tmpdir): == 'http://localhost' -@pytest.mark.skipif( - hasattr(sys, "pypy_version_info"), reason="Does not work reliably on PyPy") -@mock.patch('requests.post') -def test_msteams_yaml_config(mock_post, tmpdir): +def test_msteams_yaml_config_invalid_template_filename( + request_mock, msteams_url, simple_template, tmpdir): """ - NotifyMSTeams() YAML Configuration Entries - + NotifyMSTeams() YAML Configuration Entries - invalid template filename. """ - # Prepare Mock - mock_post.return_value = requests.Request() - mock_post.return_value.status_code = requests.codes.ok - - uuid4 = '8b799edf-6f98-4d3a-9be7-2862fb4e5752' - url = 'msteams://{}@{}/{}/{}'.format(uuid4, uuid4, 'a' * 32, uuid4) - - # Test cases where our URL is invalid - template = tmpdir.join("simple.json") - template.write(""" - { - "@type": "MessageCard", - "@context": "https://schema.org/extensions", - "summary": "{{name}}", - "themeColor": "{{app_color}}", - "sections": [ - { - "activityImage": null, - "activityTitle": "{{title}}", - "text": "{{body}}" - } - ] - } - """) - - # Test Invalid Filename config = tmpdir.join("msteams01.yml") config.write(""" urls: @@ -429,7 +465,7 @@ def test_msteams_yaml_config(mock_post, tmpdir): :name: 'Template.Missing' :body: 'test body' :title: 'test title' - """.format(url=url, template=str(template))) + """.format(url=msteams_url, template=str(simple_template))) cfg = AppriseConfig() cfg.add(str(config)) @@ -441,9 +477,15 @@ def test_msteams_yaml_config(mock_post, tmpdir): assert obj.notify( body="body", title='title', notify_type=NotifyType.INFO) is False - assert mock_post.called is False + assert request_mock.called is False + + +def test_msteams_yaml_config_token_identifiers( + request_mock, msteams_url, simple_template, tmpdir): + """ + NotifyMSTeams() YAML Configuration Entries - test token identifiers. + """ - # Test token identifiers config = tmpdir.join("msteams01.yml") config.write(""" urls: @@ -453,7 +495,7 @@ def test_msteams_yaml_config(mock_post, tmpdir): :name: 'Testing' :body: 'test body' :title: 'test title' - """.format(url=url, template=str(template))) + """.format(url=msteams_url, template=str(simple_template))) cfg = AppriseConfig() cfg.add(str(config)) @@ -466,22 +508,26 @@ def test_msteams_yaml_config(mock_post, tmpdir): body="body", title='title', notify_type=NotifyType.INFO) is True - assert mock_post.called is True - assert mock_post.call_args_list[0][0][0].startswith( + assert request_mock.called is True + assert request_mock.call_args_list[0][0][0].startswith( 'https://outlook.office.com/webhook/') # Our Posted JSON Object - posted_json = json.loads(mock_post.call_args_list[0][1]['data']) + posted_json = json.loads(request_mock.call_args_list[0][1]['data']) assert 'summary' in posted_json assert posted_json['summary'] == 'Testing' assert posted_json['themeColor'] == '#3AA3E3' assert posted_json['sections'][0]['activityTitle'] == 'test title' assert posted_json['sections'][0]['text'] == 'test body' - # - # Now again but without a bullet under the url definition - # - mock_post.reset_mock() + +def test_msteams_yaml_config_no_bullet_under_url_1( + request_mock, msteams_url, simple_template, tmpdir): + """ + NotifyMSTeams() YAML Configuration Entries - no bullet 1. + Now again but without a bullet under the url definition. + """ + config = tmpdir.join("msteams02.yml") config.write(""" urls: @@ -491,7 +537,7 @@ def test_msteams_yaml_config(mock_post, tmpdir): :name: 'Testing2' :body: 'test body2' :title: 'test title2' - """.format(url=url, template=str(template))) + """.format(url=msteams_url, template=str(simple_template))) cfg = AppriseConfig() cfg.add(str(config)) @@ -504,22 +550,26 @@ def test_msteams_yaml_config(mock_post, tmpdir): body="body", title='title', notify_type=NotifyType.INFO) is True - assert mock_post.called is True - assert mock_post.call_args_list[0][0][0].startswith( + assert request_mock.called is True + assert request_mock.call_args_list[0][0][0].startswith( 'https://outlook.office.com/webhook/') # Our Posted JSON Object - posted_json = json.loads(mock_post.call_args_list[0][1]['data']) + posted_json = json.loads(request_mock.call_args_list[0][1]['data']) assert 'summary' in posted_json assert posted_json['summary'] == 'Testing2' assert posted_json['themeColor'] == '#3AA3E3' assert posted_json['sections'][0]['activityTitle'] == 'test title2' assert posted_json['sections'][0]['text'] == 'test body2' - # - # Try again but store the content as a dictionary in the cofiguration file - # - mock_post.reset_mock() + +def test_msteams_yaml_config_dictionary_file( + request_mock, msteams_url, simple_template, tmpdir): + """ + NotifyMSTeams() YAML Configuration Entries. + Try again but store the content as a dictionary in the configuration file. + """ + config = tmpdir.join("msteams03.yml") config.write(""" urls: @@ -530,7 +580,7 @@ def test_msteams_yaml_config(mock_post, tmpdir): name: 'Testing3' body: 'test body3' title: 'test title3' - """.format(url=url, template=str(template))) + """.format(url=msteams_url, template=str(simple_template))) cfg = AppriseConfig() cfg.add(str(config)) @@ -543,22 +593,26 @@ def test_msteams_yaml_config(mock_post, tmpdir): body="body", title='title', notify_type=NotifyType.INFO) is True - assert mock_post.called is True - assert mock_post.call_args_list[0][0][0].startswith( + assert request_mock.called is True + assert request_mock.call_args_list[0][0][0].startswith( 'https://outlook.office.com/webhook/') # Our Posted JSON Object - posted_json = json.loads(mock_post.call_args_list[0][1]['data']) + posted_json = json.loads(request_mock.call_args_list[0][1]['data']) assert 'summary' in posted_json assert posted_json['summary'] == 'Testing3' assert posted_json['themeColor'] == '#3AA3E3' assert posted_json['sections'][0]['activityTitle'] == 'test title3' assert posted_json['sections'][0]['text'] == 'test body3' - # - # Now again but without a bullet under the url definition - # - mock_post.reset_mock() + +def test_msteams_yaml_config_no_bullet_under_url_2( + request_mock, msteams_url, simple_template, tmpdir): + """ + NotifyMSTeams() YAML Configuration Entries - no bullet 2. + Now again but without a bullet under the url definition. + """ + config = tmpdir.join("msteams04.yml") config.write(""" urls: @@ -569,7 +623,7 @@ def test_msteams_yaml_config(mock_post, tmpdir): name: 'Testing4' body: 'test body4' title: 'test title4' - """.format(url=url, template=str(template))) + """.format(url=msteams_url, template=str(simple_template))) cfg = AppriseConfig() cfg.add(str(config)) @@ -582,20 +636,26 @@ def test_msteams_yaml_config(mock_post, tmpdir): body="body", title='title', notify_type=NotifyType.INFO) is True - assert mock_post.called is True - assert mock_post.call_args_list[0][0][0].startswith( + assert request_mock.called is True + assert request_mock.call_args_list[0][0][0].startswith( 'https://outlook.office.com/webhook/') # Our Posted JSON Object - posted_json = json.loads(mock_post.call_args_list[0][1]['data']) + posted_json = json.loads(request_mock.call_args_list[0][1]['data']) assert 'summary' in posted_json assert posted_json['summary'] == 'Testing4' assert posted_json['themeColor'] == '#3AA3E3' assert posted_json['sections'][0]['activityTitle'] == 'test title4' assert posted_json['sections'][0]['text'] == 'test body4' - # Now let's do a combination of the two - mock_post.reset_mock() + +def test_msteams_yaml_config_combined( + request_mock, msteams_url, simple_template, tmpdir): + """ + NotifyMSTeams() YAML Configuration Entries. + Now let's do a combination of the two. + """ + config = tmpdir.join("msteams05.yml") config.write(""" urls: @@ -606,7 +666,7 @@ def test_msteams_yaml_config(mock_post, tmpdir): body: 'test body5' title: 'test title5' :name: 'Testing5' - """.format(url=url, template=str(template))) + """.format(url=msteams_url, template=str(simple_template))) cfg = AppriseConfig() cfg.add(str(config)) @@ -619,21 +679,27 @@ def test_msteams_yaml_config(mock_post, tmpdir): body="body", title='title', notify_type=NotifyType.INFO) is True - assert mock_post.called is True - assert mock_post.call_args_list[0][0][0].startswith( + assert request_mock.called is True + assert request_mock.call_args_list[0][0][0].startswith( 'https://outlook.office.com/webhook/') # Our Posted JSON Object - posted_json = json.loads(mock_post.call_args_list[0][1]['data']) + posted_json = json.loads(request_mock.call_args_list[0][1]['data']) assert 'summary' in posted_json assert posted_json['summary'] == 'Testing5' assert posted_json['themeColor'] == '#3AA3E3' assert posted_json['sections'][0]['activityTitle'] == 'test title5' assert posted_json['sections'][0]['text'] == 'test body5' - # Now let's do a test where our tokens is not the expected - # dictionary we want to see - mock_post.reset_mock() + +def test_msteams_yaml_config_token_mismatch( + request_mock, msteams_url, simple_template, tmpdir): + """ + NotifyMSTeams() YAML Configuration Entries. + Now let's do a test where our tokens is not the + expected dictionary we want to see. + """ + config = tmpdir.join("msteams06.yml") config.write(""" urls: @@ -643,7 +709,7 @@ def test_msteams_yaml_config(mock_post, tmpdir): # Not a dictionary tokens: body - """.format(url=url, template=str(template))) + """.format(url=msteams_url, template=str(simple_template))) cfg = AppriseConfig() cfg.add(str(config)) diff --git a/test/test_plugin_twitter.py b/test/test_plugin_twitter.py index 5a7dfcc2..87aef9b3 100644 --- a/test/test_plugin_twitter.py +++ b/test/test_plugin_twitter.py @@ -23,14 +23,15 @@ # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN # THE SOFTWARE. +import json +import logging import os -import sys -from unittest import mock +from datetime import datetime +from unittest.mock import Mock, patch import pytest import requests -from json import dumps -from datetime import datetime + from apprise import Apprise from apprise import NotifyType from apprise import AppriseAttachment @@ -38,12 +39,14 @@ from apprise.plugins.NotifyTwitter import NotifyTwitter from helpers import AppriseURLTester # Disable logging for a cleaner testing output -import logging logging.disable(logging.CRITICAL) # Attachment Directory TEST_VAR_DIR = os.path.join(os.path.dirname(__file__), 'var') +TWITTER_SCREEN_NAME = 'apprise' + + # Our Testing URLs apprise_url_tests = ( ################################## @@ -200,39 +203,141 @@ apprise_url_tests = ( ) +def good_response(data): + """ + Prepare a good response. + """ + response = Mock() + response.content = json.dumps(data) + response.status_code = requests.codes.ok + return response + + +def bad_response(data): + """ + Prepare a bad response. + """ + response = Mock() + response.content = json.dumps(data) + response.status_code = requests.codes.internal_server_error + return response + + +@pytest.fixture +def twitter_url(): + ckey = 'ckey' + csecret = 'csecret' + akey = 'akey' + asecret = 'asecret' + url = 'twitter://{}/{}/{}/{}'.format(ckey, csecret, akey, asecret) + return url + + +@pytest.fixture +def good_message_response(): + """ + Prepare a good tweet response. + """ + response = good_response({ + 'screen_name': TWITTER_SCREEN_NAME, + 'id': 9876, + }) + return response + + +@pytest.fixture +def bad_message_response(): + """ + Prepare a bad message response. + """ + response = bad_response({ + "errors": [ + { + "code": 999, + "message": "Something failed", + }]}) + return response + + +@pytest.fixture +def good_media_response(): + """ + Prepare a good media response. + """ + response = Mock() + response.content = json.dumps({ + "media_id": 710511363345354753, + "media_id_string": "710511363345354753", + "media_key": "3_710511363345354753", + "size": 11065, + "expires_after_secs": 86400, + "image": { + "image_type": "image/jpeg", + "w": 800, + "h": 320 + } + }) + response.status_code = requests.codes.ok + return response + + +@pytest.fixture +def bad_media_response(): + """ + Prepare a bad media response. + """ + response = bad_response({ + "errors": [ + { + "code": 93, + "message": "This application is not allowed to access or " + "delete your direct messages.", + }]}) + return response + + +@pytest.fixture(autouse=True) +def ensure_get_verify_credentials_is_mocked(mocker, good_message_response): + """ + Make sure requests to https://api.twitter.com/1.1/account/verify_credentials.json + do not escape the test harness, for all test case functions. + """ # noqa:E501 + mock_get = mocker.patch("requests.get") + mock_get.return_value = good_message_response + + def test_plugin_twitter_urls(): """ NotifyTwitter() Apprise URLs - """ # Run our general tests AppriseURLTester(tests=apprise_url_tests).run_all() -@mock.patch('requests.get') -@mock.patch('requests.post') -def test_plugin_twitter_general(mock_post, mock_get): +def test_plugin_twitter_general(mocker): """ NotifyTwitter() General Tests - """ + + mock_get = mocker.patch("requests.get") + mock_post = mocker.patch("requests.post") + ckey = 'ckey' csecret = 'csecret' akey = 'akey' asecret = 'asecret' - screen_name = 'apprise' response_obj = [{ - 'screen_name': screen_name, + 'screen_name': TWITTER_SCREEN_NAME, 'id': 9876, }] # Epoch time: epoch = datetime.utcfromtimestamp(0) - request = mock.Mock() - request.content = dumps(response_obj) + request = Mock() + request.content = json.dumps(response_obj) request.status_code = requests.codes.ok request.headers = { 'x-rate-limit-reset': (datetime.utcnow() - epoch).total_seconds(), @@ -249,7 +354,7 @@ def test_plugin_twitter_general(mock_post, mock_get): csecret=csecret, akey=akey, asecret=asecret, - targets=screen_name) + targets=TWITTER_SCREEN_NAME) assert isinstance(obj, NotifyTwitter) is True assert isinstance(obj.url(), str) is True @@ -310,7 +415,7 @@ def test_plugin_twitter_general(mock_post, mock_get): # Alter pending targets obj.targets.append('usera') - request.content = dumps(response_obj) + request.content = json.dumps(response_obj) response_obj = [{ 'screen_name': 'usera', 'id': 1234, @@ -318,7 +423,7 @@ def test_plugin_twitter_general(mock_post, mock_get): assert obj.send(body="test") is True - # Flush our cache forcing it's re-creating + # Flush our cache forcing it is re-creating NotifyTwitter._user_cache = {} assert obj.send(body="test") is True @@ -335,9 +440,9 @@ def test_plugin_twitter_general(mock_post, mock_get): results = NotifyTwitter.parse_url( 'twitter://{}/{}/{}/{}?to={}'.format( - ckey, csecret, akey, asecret, screen_name)) + ckey, csecret, akey, asecret, TWITTER_SCREEN_NAME)) assert isinstance(results, dict) is True - assert screen_name in results['targets'] + assert TWITTER_SCREEN_NAME in results['targets'] # cause a json parsing issue now response_obj = None @@ -352,10 +457,10 @@ def test_plugin_twitter_general(mock_post, mock_get): NotifyTwitter._user_cache = {} response_obj = { - 'screen_name': screen_name, + 'screen_name': TWITTER_SCREEN_NAME, 'id': 9876, } - request.content = dumps(response_obj) + request.content = json.dumps(response_obj) obj = NotifyTwitter( ckey=ckey, @@ -379,7 +484,6 @@ def test_plugin_twitter_general(mock_post, mock_get): def test_plugin_twitter_edge_cases(): """ NotifyTwitter() Edge Cases - """ with pytest.raises(TypeError): @@ -418,171 +522,188 @@ def test_plugin_twitter_edge_cases(): targets='%G@rB@g3') -@pytest.mark.skipif( - hasattr(sys, "pypy_version_info") and sys.version_info < (3, 7), - reason="Does not work or is flaky on PyPy 3.6") -@mock.patch('requests.post') -@mock.patch('requests.get') -def test_plugin_twitter_dm_attachments(mock_get, mock_post): +def test_plugin_twitter_dm_caching( + mocker, twitter_url, + good_message_response, good_media_response): """ - NotifyTwitter() DM Attachment Checks - + Verify that the `NotifyTwitter.{_user_cache,_whoami_cache}` caches + work as intended. """ - ckey = 'ckey' - csecret = 'csecret' - akey = 'akey' - asecret = 'asecret' - screen_name = 'apprise' - good_dm_response_obj = { - 'screen_name': screen_name, - 'id': 9876, - } + # This is the request to `account/verify_credentials.json`. + # Explicitly mock it here so the calls to it can be evaluated. + mock_get = mocker.patch("requests.get") + mock_get.return_value = good_message_response - # Epoch time: - epoch = datetime.utcfromtimestamp(0) + # This test case submits two notifications, so make sure to provide two + # mocked responses. + mock_post = mocker.patch("requests.post") + mock_post.side_effect = [good_message_response, good_message_response] - # Prepare a good DM response - good_dm_response = mock.Mock() - good_dm_response.content = dumps(good_dm_response_obj) - good_dm_response.status_code = requests.codes.ok - good_dm_response.headers = { - 'x-rate-limit-reset': (datetime.utcnow() - epoch).total_seconds(), - 'x-rate-limit-remaining': 1, - } + # Make sure to start with empty caches. + if hasattr(NotifyTwitter, "_user_cache"): + NotifyTwitter._user_cache = {} + if hasattr(NotifyTwitter, "_whoami_cache"): + NotifyTwitter._whoami_cache = {} - # Prepare bad response - bad_response = mock.Mock() - bad_response.content = dumps({}) - bad_response.status_code = requests.codes.internal_server_error + # Create application objects. + obj = Apprise.instantiate(twitter_url) - # Prepare a good media response - good_media_response = mock.Mock() - good_media_response.content = dumps({ - "media_id": 710511363345354753, - "media_id_string": "710511363345354753", - "media_key": "3_710511363345354753", - "size": 11065, - "expires_after_secs": 86400, - "image": { - "image_type": "image/jpeg", - "w": 800, - "h": 320 - } - }) - good_media_response.status_code = requests.codes.ok + # Send the first notification. + assert obj.notify( + body='body', title='title', notify_type=NotifyType.INFO) is True - # Prepare a bad media response - bad_media_response = mock.Mock() - bad_media_response.content = dumps({ - "errors": [ - { - "code": 93, - "message": "This application is not allowed to access or " - "delete your direct messages.", - }]}) - bad_media_response.status_code = requests.codes.internal_server_error + # Test call counts. + assert mock_get.call_count == 1 + assert mock_get.call_args_list[0][0][0] == \ + 'https://api.twitter.com/1.1/account/verify_credentials.json' + + assert mock_post.call_count == 1 + assert mock_post.call_args_list[0][0][0] == \ + 'https://api.twitter.com/1.1/direct_messages/events/new.json' + + # Reset the mocks to start counting calls from scratch. + mock_get.reset_mock() + mock_post.reset_mock() - mock_post.side_effect = [good_media_response, good_dm_response] - mock_get.return_value = good_dm_response + # Send another notification. + assert obj.notify( + body='body', title='title', notify_type=NotifyType.INFO) is True - twitter_url = 'twitter://{}/{}/{}/{}'.format(ckey, csecret, akey, asecret) + # Calls to `verify_credentials.json` will get cached by `NotifyTwitter`. + # So, the `GET` request to `verify_credentials.json` should only have been + # issued once. + assert mock_get.call_count == 0 + assert mock_post.call_count == 1 - # attach our content - attach = AppriseAttachment(os.path.join(TEST_VAR_DIR, 'apprise-test.gif')) - # instantiate our object +def test_plugin_twitter_dm_attachments_basic( + mocker, twitter_url, + good_message_response, good_media_response): + """ + NotifyTwitter() DM Attachment Checks - Basic + """ + + mock_get = mocker.patch("requests.get") + mock_post = mocker.patch("requests.post") + + # Epoch time: + epoch = datetime.utcfromtimestamp(0) + mock_get.return_value = good_message_response + mock_post.return_value.headers = { + 'x-rate-limit-reset': (datetime.utcnow() - epoch).total_seconds(), + 'x-rate-limit-remaining': 1, + } + + # The first response is for uploading the attachment, + # the second one for posting the actual message. + mock_post.side_effect = [good_media_response, good_message_response] + + # Create application objects. obj = Apprise.instantiate(twitter_url) + attach = AppriseAttachment(os.path.join(TEST_VAR_DIR, 'apprise-test.gif')) - # Send our notification + # Send our notification. assert obj.notify( body='body', title='title', notify_type=NotifyType.INFO, attach=attach) is True - # Test our call count + # Test call counts. assert mock_get.call_count == 1 assert mock_get.call_args_list[0][0][0] == \ 'https://api.twitter.com/1.1/account/verify_credentials.json' + assert mock_post.call_count == 2 assert mock_post.call_args_list[0][0][0] == \ 'https://upload.twitter.com/1.1/media/upload.json' assert mock_post.call_args_list[1][0][0] == \ 'https://api.twitter.com/1.1/direct_messages/events/new.json' - mock_get.reset_mock() - mock_post.reset_mock() - # Test case where upload fails - mock_get.return_value = good_dm_response - mock_post.side_effect = [bad_media_response, good_dm_response] +def test_plugin_twitter_dm_attachments_message_fails( + mocker, twitter_url, + good_media_response, bad_message_response): + """ + Test case with a bad media response. + """ - # instantiate our object + mock_post = mocker.patch("requests.post") + mock_post.side_effect = [good_media_response, bad_message_response] + + # Create application objects. obj = Apprise.instantiate(twitter_url) + attach = AppriseAttachment(os.path.join(TEST_VAR_DIR, 'apprise-test.gif')) - # Send our notification; it will fail because of the media response + # Send our notification; it will fail because of the message response. assert obj.notify( body='body', title='title', notify_type=NotifyType.INFO, attach=attach) is False - # Test our call count - assert mock_get.call_count == 0 - # No get request as cached response is used - assert mock_post.call_count == 1 + assert mock_post.call_count == 2 assert mock_post.call_args_list[0][0][0] == \ 'https://upload.twitter.com/1.1/media/upload.json' + assert mock_post.call_args_list[1][0][0] == \ + 'https://api.twitter.com/1.1/direct_messages/events/new.json' - mock_get.reset_mock() - mock_post.reset_mock() - # Test case where upload fails - mock_get.return_value = good_dm_response - mock_post.side_effect = [good_media_response, bad_response] +def test_plugin_twitter_dm_attachments_upload_fails( + mocker, twitter_url, + good_message_response, bad_media_response): + """ + Test case where upload fails. + """ - # instantiate our object + mock_post = mocker.patch("requests.post") + mock_post.side_effect = [bad_media_response, good_message_response] + + # Create application objects. obj = Apprise.instantiate(twitter_url) + attach = AppriseAttachment(os.path.join(TEST_VAR_DIR, 'apprise-test.gif')) - # Send our notification; it will fail because of the media response + # Send our notification; it will fail because of the media response. assert obj.notify( body='body', title='title', notify_type=NotifyType.INFO, attach=attach) is False - assert mock_get.call_count == 1 - assert mock_get.call_args_list[0][0][0] == \ - 'https://api.twitter.com/1.1/account/verify_credentials.json' - - # No get request as cached response is used - assert mock_post.call_count == 2 + # Test call counts. + assert mock_post.call_count == 1 assert mock_post.call_args_list[0][0][0] == \ 'https://upload.twitter.com/1.1/media/upload.json' - assert mock_post.call_args_list[1][0][0] == \ - 'https://api.twitter.com/1.1/direct_messages/events/new.json' - mock_get.reset_mock() - mock_post.reset_mock() - mock_post.side_effect = [good_media_response, good_dm_response] - mock_get.return_value = good_dm_response +def test_plugin_twitter_dm_attachments_invalid_attachment( + mocker, twitter_url, good_message_response): + """ + Test case with an invalid attachment. + """ + + mock_post: Mock = mocker.patch("requests.post") + mock_post.side_effect = [good_media_response, good_message_response] + + # Create application objects. + # An invalid attachment will cause a failure. + obj = Apprise.instantiate(twitter_url) + attach = AppriseAttachment( + os.path.join(TEST_VAR_DIR, '/invalid/path/to/an/invalid/file.jpg')) - # An invalid attachment will cause a failure - path = os.path.join(TEST_VAR_DIR, '/invalid/path/to/an/invalid/file.jpg') assert obj.notify( body='body', title='title', notify_type=NotifyType.INFO, - attach=path) is False + attach=attach) is False - # No get request as cached response is used - assert mock_get.call_count == 0 + # Verify no post requests have been issued, because attachment is not good. + assert mock_post.mock_calls == [] - # No post request as attachment is no good anyway - assert mock_post.call_count == 0 - mock_get.reset_mock() - mock_post.reset_mock() +def test_plugin_twitter_dm_attachments_multiple( + mocker, twitter_url, + good_message_response, good_media_response): + + mock_post = mocker.patch("requests.post") mock_post.side_effect = [ good_media_response, good_media_response, good_media_response, - good_media_response, good_dm_response, good_dm_response, - good_dm_response, good_dm_response] - mock_get.return_value = good_dm_response + good_media_response, good_message_response, good_message_response, + good_message_response, good_message_response] # 4 images are produced attach = [ @@ -594,12 +715,13 @@ def test_plugin_twitter_dm_attachments(mock_get, mock_post): os.path.join(TEST_VAR_DIR, 'apprise-test.mp4'), ] + # Create application objects. + obj = Apprise.instantiate(twitter_url) + assert obj.notify( body='body', title='title', notify_type=NotifyType.INFO, attach=attach) is True - assert mock_get.call_count == 0 - # No get request as cached response is used assert mock_post.call_count == 8 assert mock_post.call_args_list[0][0][0] == \ 'https://upload.twitter.com/1.1/media/upload.json' @@ -618,12 +740,14 @@ def test_plugin_twitter_dm_attachments(mock_get, mock_post): assert mock_post.call_args_list[7][0][0] == \ 'https://api.twitter.com/1.1/direct_messages/events/new.json' - mock_get.reset_mock() - mock_post.reset_mock() - # We have an OSError thrown in the middle of our preparation +def test_plugin_twitter_dm_attachments_multiple_oserror( + mocker, twitter_url, + good_message_response, good_media_response): + + # Inject an `OSError` into the middle of the operation. + mock_post = mocker.patch("requests.post") mock_post.side_effect = [good_media_response, OSError()] - mock_get.return_value = good_dm_response # 2 images are produced attach = [ @@ -633,13 +757,14 @@ def test_plugin_twitter_dm_attachments(mock_get, mock_post): os.path.join(TEST_VAR_DIR, 'apprise-test.mp4'), ] + # Create application objects. + obj = Apprise.instantiate(twitter_url) + # We'll fail to send this time assert obj.notify( body='body', title='title', notify_type=NotifyType.INFO, attach=attach) is False - assert mock_get.call_count == 0 - # No get request as cached response is used assert mock_post.call_count == 2 assert mock_post.call_args_list[0][0][0] == \ 'https://upload.twitter.com/1.1/media/upload.json' @@ -647,217 +772,184 @@ def test_plugin_twitter_dm_attachments(mock_get, mock_post): 'https://upload.twitter.com/1.1/media/upload.json' -@mock.patch('requests.post') -@mock.patch('requests.get') -def test_plugin_twitter_tweet_attachments(mock_get, mock_post): +@patch('requests.post') +def test_plugin_twitter_tweet_attachments_basic( + mock_post, twitter_url, good_message_response, good_media_response): """ - NotifyTwitter() Tweet Attachment Checks - + NotifyTwitter() Tweet Attachment Checks - Basic """ - ckey = 'ckey' - csecret = 'csecret' - akey = 'akey' - asecret = 'asecret' - screen_name = 'apprise' - - good_tweet_response_obj = { - 'screen_name': screen_name, - 'id': 9876, - } - # Prepare a good DM response - good_tweet_response = mock.Mock() - good_tweet_response.content = dumps(good_tweet_response_obj) - good_tweet_response.status_code = requests.codes.ok + mock_post.side_effect = [good_media_response, good_message_response] - # Prepare bad response - bad_response = mock.Mock() - bad_response.content = dumps({}) - bad_response.status_code = requests.codes.internal_server_error - - # Prepare a good media response - good_media_response = mock.Mock() - good_media_response.content = dumps({ - "media_id": 710511363345354753, - "media_id_string": "710511363345354753", - "media_key": "3_710511363345354753", - "size": 11065, - "expires_after_secs": 86400, - "image": { - "image_type": "image/jpeg", - "w": 800, - "h": 320 - } - }) - good_media_response.status_code = requests.codes.ok - - # Prepare a bad media response - bad_media_response = mock.Mock() - bad_media_response.content = dumps({ - "errors": [ - { - "code": 93, - "message": "This application is not allowed to access or " - "delete your direct messages.", - }]}) - bad_media_response.status_code = requests.codes.internal_server_error - - mock_post.side_effect = [good_media_response, good_tweet_response] - mock_get.return_value = good_tweet_response - - twitter_url = 'twitter://{}/{}/{}/{}?mode=tweet'.format( - ckey, csecret, akey, asecret) - - # attach our content - attach = AppriseAttachment(os.path.join(TEST_VAR_DIR, 'apprise-test.gif')) - - # instantiate our object + # Create application objects. + twitter_url += '?mode=tweet' obj = Apprise.instantiate(twitter_url) + attach = AppriseAttachment(os.path.join(TEST_VAR_DIR, 'apprise-test.gif')) # Send our notification assert obj.notify( body='body', title='title', notify_type=NotifyType.INFO, attach=attach) is True - # Test our call count - assert mock_get.call_count == 0 + # Verify API calls. assert mock_post.call_count == 2 assert mock_post.call_args_list[0][0][0] == \ 'https://upload.twitter.com/1.1/media/upload.json' assert mock_post.call_args_list[1][0][0] == \ 'https://api.twitter.com/1.1/statuses/update.json' - # Update our good response to have more details in it - good_tweet_response_obj = { - 'screen_name': screen_name, + +@patch('requests.post') +def test_plugin_twitter_tweet_attachments_more_logging( + mock_post, twitter_url, good_media_response): + """ + NotifyTwitter() Tweet Attachment Checks - More logging + + TODO: The "more logging" aspect is not verified yet? + """ + + good_tweet_response = good_response({ + 'screen_name': TWITTER_SCREEN_NAME, 'id': 9876, # needed for additional logging 'id_str': '12345', 'user': { - 'screen_name': screen_name, + 'screen_name': TWITTER_SCREEN_NAME, } - } - - good_tweet_response.content = dumps(good_tweet_response_obj) - - mock_get.reset_mock() - mock_post.reset_mock() + }) mock_post.side_effect = [good_media_response, good_tweet_response] - mock_get.return_value = good_tweet_response - # instantiate our object + # Create application objects. + twitter_url += '?mode=tweet' obj = Apprise.instantiate(twitter_url) + attach = AppriseAttachment(os.path.join(TEST_VAR_DIR, 'apprise-test.gif')) - # Send our notification (again); this time there willb e more tweet logging + # Send our notification (again); this time there will be more logging + # TODO: The "more logging" aspect is not verified yet? assert obj.notify( body='body', title='title', notify_type=NotifyType.INFO, attach=attach) is True - # Test our call count - assert mock_get.call_count == 0 + # Verify API calls. assert mock_post.call_count == 2 assert mock_post.call_args_list[0][0][0] == \ 'https://upload.twitter.com/1.1/media/upload.json' assert mock_post.call_args_list[1][0][0] == \ 'https://api.twitter.com/1.1/statuses/update.json' - mock_get.reset_mock() - mock_post.reset_mock() - mock_post.side_effect = [good_media_response, bad_media_response] +@patch('requests.post') +def test_plugin_twitter_tweet_attachments_bad_message_response( + mock_post, twitter_url, good_media_response, bad_message_response): - # instantiate our object + mock_post.side_effect = [good_media_response, bad_message_response] + + # Create application objects. + twitter_url += '?mode=tweet' obj = Apprise.instantiate(twitter_url) + attach = AppriseAttachment(os.path.join(TEST_VAR_DIR, 'apprise-test.gif')) - # Our notification will fail now since our tweet will error out + # Our notification will fail now since our tweet will error out. assert obj.notify( body='body', title='title', notify_type=NotifyType.INFO, attach=attach) is False - # Test our call count - assert mock_get.call_count == 0 + # Verify API calls. assert mock_post.call_count == 2 assert mock_post.call_args_list[0][0][0] == \ 'https://upload.twitter.com/1.1/media/upload.json' assert mock_post.call_args_list[1][0][0] == \ 'https://api.twitter.com/1.1/statuses/update.json' - mock_get.reset_mock() - mock_post.reset_mock() - bad_media_response.content = '' +@patch('requests.post') +def test_plugin_twitter_tweet_attachments_bad_message_response_unparseable( + mock_post, twitter_url, good_media_response): - mock_post.side_effect = [good_media_response, bad_media_response] + bad_message_response = bad_response("") + mock_post.side_effect = [good_media_response, bad_message_response] - # instantiate our object + # Create application objects. + twitter_url += '?mode=tweet' obj = Apprise.instantiate(twitter_url) + attach = AppriseAttachment(os.path.join(TEST_VAR_DIR, 'apprise-test.gif')) - # Our notification will fail now since our tweet will error out - # This is the same test as above, except our error response isn't parseable + # The notification will fail now since the tweet will error out. + # This is the same test as above, except that the error response is not + # parseable. assert obj.notify( body='body', title='title', notify_type=NotifyType.INFO, attach=attach) is False - # Test our call count - assert mock_get.call_count == 0 + # Verify API calls. assert mock_post.call_count == 2 assert mock_post.call_args_list[0][0][0] == \ 'https://upload.twitter.com/1.1/media/upload.json' assert mock_post.call_args_list[1][0][0] == \ 'https://api.twitter.com/1.1/statuses/update.json' - mock_get.reset_mock() - mock_post.reset_mock() - # Test case where upload fails - mock_get.return_value = good_tweet_response - mock_post.side_effect = [good_media_response, bad_response] +@patch('requests.post') +def test_plugin_twitter_tweet_attachments_upload_fails( + mock_post, twitter_url, good_media_response): - # instantiate our object + # Prepare a bad tweet response. + bad_tweet_response = bad_response({}) + + # Test case where upload fails. + mock_post.side_effect = [good_media_response, bad_tweet_response] + + # Create application objects. + twitter_url += '?mode=tweet' obj = Apprise.instantiate(twitter_url) + attach = AppriseAttachment(os.path.join(TEST_VAR_DIR, 'apprise-test.gif')) - # Send our notification; it will fail because of the media response + # Send our notification; it will fail because of the message response. assert obj.notify( body='body', title='title', notify_type=NotifyType.INFO, attach=attach) is False - assert mock_get.call_count == 0 + # Verify API calls. assert mock_post.call_count == 2 assert mock_post.call_args_list[0][0][0] == \ 'https://upload.twitter.com/1.1/media/upload.json' assert mock_post.call_args_list[1][0][0] == \ 'https://api.twitter.com/1.1/statuses/update.json' - mock_get.reset_mock() - mock_post.reset_mock() - mock_post.side_effect = [good_media_response, good_tweet_response] - mock_get.return_value = good_tweet_response +@patch('requests.post') +def test_plugin_twitter_tweet_attachments_invalid_attachment( + mock_post, twitter_url, good_message_response, good_media_response): + + mock_post.side_effect = [good_media_response, good_message_response] + + # Create application objects. + twitter_url += '?mode=tweet' + obj = Apprise.instantiate(twitter_url) + attach = AppriseAttachment( + os.path.join(TEST_VAR_DIR, '/invalid/path/to/an/invalid/file.jpg')) - # An invalid attachment will cause a failure - path = os.path.join(TEST_VAR_DIR, '/invalid/path/to/an/invalid/file.jpg') + # An invalid attachment will cause a failure. assert obj.notify( body='body', title='title', notify_type=NotifyType.INFO, - attach=path) is False - - # No get request as cached response is used - assert mock_get.call_count == 0 + attach=attach) is False - # No post request as attachment is no good anyway + # No post request as attachment is not good. assert mock_post.call_count == 0 - mock_get.reset_mock() - mock_post.reset_mock() + +@patch('requests.post') +def test_plugin_twitter_tweet_attachments_multiple_batch( + mock_post, twitter_url, good_message_response, good_media_response): mock_post.side_effect = [ good_media_response, good_media_response, good_media_response, - good_media_response, good_tweet_response, good_tweet_response, - good_tweet_response, good_tweet_response] - mock_get.return_value = good_tweet_response + good_media_response, good_message_response, good_message_response, + good_message_response, good_message_response] - # instantiate our object (without a batch mode) - obj = Apprise.instantiate(twitter_url + "&batch=no") + # instantiate our object + obj = Apprise.instantiate(twitter_url + "?mode=tweet") # 4 images are produced attach = [ @@ -873,9 +965,7 @@ def test_plugin_twitter_tweet_attachments(mock_get, mock_post): body='body', title='title', notify_type=NotifyType.INFO, attach=attach) is True - assert mock_get.call_count == 0 - # No get request as cached response is used - assert mock_post.call_count == 8 + assert mock_post.call_count == 7 assert mock_post.call_args_list[0][0][0] == \ 'https://upload.twitter.com/1.1/media/upload.json' assert mock_post.call_args_list[1][0][0] == \ @@ -888,22 +978,22 @@ def test_plugin_twitter_tweet_attachments(mock_get, mock_post): 'https://api.twitter.com/1.1/statuses/update.json' assert mock_post.call_args_list[5][0][0] == \ 'https://api.twitter.com/1.1/statuses/update.json' + # The 2 images are grouped together (batch mode) assert mock_post.call_args_list[6][0][0] == \ 'https://api.twitter.com/1.1/statuses/update.json' - assert mock_post.call_args_list[7][0][0] == \ - 'https://api.twitter.com/1.1/statuses/update.json' - mock_get.reset_mock() - mock_post.reset_mock() + +@patch('requests.post') +def test_plugin_twitter_tweet_attachments_multiple_nobatch( + mock_post, twitter_url, good_message_response, good_media_response): mock_post.side_effect = [ good_media_response, good_media_response, good_media_response, - good_media_response, good_tweet_response, good_tweet_response, - good_tweet_response, good_tweet_response] - mock_get.return_value = good_tweet_response + good_media_response, good_message_response, good_message_response, + good_message_response, good_message_response] - # instantiate our object - obj = Apprise.instantiate(twitter_url) + # instantiate our object (without a batch mode) + obj = Apprise.instantiate(twitter_url + "?mode=tweet&batch=no") # 4 images are produced attach = [ @@ -919,9 +1009,7 @@ def test_plugin_twitter_tweet_attachments(mock_get, mock_post): body='body', title='title', notify_type=NotifyType.INFO, attach=attach) is True - assert mock_get.call_count == 0 - # No get request as cached response is used - assert mock_post.call_count == 7 + assert mock_post.call_count == 8 assert mock_post.call_args_list[0][0][0] == \ 'https://upload.twitter.com/1.1/media/upload.json' assert mock_post.call_args_list[1][0][0] == \ @@ -934,16 +1022,18 @@ def test_plugin_twitter_tweet_attachments(mock_get, mock_post): 'https://api.twitter.com/1.1/statuses/update.json' assert mock_post.call_args_list[5][0][0] == \ 'https://api.twitter.com/1.1/statuses/update.json' - # The 2 images are grouped together (batch mode) assert mock_post.call_args_list[6][0][0] == \ 'https://api.twitter.com/1.1/statuses/update.json' + assert mock_post.call_args_list[7][0][0] == \ + 'https://api.twitter.com/1.1/statuses/update.json' - mock_get.reset_mock() - mock_post.reset_mock() + +@patch('requests.post') +def test_plugin_twitter_tweet_attachments_multiple_oserror( + mock_post, twitter_url, good_media_response): # We have an OSError thrown in the middle of our preparation mock_post.side_effect = [good_media_response, OSError()] - mock_get.return_value = good_tweet_response # 2 images are produced attach = [ @@ -954,12 +1044,11 @@ def test_plugin_twitter_tweet_attachments(mock_get, mock_post): ] # We'll fail to send this time + obj = Apprise.instantiate(twitter_url + "?mode=tweet") assert obj.notify( body='body', title='title', notify_type=NotifyType.INFO, attach=attach) is False - assert mock_get.call_count == 0 - # No get request as cached response is used assert mock_post.call_count == 2 assert mock_post.call_args_list[0][0][0] == \ 'https://upload.twitter.com/1.1/media/upload.json'