2017-12-25 20:07:41 +00:00
|
|
|
# -*- coding: utf-8 -*-
|
|
|
|
#
|
2019-02-01 02:05:56 +00:00
|
|
|
# Copyright (C) 2019 Chris Caron <lead2gold@gmail.com>
|
|
|
|
# All rights reserved.
|
2017-12-25 20:07:41 +00:00
|
|
|
#
|
2019-02-01 02:05:56 +00:00
|
|
|
# This code is licensed under the MIT License.
|
2017-12-25 20:07:41 +00:00
|
|
|
#
|
2019-02-01 02:05:56 +00:00
|
|
|
# Permission is hereby granted, free of charge, to any person obtaining a copy
|
|
|
|
# of this software and associated documentation files(the "Software"), to deal
|
|
|
|
# in the Software without restriction, including without limitation the rights
|
|
|
|
# to use, copy, modify, merge, publish, distribute, sublicense, and / or sell
|
|
|
|
# copies of the Software, and to permit persons to whom the Software is
|
|
|
|
# furnished to do so, subject to the following conditions :
|
2017-12-25 20:07:41 +00:00
|
|
|
#
|
2019-02-01 02:05:56 +00:00
|
|
|
# The above copyright notice and this permission notice shall be included in
|
|
|
|
# all copies or substantial portions of the Software.
|
2017-12-25 20:07:41 +00:00
|
|
|
#
|
2019-02-01 02:05:56 +00:00
|
|
|
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
|
|
|
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
|
|
|
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE
|
|
|
|
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
|
|
|
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
|
|
|
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
|
|
|
|
# THE SOFTWARE.
|
2017-12-25 20:07:41 +00:00
|
|
|
|
2022-03-07 22:52:22 +00:00
|
|
|
import os
|
2022-10-18 11:18:50 +00:00
|
|
|
import sys
|
2022-10-08 00:28:36 +00:00
|
|
|
from unittest import mock
|
2022-09-01 00:05:40 +00:00
|
|
|
|
2019-10-09 16:39:31 +00:00
|
|
|
import pytest
|
2019-06-29 22:27:59 +00:00
|
|
|
import requests
|
|
|
|
from json import dumps
|
|
|
|
from datetime import datetime
|
2022-03-07 22:52:22 +00:00
|
|
|
from apprise import Apprise
|
|
|
|
from apprise import NotifyType
|
|
|
|
from apprise import AppriseAttachment
|
2022-10-09 09:28:18 +00:00
|
|
|
from apprise.plugins.NotifyTwitter import NotifyTwitter
|
2021-11-25 20:20:22 +00:00
|
|
|
from helpers import AppriseURLTester
|
2017-12-25 20:07:41 +00:00
|
|
|
|
2019-02-25 06:02:29 +00:00
|
|
|
# Disable logging for a cleaner testing output
|
|
|
|
import logging
|
|
|
|
logging.disable(logging.CRITICAL)
|
|
|
|
|
2022-03-07 22:52:22 +00:00
|
|
|
# Attachment Directory
|
|
|
|
TEST_VAR_DIR = os.path.join(os.path.dirname(__file__), 'var')
|
|
|
|
|
2021-11-25 20:20:22 +00:00
|
|
|
# Our Testing URLs
|
|
|
|
apprise_url_tests = (
|
|
|
|
##################################
|
|
|
|
# NotifyTwitter
|
|
|
|
##################################
|
|
|
|
('twitter://', {
|
|
|
|
# Missing Consumer API Key
|
|
|
|
'instance': TypeError,
|
|
|
|
}),
|
|
|
|
('twitter://:@/', {
|
|
|
|
'instance': TypeError,
|
|
|
|
}),
|
|
|
|
('twitter://consumer_key', {
|
|
|
|
# Missing Keys
|
|
|
|
'instance': TypeError,
|
|
|
|
}),
|
|
|
|
('twitter://consumer_key/consumer_secret/', {
|
|
|
|
# Missing Keys
|
|
|
|
'instance': TypeError,
|
|
|
|
}),
|
|
|
|
('twitter://consumer_key/consumer_secret/access_token/', {
|
|
|
|
# Missing Access Secret
|
|
|
|
'instance': TypeError,
|
|
|
|
}),
|
|
|
|
('twitter://consumer_key/consumer_secret/access_token/access_secret', {
|
|
|
|
# No user mean's we message ourselves
|
2022-10-09 09:28:18 +00:00
|
|
|
'instance': NotifyTwitter,
|
2021-11-25 20:20:22 +00:00
|
|
|
# Expected notify() response False (because we won't be able
|
|
|
|
# to detect our user)
|
|
|
|
'notify_response': False,
|
|
|
|
|
|
|
|
# Our expected url(privacy=True) startswith() response:
|
|
|
|
'privacy_url': 'twitter://c...y/****/a...n/****',
|
|
|
|
}),
|
|
|
|
('twitter://consumer_key/consumer_secret/access_token/access_secret'
|
|
|
|
'?cache=no', {
|
|
|
|
# No user mean's we message ourselves
|
2022-10-09 09:28:18 +00:00
|
|
|
'instance': NotifyTwitter,
|
2021-11-25 20:20:22 +00:00
|
|
|
# However we'll be okay if we return a proper response
|
|
|
|
'requests_response_text': {
|
|
|
|
'id': 12345,
|
2022-03-07 22:52:22 +00:00
|
|
|
'screen_name': 'test',
|
|
|
|
# For attachment handling
|
|
|
|
'media_id': 123,
|
2021-11-25 20:20:22 +00:00
|
|
|
},
|
|
|
|
}),
|
|
|
|
('twitter://consumer_key/consumer_secret/access_token/access_secret', {
|
|
|
|
# No user mean's we message ourselves
|
2022-10-09 09:28:18 +00:00
|
|
|
'instance': NotifyTwitter,
|
2021-11-25 20:20:22 +00:00
|
|
|
# However we'll be okay if we return a proper response
|
|
|
|
'requests_response_text': {
|
|
|
|
'id': 12345,
|
2022-03-07 22:52:22 +00:00
|
|
|
'screen_name': 'test',
|
|
|
|
# For attachment handling
|
|
|
|
'media_id': 123,
|
2021-11-25 20:20:22 +00:00
|
|
|
},
|
|
|
|
}),
|
|
|
|
# A duplicate of the entry above, this will cause cache to be referenced
|
|
|
|
('twitter://consumer_key/consumer_secret/access_token/access_secret', {
|
|
|
|
# No user mean's we message ourselves
|
2022-10-09 09:28:18 +00:00
|
|
|
'instance': NotifyTwitter,
|
2021-11-25 20:20:22 +00:00
|
|
|
# However we'll be okay if we return a proper response
|
|
|
|
'requests_response_text': {
|
|
|
|
'id': 12345,
|
2022-03-07 22:52:22 +00:00
|
|
|
'screen_name': 'test',
|
|
|
|
# For attachment handling
|
|
|
|
'media_id': 123,
|
2021-11-25 20:20:22 +00:00
|
|
|
},
|
|
|
|
}),
|
|
|
|
# handle cases where the screen_name is missing from the response causing
|
|
|
|
# an exception during parsing
|
|
|
|
('twitter://consumer_key/consumer_secret2/access_token/access_secret', {
|
|
|
|
# No user mean's we message ourselves
|
2022-10-09 09:28:18 +00:00
|
|
|
'instance': NotifyTwitter,
|
2021-11-25 20:20:22 +00:00
|
|
|
# However we'll be okay if we return a proper response
|
|
|
|
'requests_response_text': {
|
|
|
|
'id': 12345,
|
2022-03-07 22:52:22 +00:00
|
|
|
# For attachment handling
|
|
|
|
'media_id': 123,
|
2021-11-25 20:20:22 +00:00
|
|
|
},
|
|
|
|
# due to a mangled response_text we'll fail
|
|
|
|
'notify_response': False,
|
|
|
|
}),
|
|
|
|
('twitter://user@consumer_key/csecret2/access_token/access_secret/-/%/', {
|
|
|
|
# One Invalid User
|
2022-10-09 09:28:18 +00:00
|
|
|
'instance': NotifyTwitter,
|
2021-11-25 20:20:22 +00:00
|
|
|
# Expected notify() response False (because we won't be able
|
|
|
|
# to detect our user)
|
|
|
|
'notify_response': False,
|
|
|
|
}),
|
|
|
|
('twitter://user@consumer_key/csecret/access_token/access_secret'
|
2022-03-07 22:52:22 +00:00
|
|
|
'?cache=No&batch=No', {
|
|
|
|
# No Cache & No Batch
|
2022-10-09 09:28:18 +00:00
|
|
|
'instance': NotifyTwitter,
|
2021-11-25 20:20:22 +00:00
|
|
|
'requests_response_text': [{
|
|
|
|
'id': 12345,
|
|
|
|
'screen_name': 'user'
|
|
|
|
}],
|
|
|
|
}),
|
|
|
|
('twitter://user@consumer_key/csecret/access_token/access_secret', {
|
|
|
|
# We're good!
|
2022-10-09 09:28:18 +00:00
|
|
|
'instance': NotifyTwitter,
|
2021-11-25 20:20:22 +00:00
|
|
|
'requests_response_text': [{
|
|
|
|
'id': 12345,
|
|
|
|
'screen_name': 'user'
|
|
|
|
}],
|
|
|
|
}),
|
|
|
|
# A duplicate of the entry above, this will cause cache to be referenced
|
|
|
|
# for this reason, we don't even need to return a valid response
|
|
|
|
('twitter://user@consumer_key/csecret/access_token/access_secret', {
|
|
|
|
# We're identifying the same user we already sent to
|
2022-10-09 09:28:18 +00:00
|
|
|
'instance': NotifyTwitter,
|
2021-11-25 20:20:22 +00:00
|
|
|
}),
|
|
|
|
('twitter://ckey/csecret/access_token/access_secret?mode=tweet', {
|
|
|
|
# A Public Tweet
|
2022-10-09 09:28:18 +00:00
|
|
|
'instance': NotifyTwitter,
|
2021-11-25 20:20:22 +00:00
|
|
|
}),
|
|
|
|
('twitter://user@ckey/csecret/access_token/access_secret?mode=invalid', {
|
|
|
|
# An invalid mode
|
|
|
|
'instance': TypeError,
|
|
|
|
}),
|
|
|
|
('twitter://usera@consumer_key/consumer_secret/access_token/'
|
|
|
|
'access_secret/user/?to=userb', {
|
|
|
|
# We're good!
|
2022-10-09 09:28:18 +00:00
|
|
|
'instance': NotifyTwitter,
|
2021-11-25 20:20:22 +00:00
|
|
|
'requests_response_text': [{
|
|
|
|
'id': 12345,
|
|
|
|
'screen_name': 'usera'
|
|
|
|
}, {
|
|
|
|
'id': 12346,
|
|
|
|
'screen_name': 'userb'
|
|
|
|
}, {
|
|
|
|
# A garbage entry we can test exception handling on
|
|
|
|
'id': 123,
|
|
|
|
}],
|
|
|
|
}),
|
|
|
|
('twitter://ckey/csecret/access_token/access_secret', {
|
2022-10-09 09:28:18 +00:00
|
|
|
'instance': NotifyTwitter,
|
2021-11-25 20:20:22 +00:00
|
|
|
# throw a bizzare code forcing us to fail to look it up
|
|
|
|
'response': False,
|
|
|
|
'requests_response_code': 999,
|
|
|
|
}),
|
|
|
|
('twitter://ckey/csecret/access_token/access_secret', {
|
2022-10-09 09:28:18 +00:00
|
|
|
'instance': NotifyTwitter,
|
2021-11-25 20:20:22 +00:00
|
|
|
# Throws a series of connection and transfer exceptions when this flag
|
|
|
|
# is set and tests that we gracfully handle them
|
|
|
|
'test_requests_exceptions': True,
|
|
|
|
}),
|
|
|
|
('twitter://ckey/csecret/access_token/access_secret?mode=tweet', {
|
2022-10-09 09:28:18 +00:00
|
|
|
'instance': NotifyTwitter,
|
2021-11-25 20:20:22 +00:00
|
|
|
# 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_twitter_urls():
|
2017-12-25 20:07:41 +00:00
|
|
|
"""
|
2021-11-25 20:20:22 +00:00
|
|
|
NotifyTwitter() Apprise URLs
|
2017-12-25 20:07:41 +00:00
|
|
|
|
|
|
|
"""
|
|
|
|
|
2021-11-25 20:20:22 +00:00
|
|
|
# Run our general tests
|
|
|
|
AppriseURLTester(tests=apprise_url_tests).run_all()
|
2017-12-25 20:07:41 +00:00
|
|
|
|
|
|
|
|
2019-06-29 22:27:59 +00:00
|
|
|
@mock.patch('requests.get')
|
|
|
|
@mock.patch('requests.post')
|
2022-10-16 14:48:44 +00:00
|
|
|
def test_plugin_twitter_general(mock_post, mock_get):
|
2019-06-29 22:27:59 +00:00
|
|
|
"""
|
2021-11-25 20:20:22 +00:00
|
|
|
NotifyTwitter() General Tests
|
2017-12-25 20:07:41 +00:00
|
|
|
|
2019-06-29 22:27:59 +00:00
|
|
|
"""
|
|
|
|
ckey = 'ckey'
|
|
|
|
csecret = 'csecret'
|
|
|
|
akey = 'akey'
|
|
|
|
asecret = 'asecret'
|
|
|
|
screen_name = 'apprise'
|
|
|
|
|
|
|
|
response_obj = [{
|
|
|
|
'screen_name': screen_name,
|
|
|
|
'id': 9876,
|
|
|
|
}]
|
|
|
|
|
|
|
|
# Epoch time:
|
|
|
|
epoch = datetime.utcfromtimestamp(0)
|
|
|
|
|
|
|
|
request = mock.Mock()
|
|
|
|
request.content = dumps(response_obj)
|
|
|
|
request.status_code = requests.codes.ok
|
|
|
|
request.headers = {
|
|
|
|
'x-rate-limit-reset': (datetime.utcnow() - epoch).total_seconds(),
|
|
|
|
'x-rate-limit-remaining': 1,
|
|
|
|
}
|
|
|
|
|
|
|
|
# Prepare Mock
|
|
|
|
mock_get.return_value = request
|
|
|
|
mock_post.return_value = request
|
|
|
|
|
|
|
|
# Variation Initializations
|
2022-10-09 09:28:18 +00:00
|
|
|
obj = NotifyTwitter(
|
2019-06-29 22:27:59 +00:00
|
|
|
ckey=ckey,
|
|
|
|
csecret=csecret,
|
|
|
|
akey=akey,
|
|
|
|
asecret=asecret,
|
|
|
|
targets=screen_name)
|
|
|
|
|
2022-10-09 09:28:18 +00:00
|
|
|
assert isinstance(obj, NotifyTwitter) is True
|
2022-10-08 00:28:36 +00:00
|
|
|
assert isinstance(obj.url(), str) is True
|
2019-06-29 22:27:59 +00:00
|
|
|
|
|
|
|
# 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-rate-limit-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-rate-limit-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-rate-limit-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-rate-limit-remaining'] = 1
|
|
|
|
|
|
|
|
# Handle cases where our epoch time is wrong
|
|
|
|
del request.headers['x-rate-limit-reset']
|
|
|
|
assert obj.send(body="test") is True
|
|
|
|
|
|
|
|
# Return our object, but place it in the future forcing us to block
|
|
|
|
request.headers['x-rate-limit-reset'] = \
|
|
|
|
(datetime.utcnow() - epoch).total_seconds() + 1
|
|
|
|
request.headers['x-rate-limit-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-rate-limit-reset'] = \
|
|
|
|
(datetime.utcnow() - epoch).total_seconds() - 1
|
|
|
|
request.headers['x-rate-limit-remaining'] = 0
|
|
|
|
obj.ratelimit_remaining = 0
|
|
|
|
assert obj.send(body="test") is True
|
|
|
|
|
|
|
|
# Return our limits to always work
|
|
|
|
request.headers['x-rate-limit-reset'] = \
|
|
|
|
(datetime.utcnow() - epoch).total_seconds()
|
|
|
|
request.headers['x-rate-limit-remaining'] = 1
|
|
|
|
obj.ratelimit_remaining = 1
|
|
|
|
|
|
|
|
# Alter pending targets
|
|
|
|
obj.targets.append('usera')
|
|
|
|
request.content = dumps(response_obj)
|
|
|
|
response_obj = [{
|
|
|
|
'screen_name': 'usera',
|
|
|
|
'id': 1234,
|
|
|
|
}]
|
|
|
|
|
|
|
|
assert obj.send(body="test") is True
|
|
|
|
|
|
|
|
# Flush our cache forcing it's re-creating
|
2022-10-09 09:28:18 +00:00
|
|
|
del NotifyTwitter._user_cache
|
2019-06-29 22:27:59 +00:00
|
|
|
assert obj.send(body="test") is True
|
|
|
|
|
|
|
|
# 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 = '{}'
|
|
|
|
|
2022-10-09 09:28:18 +00:00
|
|
|
results = NotifyTwitter.parse_url(
|
2019-06-29 22:27:59 +00:00
|
|
|
'twitter://{}/{}/{}/{}?to={}'.format(
|
|
|
|
ckey, csecret, akey, asecret, screen_name))
|
|
|
|
assert isinstance(results, dict) is True
|
|
|
|
assert screen_name 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
|
|
|
|
|
|
|
|
# Set ourselves up to handle whoami calls
|
|
|
|
|
|
|
|
# Flush out our cache
|
2022-10-09 09:28:18 +00:00
|
|
|
del NotifyTwitter._user_cache
|
2019-06-29 22:27:59 +00:00
|
|
|
|
|
|
|
response_obj = {
|
|
|
|
'screen_name': screen_name,
|
|
|
|
'id': 9876,
|
|
|
|
}
|
|
|
|
request.content = dumps(response_obj)
|
|
|
|
|
2022-10-09 09:28:18 +00:00
|
|
|
obj = NotifyTwitter(
|
2019-06-29 22:27:59 +00:00
|
|
|
ckey=ckey,
|
|
|
|
csecret=csecret,
|
|
|
|
akey=akey,
|
|
|
|
asecret=asecret)
|
|
|
|
|
|
|
|
assert obj.send(body="test") is True
|
|
|
|
|
|
|
|
# Alter the key forcing us to look up a new value of ourselves again
|
2022-10-09 09:28:18 +00:00
|
|
|
del NotifyTwitter._user_cache
|
|
|
|
del NotifyTwitter._whoami_cache
|
2019-06-29 22:27:59 +00:00
|
|
|
obj.ckey = 'different.then.it.was'
|
|
|
|
assert obj.send(body="test") is True
|
|
|
|
|
2022-10-09 09:28:18 +00:00
|
|
|
del NotifyTwitter._whoami_cache
|
2019-06-29 22:27:59 +00:00
|
|
|
obj.ckey = 'different.again'
|
|
|
|
assert obj.send(body="test") is True
|
2021-11-25 20:20:22 +00:00
|
|
|
|
|
|
|
|
|
|
|
def test_plugin_twitter_edge_cases():
|
|
|
|
"""
|
|
|
|
NotifyTwitter() Edge Cases
|
|
|
|
|
|
|
|
"""
|
|
|
|
|
|
|
|
with pytest.raises(TypeError):
|
2022-10-09 09:28:18 +00:00
|
|
|
NotifyTwitter(
|
2021-11-25 20:20:22 +00:00
|
|
|
ckey=None, csecret=None, akey=None, asecret=None)
|
|
|
|
|
|
|
|
with pytest.raises(TypeError):
|
2022-10-09 09:28:18 +00:00
|
|
|
NotifyTwitter(
|
2021-11-25 20:20:22 +00:00
|
|
|
ckey='value', csecret=None, akey=None, asecret=None)
|
|
|
|
|
|
|
|
with pytest.raises(TypeError):
|
2022-10-09 09:28:18 +00:00
|
|
|
NotifyTwitter(
|
2021-11-25 20:20:22 +00:00
|
|
|
ckey='value', csecret='value', akey=None, asecret=None)
|
|
|
|
|
|
|
|
with pytest.raises(TypeError):
|
2022-10-09 09:28:18 +00:00
|
|
|
NotifyTwitter(
|
2021-11-25 20:20:22 +00:00
|
|
|
ckey='value', csecret='value', akey='value', asecret=None)
|
|
|
|
|
|
|
|
assert isinstance(
|
2022-10-09 09:28:18 +00:00
|
|
|
NotifyTwitter(
|
2021-11-25 20:20:22 +00:00
|
|
|
ckey='value', csecret='value', akey='value', asecret='value'),
|
2022-10-09 09:28:18 +00:00
|
|
|
NotifyTwitter,
|
2021-11-25 20:20:22 +00:00
|
|
|
)
|
|
|
|
|
|
|
|
assert isinstance(
|
2022-10-09 09:28:18 +00:00
|
|
|
NotifyTwitter(
|
2021-11-25 20:20:22 +00:00
|
|
|
ckey='value', csecret='value', akey='value', asecret='value',
|
|
|
|
user='l2gnux'),
|
2022-10-09 09:28:18 +00:00
|
|
|
NotifyTwitter,
|
2021-11-25 20:20:22 +00:00
|
|
|
)
|
|
|
|
|
|
|
|
# Invalid Target User
|
|
|
|
with pytest.raises(TypeError):
|
2022-10-09 09:28:18 +00:00
|
|
|
NotifyTwitter(
|
2021-11-25 20:20:22 +00:00
|
|
|
ckey='value', csecret='value', akey='value', asecret='value',
|
|
|
|
targets='%G@rB@g3')
|
2022-03-07 22:52:22 +00:00
|
|
|
|
|
|
|
|
2022-10-18 11:18:50 +00:00
|
|
|
@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")
|
2022-03-07 22:52:22 +00:00
|
|
|
@mock.patch('requests.post')
|
|
|
|
@mock.patch('requests.get')
|
2022-10-16 14:48:44 +00:00
|
|
|
def test_plugin_twitter_dm_attachments(mock_get, mock_post):
|
2022-03-07 22:52:22 +00:00
|
|
|
"""
|
|
|
|
NotifyTwitter() DM Attachment Checks
|
|
|
|
|
|
|
|
"""
|
|
|
|
ckey = 'ckey'
|
|
|
|
csecret = 'csecret'
|
|
|
|
akey = 'akey'
|
|
|
|
asecret = 'asecret'
|
|
|
|
screen_name = 'apprise'
|
|
|
|
|
|
|
|
good_dm_response_obj = {
|
|
|
|
'screen_name': screen_name,
|
|
|
|
'id': 9876,
|
|
|
|
}
|
|
|
|
|
|
|
|
# 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
|
|
|
|
|
|
|
|
# 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_dm_response]
|
|
|
|
mock_get.return_value = good_dm_response
|
|
|
|
|
|
|
|
twitter_url = 'twitter://{}/{}/{}/{}'.format(ckey, csecret, akey, asecret)
|
|
|
|
|
|
|
|
# attach our content
|
|
|
|
attach = AppriseAttachment(os.path.join(TEST_VAR_DIR, 'apprise-test.gif'))
|
|
|
|
|
|
|
|
# instantiate our object
|
|
|
|
obj = Apprise.instantiate(twitter_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_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]
|
|
|
|
|
|
|
|
# instantiate our object
|
|
|
|
obj = Apprise.instantiate(twitter_url)
|
|
|
|
|
|
|
|
# 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
|
|
|
|
|
|
|
|
# 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_args_list[0][0][0] == \
|
|
|
|
'https://upload.twitter.com/1.1/media/upload.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]
|
|
|
|
|
|
|
|
# instantiate our object
|
|
|
|
obj = Apprise.instantiate(twitter_url)
|
|
|
|
|
|
|
|
# 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 == 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'
|
|
|
|
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
|
|
|
|
|
|
|
|
# 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 request as cached response is used
|
|
|
|
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()
|
|
|
|
|
|
|
|
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
|
|
|
|
|
|
|
|
# 4 images are produced
|
|
|
|
attach = [
|
|
|
|
os.path.join(TEST_VAR_DIR, 'apprise-test.gif'),
|
|
|
|
os.path.join(TEST_VAR_DIR, 'apprise-test.gif'),
|
|
|
|
os.path.join(TEST_VAR_DIR, 'apprise-test.jpeg'),
|
|
|
|
os.path.join(TEST_VAR_DIR, 'apprise-test.png'),
|
|
|
|
# This one is not supported, so it's ignored gracefully
|
|
|
|
os.path.join(TEST_VAR_DIR, 'apprise-test.mp4'),
|
|
|
|
]
|
|
|
|
|
|
|
|
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'
|
|
|
|
assert mock_post.call_args_list[1][0][0] == \
|
|
|
|
'https://upload.twitter.com/1.1/media/upload.json'
|
|
|
|
assert mock_post.call_args_list[2][0][0] == \
|
|
|
|
'https://upload.twitter.com/1.1/media/upload.json'
|
|
|
|
assert mock_post.call_args_list[3][0][0] == \
|
|
|
|
'https://upload.twitter.com/1.1/media/upload.json'
|
|
|
|
assert mock_post.call_args_list[4][0][0] == \
|
|
|
|
'https://api.twitter.com/1.1/direct_messages/events/new.json'
|
|
|
|
assert mock_post.call_args_list[5][0][0] == \
|
|
|
|
'https://api.twitter.com/1.1/direct_messages/events/new.json'
|
|
|
|
assert mock_post.call_args_list[6][0][0] == \
|
|
|
|
'https://api.twitter.com/1.1/direct_messages/events/new.json'
|
|
|
|
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
|
|
|
|
mock_post.side_effect = [good_media_response, OSError()]
|
|
|
|
mock_get.return_value = good_dm_response
|
|
|
|
|
|
|
|
# 2 images are produced
|
|
|
|
attach = [
|
|
|
|
os.path.join(TEST_VAR_DIR, 'apprise-test.gif'),
|
|
|
|
os.path.join(TEST_VAR_DIR, 'apprise-test.png'),
|
|
|
|
# This one is not supported, so it's ignored gracefully
|
|
|
|
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://upload.twitter.com/1.1/media/upload.json'
|
|
|
|
assert mock_post.call_args_list[1][0][0] == \
|
|
|
|
'https://upload.twitter.com/1.1/media/upload.json'
|
|
|
|
|
|
|
|
|
|
|
|
@mock.patch('requests.post')
|
|
|
|
@mock.patch('requests.get')
|
2022-10-16 14:48:44 +00:00
|
|
|
def test_plugin_twitter_tweet_attachments(mock_get, mock_post):
|
2022-03-07 22:52:22 +00:00
|
|
|
"""
|
|
|
|
NotifyTwitter() Tweet Attachment Checks
|
|
|
|
|
|
|
|
"""
|
|
|
|
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
|
|
|
|
|
|
|
|
# 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
|
|
|
|
obj = Apprise.instantiate(twitter_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] == \
|
|
|
|
'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'
|
|
|
|
|
2022-03-11 01:59:40 +00:00
|
|
|
# Update our good response to have more details in it
|
|
|
|
good_tweet_response_obj = {
|
|
|
|
'screen_name': screen_name,
|
|
|
|
'id': 9876,
|
|
|
|
# needed for additional logging
|
|
|
|
'id_str': '12345',
|
|
|
|
'user': {
|
|
|
|
'screen_name': 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
|
|
|
|
obj = Apprise.instantiate(twitter_url)
|
|
|
|
|
|
|
|
# Send our notification (again); this time there willb e more tweet logging
|
|
|
|
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] == \
|
|
|
|
'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]
|
|
|
|
|
|
|
|
# instantiate our object
|
|
|
|
obj = Apprise.instantiate(twitter_url)
|
|
|
|
|
|
|
|
# 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
|
|
|
|
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 = ''
|
|
|
|
|
|
|
|
mock_post.side_effect = [good_media_response, bad_media_response]
|
|
|
|
|
|
|
|
# instantiate our object
|
|
|
|
obj = Apprise.instantiate(twitter_url)
|
|
|
|
|
|
|
|
# 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
|
|
|
|
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://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'
|
|
|
|
|
2022-03-07 22:52:22 +00:00
|
|
|
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]
|
|
|
|
|
|
|
|
# instantiate our object
|
|
|
|
obj = Apprise.instantiate(twitter_url)
|
|
|
|
|
|
|
|
# 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 == 0
|
|
|
|
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
|
|
|
|
|
|
|
|
# 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 request as cached response is used
|
|
|
|
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()
|
|
|
|
|
|
|
|
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
|
|
|
|
|
|
|
|
# instantiate our object (without a batch mode)
|
|
|
|
obj = Apprise.instantiate(twitter_url + "&batch=no")
|
|
|
|
|
|
|
|
# 4 images are produced
|
|
|
|
attach = [
|
|
|
|
os.path.join(TEST_VAR_DIR, 'apprise-test.gif'),
|
|
|
|
os.path.join(TEST_VAR_DIR, 'apprise-test.gif'),
|
|
|
|
os.path.join(TEST_VAR_DIR, 'apprise-test.jpeg'),
|
|
|
|
os.path.join(TEST_VAR_DIR, 'apprise-test.png'),
|
|
|
|
# This one is not supported, so it's ignored gracefully
|
|
|
|
os.path.join(TEST_VAR_DIR, 'apprise-test.mp4'),
|
|
|
|
]
|
|
|
|
|
|
|
|
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'
|
|
|
|
assert mock_post.call_args_list[1][0][0] == \
|
|
|
|
'https://upload.twitter.com/1.1/media/upload.json'
|
|
|
|
assert mock_post.call_args_list[2][0][0] == \
|
|
|
|
'https://upload.twitter.com/1.1/media/upload.json'
|
|
|
|
assert mock_post.call_args_list[3][0][0] == \
|
|
|
|
'https://upload.twitter.com/1.1/media/upload.json'
|
|
|
|
assert mock_post.call_args_list[4][0][0] == \
|
|
|
|
'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'
|
|
|
|
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()
|
|
|
|
|
|
|
|
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
|
|
|
|
|
|
|
|
# instantiate our object
|
|
|
|
obj = Apprise.instantiate(twitter_url)
|
|
|
|
|
|
|
|
# 4 images are produced
|
|
|
|
attach = [
|
|
|
|
os.path.join(TEST_VAR_DIR, 'apprise-test.gif'),
|
|
|
|
os.path.join(TEST_VAR_DIR, 'apprise-test.gif'),
|
|
|
|
os.path.join(TEST_VAR_DIR, 'apprise-test.jpeg'),
|
|
|
|
os.path.join(TEST_VAR_DIR, 'apprise-test.png'),
|
|
|
|
# This one is not supported, so it's ignored gracefully
|
|
|
|
os.path.join(TEST_VAR_DIR, 'apprise-test.mp4'),
|
|
|
|
]
|
|
|
|
|
|
|
|
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 == 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] == \
|
|
|
|
'https://upload.twitter.com/1.1/media/upload.json'
|
|
|
|
assert mock_post.call_args_list[2][0][0] == \
|
|
|
|
'https://upload.twitter.com/1.1/media/upload.json'
|
|
|
|
assert mock_post.call_args_list[3][0][0] == \
|
|
|
|
'https://upload.twitter.com/1.1/media/upload.json'
|
|
|
|
assert mock_post.call_args_list[4][0][0] == \
|
|
|
|
'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'
|
|
|
|
|
|
|
|
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()]
|
|
|
|
mock_get.return_value = good_tweet_response
|
|
|
|
|
|
|
|
# 2 images are produced
|
|
|
|
attach = [
|
|
|
|
os.path.join(TEST_VAR_DIR, 'apprise-test.gif'),
|
|
|
|
os.path.join(TEST_VAR_DIR, 'apprise-test.png'),
|
|
|
|
# This one is not supported, so it's ignored gracefully
|
|
|
|
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://upload.twitter.com/1.1/media/upload.json'
|
|
|
|
assert mock_post.call_args_list[1][0][0] == \
|
|
|
|
'https://upload.twitter.com/1.1/media/upload.json'
|