mirror of https://github.com/caronc/apprise
886 lines
26 KiB
Python
886 lines
26 KiB
Python
# BSD 2-Clause License
|
|
#
|
|
# Apprise - Push Notification Library.
|
|
# Copyright (c) 2025, 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.
|
|
|
|
from datetime import datetime, timezone
|
|
from json import dumps, loads
|
|
|
|
# Disable logging for a cleaner testing output
|
|
import logging
|
|
import os
|
|
from unittest import mock
|
|
|
|
from helpers import AppriseURLTester
|
|
import requests
|
|
|
|
from apprise import Apprise, AppriseAttachment, NotifyType
|
|
from apprise.plugins.mastodon import NotifyMastodon
|
|
|
|
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 i/o exceptions with 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(
|
|
f"mastodon://{token}@{host}/@user?visbility=direct"
|
|
)
|
|
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 = f"mastodon://{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 = f"mastodon://{akey}@{host}?visibility=direct"
|
|
|
|
# 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 = (
|
|
f"mastodons://{akey}@{host}?visibility=direct&sensitive=yes&key=abcd"
|
|
)
|
|
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 = f"mastodons://{akey}@{host}?batch=no"
|
|
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 = (
|
|
f"mastodons://{akey}@{host}?visibility=public&spoiler=uhoh"
|
|
)
|
|
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 = f"mastodons://{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 = f"mastodons://{akey}@{host}?visibility=direct"
|
|
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 = f"mastodons://{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"
|
|
)
|