Browse Source

Tests: Modularize some test case modules (#728)

pull/776/head
Andreas Motl 2 years ago committed by GitHub
parent
commit
c9261d8459
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
  1. 349
      test/test_plugin_fcm.py
  2. 276
      test/test_plugin_msteams.py
  3. 643
      test/test_plugin_twitter.py

349
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
"""

276
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))

643
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'

Loading…
Cancel
Save