extern/apprise
1
0
Fork 0
mirror of https://github.com/caronc/apprise.git synced 2024-10-04 17:22:18 +02:00
Code Issues Packages Projects Releases Wiki Activity
master
apprise/test/test_plugin_mastodon.py

747 lines
24 KiB
Python
Raw Permalink Blame History

# -*- coding: utf-8 -*-
# BSD 2-Clause License
#
# Apprise - Push Notification Library.
# Copyright (c) 2024, Chris Caron <lead2gold@gmail.com>
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
# 1. Redistributions of source code must retain the above copyright notice,
# this list of conditions and the following disclaimer.
#
# 2. Redistributions in binary form must reproduce the above copyright notice,
# this list of conditions and the following disclaimer in the documentation
# and/or other materials provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
import os
from unittest import mock
import requests
from json import dumps, loads
from datetime import datetime
from datetime import timezone
from apprise import Apprise
from apprise import NotifyType
from apprise import AppriseAttachment
from apprise.plugins.mastodon import NotifyMastodon
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')
# Our Testing URLs
apprise_url_tests = (
##################################
# NotifyMastodon
##################################
('mastodon://', {
# Missing Everything :)
'instance': None,
}),
('mastodon://:@/', {
'instance': None,
}),
('mastodon://hostname', {
# Missing Access Token
'instance': TypeError,
}),
('toot://access_token@hostname', {
# We're good; it's a simple notification
'instance': NotifyMastodon,
}),
('toots://access_token@hostname', {
# We're good; it's another simple notification
'instance': NotifyMastodon,
# Our expected url(privacy=True) startswith() response:
'privacy_url': 'mastodons://****@hostname/'
}),
('mastodon://access_token@hostname/@user/@user2', {
# We're good; it's another simple notification
'instance': NotifyMastodon,
# Our expected url(privacy=True) startswith() response:
'privacy_url': 'mastodon://****@hostname/@user/@user2'
}),
('mastodon://hostname/@user/@user2?token=abcd123', {
# Our access token can be provided as a token= variable
'instance': NotifyMastodon,
# Our expected url(privacy=True) startswith() response:
'privacy_url': 'mastodon://****@hostname/@user/@user2'
}),
('mastodon://access_token@hostname?to=@user, @user2', {
# We're good; it's another simple notification
'instance': NotifyMastodon,
# Our expected url(privacy=True) startswith() response:
'privacy_url': 'mastodon://****@hostname/@user/@user2'
}),
('mastodon://access_token@hostname/?cache=no', {
# disable cache as a test
'instance': NotifyMastodon,
}),
('mastodon://access_token@hostname/?spoiler=spoiler%20text', {
# a public post
'instance': NotifyMastodon,
}),
('mastodon://access_token@hostname/?language=en', {
# over-ride our language
'instance': NotifyMastodon,
}),
('mastodons://access_token@hostname:8443', {
# A custom port specified
'instance': NotifyMastodon,
}),
('mastodon://access_token@hostname/?key=My%20Idempotency%20Key', {
# Prevent duplicate submissions of the same status. Idempotency
# keys are stored for up to 1 hour, and can be any arbitrary string.
# Consider using a hash or UUID generated client-side.
'instance': NotifyMastodon,
}),
('mastodon://access_token@hostname/-/%/', {
# Invalid users specified
'instance': TypeError,
}),
('mastodon://access_token@hostname?visibility=invalid', {
# An invalid visibility
'instance': TypeError,
}),
('mastodon://access_token@hostname?visibility=direct', {
# A direct message
'instance': NotifyMastodon,
# Expected notify() response False (because we won't
# get the response we were expecting from the upstream
# server
'notify_response': False,
}),
('mastodon://access_token@hostname?visibility=direct', {
# A direct message
'instance': NotifyMastodon,
# Provide a response that allows us to look our content up
'requests_response_text': {
'id': '12345',
'username': 'test',
},
}),
('toots://access_token@hostname', {
'instance': NotifyMastodon,
# Throws a series of connection and transfer exceptions when this flag
# is set and tests that we gracfully handle them
'test_requests_exceptions': True,
}),
)
def test_plugin_mastodon_urls():
"""
NotifyMastodon() Apprise URLs
"""
# Run our general tests
AppriseURLTester(tests=apprise_url_tests).run_all()
@mock.patch('requests.get')
@mock.patch('requests.post')
def test_plugin_mastodon_general(mock_post, mock_get):
"""
NotifyMastodon() General Tests
"""
token = 'access_key'
host = 'nuxref.com'
response_obj = {
'username': 'caronc',
'id': 1234,
}
# Epoch time:
epoch = datetime.fromtimestamp(0, timezone.utc)
request = mock.Mock()
request.content = dumps(response_obj)
request.status_code = requests.codes.ok
request.headers = {
'X-RateLimit-Limit': (
datetime.now(timezone.utc) - epoch).total_seconds(),
'X-RateLimit-Remaining': 1,
}
# Prepare Mock
mock_get.return_value = request
mock_post.return_value = request
# Instantiate our object
obj = NotifyMastodon(token=token, host=host)
assert isinstance(obj, NotifyMastodon)
assert isinstance(obj.url(), str)
# apprise room was found
assert obj.send(body="test") is True
# Change our status code and try again
request.status_code = 403
assert obj.send(body="test") is False
assert obj.ratelimit_remaining == 1
# Return the status
request.status_code = requests.codes.ok
# Force a reset
request.headers['X-RateLimit-Remaining'] = 0
# behind the scenes, it should cause us to update our rate limit
assert obj.send(body="test") is True
assert obj.ratelimit_remaining == 0
# This should cause us to block
request.headers['X-RateLimit-Remaining'] = 10
assert obj.send(body="test") is True
assert obj.ratelimit_remaining == 10
# Handle cases where we simply couldn't get this field
del request.headers['X-RateLimit-Remaining']
assert obj.send(body="test") is True
# It remains set to the last value
assert obj.ratelimit_remaining == 10
# Reset our variable back to 1
request.headers['X-RateLimit-Remaining'] = 1
# Handle cases where our epoch time is wrong
del request.headers['X-RateLimit-Limit']
assert obj.send(body="test") is True
# Return our object, but place it in the future forcing us to block
request.headers['X-RateLimit-Limit'] = \
(datetime.now(timezone.utc) - epoch).total_seconds() + 1
request.headers['X-RateLimit-Remaining'] = 0
obj.ratelimit_remaining = 0
assert obj.send(body="test") is True
# Return our object, but place it in the future forcing us to block
request.headers['X-RateLimit-Limit'] = \
(datetime.now(timezone.utc) - epoch).total_seconds() - 1
request.headers['X-RateLimit-Remaining'] = 0
obj.ratelimit_remaining = 0
assert obj.send(body="test") is True
# Return our limits to always work
request.headers['X-RateLimit-Limit'] = \
(datetime.now(timezone.utc) - epoch).total_seconds()
request.headers['X-RateLimit-Remaining'] = 1
obj.ratelimit_remaining = 1
# Alter pending targets
obj.targets.append('usera')
request.content = dumps(response_obj)
response_obj = {
'username': 'usera',
'id': 4321,
}
# Cause content response to be None
request.content = None
assert obj.send(body="test") is True
# Invalid JSON
request.content = '{'
assert obj.send(body="test") is True
# Return it to a parseable string
request.content = '{}'
results = NotifyMastodon.parse_url(
'mastodon://{}@{}/@user?visbility=direct'.format(token, host))
assert isinstance(results, dict)
assert '@user' in results['targets']
# cause a json parsing issue now
response_obj = None
assert obj.send(body="test") is True
response_obj = '{'
assert obj.send(body="test") is True
mock_get.reset_mock()
mock_post.reset_mock()
#
# Test our lazy lookups
#
# Prepare a good status response
request = mock.Mock()
request.content = dumps({'id': '1234', 'username': 'caronc'})
request.status_code = requests.codes.ok
mock_get.return_value = request
mastodon_url = 'mastodons://key@host?visibility=direct'
obj = Apprise.instantiate(mastodon_url)
obj._whoami(lazy=True)
assert mock_get.call_count == 1
assert mock_get.call_args_list[0][0][0] == \
'https://host/api/v1/accounts/verify_credentials'
mock_get.reset_mock()
obj._whoami(lazy=True)
assert mock_get.call_count == 0
mock_get.reset_mock()
obj._whoami(lazy=False)
assert mock_get.call_count == 1
assert mock_get.call_args_list[0][0][0] == \
'https://host/api/v1/accounts/verify_credentials'
@mock.patch('requests.post')
@mock.patch('requests.get')
def test_plugin_mastodon_attachments(mock_get, mock_post):
"""
NotifyMastodon() Toot Attachment Checks
"""
akey = 'access_key'
host = 'nuxref.com'
username = 'caronc'
# Prepare a good status response
good_response_obj = {
'id': '1234',
}
good_response = mock.Mock()
good_response.content = dumps(good_response_obj)
good_response.status_code = requests.codes.ok
# Prepare a good whoami response
good_whoami_response_obj = {
'username': username,
'id': '9876',
}
good_whoami_response = mock.Mock()
good_whoami_response.content = dumps(good_whoami_response_obj)
good_whoami_response.status_code = requests.codes.ok
# 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({
"id": '710511363345354753',
"file_mime": "image/jpeg",
})
good_media_response.status_code = requests.codes.ok
#
# Start testing using fixtures above
#
mock_post.side_effect = [good_media_response, good_response]
mock_get.return_value = good_whoami_response
mastodon_url = 'mastodon://{}@{}'.format(akey, host)
# attach our content
attach = AppriseAttachment(os.path.join(TEST_VAR_DIR, 'apprise-test.gif'))
# instantiate our object
obj = Apprise.instantiate(mastodon_url)
# 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
assert mock_post.call_count == 2
assert mock_post.call_args_list[0][0][0] == \
'http://nuxref.com/api/v1/media'
assert mock_post.call_args_list[1][0][0] == \
'http://nuxref.com/api/v1/statuses'
# Test our media payload
assert 'files' in mock_post.call_args_list[0][1]
assert 'file' in mock_post.call_args_list[0][1]['files']
assert mock_post.call_args_list[0][1]['files']['file'][0] \
== 'apprise-test.gif'
# Test our status payload
payload = loads(mock_post.call_args_list[1][1]['data'])
assert 'status' in payload
assert payload['status'] == 'title\r\nbody'
assert 'sensitive' in payload
assert payload['sensitive'] is False
assert 'media_ids' in payload
assert isinstance(payload['media_ids'], list)
assert len(payload['media_ids']) == 1
assert payload['media_ids'][0] == '710511363345354753'
# Verify we don't set incorrect keys not otherwise specified
assert 'spoiler_text' not in payload
mock_get.reset_mock()
mock_post.reset_mock()
#
# Handle the query again, but this time perform a direct message
# requiring us to look up who we are
#
mock_post.side_effect = [good_media_response, good_response]
mock_get.return_value = good_whoami_response
mastodon_url = 'mastodon://{}@{}?visibility=direct'.format(akey, host)
# attach our content
attach = AppriseAttachment(os.path.join(TEST_VAR_DIR, 'apprise-test.gif'))
# instantiate our object
obj = Apprise.instantiate(mastodon_url)
# 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 == 1
assert mock_post.call_count == 2
assert mock_get.call_args_list[0][0][0] == \
'http://nuxref.com/api/v1/accounts/verify_credentials'
assert mock_post.call_args_list[0][0][0] == \
'http://nuxref.com/api/v1/media'
assert mock_post.call_args_list[1][0][0] == \
'http://nuxref.com/api/v1/statuses'
# Test our status payload
payload = loads(mock_post.call_args_list[1][1]['data'])
assert 'status' in payload
# Our ID was added into the payload
assert payload['status'] == '@caronc title\r\nbody'
assert 'sensitive' in payload
assert payload['sensitive'] is False
assert 'media_ids' in payload
assert isinstance(payload['media_ids'], list)
assert len(payload['media_ids']) == 1
assert payload['media_ids'][0] == '710511363345354753'
mock_get.reset_mock()
mock_post.reset_mock()
# Store 3 attachments
attach = (
AppriseAttachment(os.path.join(TEST_VAR_DIR, 'apprise-test.gif')),
AppriseAttachment(os.path.join(TEST_VAR_DIR, 'apprise-test.png')),
AppriseAttachment(os.path.join(TEST_VAR_DIR, 'apprise-test.jpeg')),
)
# Prepare a good media response
mr1 = mock.Mock()
mr1.content = dumps({
"id": '1',
"file_mime": "image/gif",
})
mr1.status_code = requests.codes.ok
mr2 = mock.Mock()
mr2.content = dumps({
"id": '2',
"file_mime": "image/png",
})
mr2.status_code = requests.codes.ok
mr3 = mock.Mock()
mr3.content = dumps({
"id": '3',
"file_mime": "image/jpeg",
})
mr3.status_code = requests.codes.ok
# Return 3 good uploads and a good status response
mock_post.side_effect = [mr1, mr2, mr3, good_response, good_response]
mock_get.return_value = good_whoami_response
# instantiate our object
mastodon_url = 'mastodons://{}@{}' \
'?visibility=direct&sensitive=yes&key=abcd'.format(akey, host)
obj = Apprise.instantiate(mastodon_url)
# Send ourselves a direct message where our ID was already found
# in the body. This smart detection method will prevent us from
# adding the @caronc to the begining of the same message (since it's a
# direct message)
assert obj.notify(
body='Check this out @caronc', title='Apprise',
notify_type=NotifyType.INFO,
attach=attach) is True
# Test our call count
assert mock_get.call_count == 1
assert mock_post.call_count == 5
assert mock_post.call_args_list[0][0][0] == \
'https://nuxref.com/api/v1/media'
assert mock_post.call_args_list[1][0][0] == \
'https://nuxref.com/api/v1/media'
assert mock_post.call_args_list[2][0][0] == \
'https://nuxref.com/api/v1/media'
# Our status's will batch up and send the last 2 images in one
# and our animated gif in one.
assert mock_post.call_args_list[3][0][0] == \
'https://nuxref.com/api/v1/statuses'
assert mock_post.call_args_list[4][0][0] == \
'https://nuxref.com/api/v1/statuses'
assert mock_get.call_args_list[0][0][0] == \
'https://nuxref.com/api/v1/accounts/verify_credentials'
# Test our status payload
payload = loads(mock_post.call_args_list[3][1]['data'])
assert 'status' in payload
assert payload['status'] == 'Apprise\r\nCheck this out @caronc'
assert 'sensitive' in payload
assert payload['sensitive'] is True
assert 'language' not in payload
assert 'Idempotency-Key' in payload
assert payload['Idempotency-Key'] == 'abcd'
assert 'media_ids' in payload
assert isinstance(payload['media_ids'], list)
assert len(payload['media_ids']) == 1
assert payload['media_ids'][0] == '1'
payload = loads(mock_post.call_args_list[4][1]['data'])
assert 'status' in payload
assert payload['status'] == '02/02'
assert 'sensitive' in payload
assert payload['sensitive'] is False
assert 'language' not in payload
assert 'Idempotency-Key' in payload
assert payload['Idempotency-Key'] == 'abcd-part01'
assert 'media_ids' in payload
assert isinstance(payload['media_ids'], list)
assert len(payload['media_ids']) == 2
assert '2' in payload['media_ids']
assert '3' in payload['media_ids']
# A second call does not cause us to look up our ID as we already know it
mock_get.reset_mock()
mock_post.reset_mock()
mock_post.side_effect = [mr1, mr2, mr3, good_response, good_response]
mock_get.return_value = good_whoami_response
assert obj.notify(
body='Check this out @caronc', title='Apprise',
notify_type=NotifyType.INFO,
attach=attach) is True
# Same number of posts
assert mock_post.call_count == 5
# But no lookup was made
assert mock_get.call_count == 0
mock_get.reset_mock()
mock_post.reset_mock()
# Prepare an attach list
attach = (
os.path.join(TEST_VAR_DIR, 'apprise-test.png'),
os.path.join(TEST_VAR_DIR, 'apprise-test.jpeg'),
)
mock_post.side_effect = [mr2, mr3, good_response, good_response]
mock_get.return_value = good_whoami_response
# instantiate our object (but turn off the batch mode)
mastodon_url = 'mastodons://{}@{}?batch=no'.format(
akey, host)
obj = Apprise.instantiate(mastodon_url)
assert obj.notify(
body='Check this out @caronc', title='Apprise',
notify_type=NotifyType.INFO,
attach=attach) is True
# 2 attachments + 2 different status messages
assert mock_post.call_count == 4
# But no lookup was made
assert mock_get.call_count == 0
assert mock_post.call_args_list[0][0][0] == \
'https://nuxref.com/api/v1/media'
assert mock_post.call_args_list[1][0][0] == \
'https://nuxref.com/api/v1/media'
assert mock_post.call_args_list[2][0][0] == \
'https://nuxref.com/api/v1/statuses'
assert mock_post.call_args_list[3][0][0] == \
'https://nuxref.com/api/v1/statuses'
mock_get.reset_mock()
mock_post.reset_mock()
# Prepare a bad media response
bad_response = mock.Mock()
bad_response.status_code = requests.codes.internal_server_error
bad_responses = (
dumps({"error": "authorized scopes"}),
'',
)
#
# Test our Media failures
#
# Try several bad responses so we can capture the block of code where
# we try to help the end user to remind them what scope they're missing
for response in bad_responses:
mock_post.side_effect = [good_media_response, bad_response]
bad_response.content = response
# instantiate our object
mastodon_url = \
'mastodons://{}@{}?visibility=public&spoiler=uhoh'.format(
akey, host)
obj = Apprise.instantiate(mastodon_url)
# Our notification will fail now since our toot will error out
# This is the same test as above, except our error response isn't
# 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
assert mock_post.call_count == 2
assert mock_post.call_args_list[0][0][0] == \
'https://nuxref.com/api/v1/media'
assert mock_post.call_args_list[1][0][0] == \
'https://nuxref.com/api/v1/media'
mock_get.reset_mock()
mock_post.reset_mock()
#
# Test our Status failures
#
# Try several bad responses so we can capture the block of code where
# we try to help the end user to remind them what scope they're missing
for response in bad_responses:
mock_post.side_effect = [bad_response]
bad_response.content = response
# instantiate our object
mastodon_url = 'mastodons://{}@{}'.format(akey, host)
obj = Apprise.instantiate(mastodon_url)
# Our notification will fail now since our toot will error out
# This is the same test as above, except our error response isn't
# parseable
assert obj.notify(
body='body', title='title', notify_type=NotifyType.INFO) is False
# Test our call count
assert mock_get.call_count == 0
assert mock_post.call_count == 1
assert mock_post.call_args_list[0][0][0] == \
'https://nuxref.com/api/v1/statuses'
mock_get.reset_mock()
mock_post.reset_mock()
#
# Test our whoami failures
#
# Try several bad responses so we can capture the block of code where
# we try to help the end user to remind them what scope they're missing
for response in bad_responses:
mock_get.side_effect = [bad_response]
bad_response.content = response
# instantiate our object
mastodon_url = 'mastodons://{}@{}?visibility=direct'.format(akey, host)
obj = Apprise.instantiate(mastodon_url)
# Our notification will fail now since our toot will error out
# This is the same test as above, except our error response isn't
# parseable
assert obj.notify(
body='body', title='title', notify_type=NotifyType.INFO) is False
# Test our call count
assert mock_get.call_count == 1
assert mock_post.call_count == 0
assert mock_get.call_args_list[0][0][0] == \
'https://nuxref.com/api/v1/accounts/verify_credentials'
mock_get.reset_mock()
mock_post.reset_mock()
mock_post.side_effect = [mr1, mr2, mr3, good_response, good_response]
mock_get.return_value = None
# instantiate our object
mastodon_url = 'mastodons://{}@{}'.format(akey, host)
obj = Apprise.instantiate(mastodon_url)
# 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
# No get requests are made
assert mock_get.call_count == 0
# No post request as attachment is no good anyway
assert mock_post.call_count == 0
mock_get.reset_mock()
mock_post.reset_mock()
# We have an OSError thrown in the middle of our preparation
mock_post.side_effect = [
good_media_response, OSError(), good_media_response]
mock_get.return_value = good_response
# 3 images are produced
attach = [
os.path.join(TEST_VAR_DIR, 'apprise-test.gif'),
# This one is not supported, so it's ignored gracefully
os.path.join(TEST_VAR_DIR, 'apprise-archive.zip'),
# A supported video file
os.path.join(TEST_VAR_DIR, 'apprise-test.mp4'),
]
# 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://nuxref.com/api/v1/media'
assert mock_post.call_args_list[1][0][0] == \
'https://nuxref.com/api/v1/media'
Reference in New Issue View Git Blame Copy Permalink