github
/
apprise
mirror of https://github.com/caronc/apprise
1
0
Fork 0
Code Issues Releases Wiki Activity

Twitter plugin code cleanup (#750)

Browse Source
pull/751/head
Chris Caron 2 years ago committed by GitHub
parent 6d3ab9b3fd
commit 81907af448
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 65 additions and 71 deletions
Show Stats Download Patch File Download Diff File

77
apprise/plugins/NotifyTwitter.py
Unescape View File

@ -75,7 +75,7 @@ class NotifyTwitter(NotifyBase):
service_url = 'https://twitter.com/'
# The default secure protocol is twitter.
secure_protocol = 'twitter'
secure_protocol = ('twitter', 'tweet')
# A URL that takes you to the setup/help of the specific protocol
setup_url = 'https://github.com/caronc/apprise/wiki/Notify_twitter'
@ -221,21 +221,21 @@ class NotifyTwitter(NotifyBase):
raise TypeError(msg)
# Store our webhook mode
self.mode = None \
self.mode = self.template_args['mode']['default'] \
if not isinstance(mode, str) else mode.lower()
# Set Cache Flag
self.cache = cache
# Prepare Image Batch Mode Flag
self.batch = batch
if self.mode not in TWITTER_MESSAGE_MODES:
msg = 'The Twitter message mode specified ({}) is invalid.' \
.format(mode)
self.logger.warning(msg)
raise TypeError(msg)
# Set Cache Flag
self.cache = cache
# Prepare Image Batch Mode Flag
self.batch = batch
# Track any errors
has_error = False
@ -249,7 +249,7 @@ class NotifyTwitter(NotifyBase):
has_error = True
self.logger.warning(
'Dropped invalid user ({}) specified.'.format(target),
'Dropped invalid Twitter user ({}) specified.'.format(target),
)
if has_error and not self.targets:
@ -261,6 +261,10 @@ class NotifyTwitter(NotifyBase):
self.logger.warning(msg)
raise TypeError(msg)
# Initialize our cache values
self._whoami_cache = None
self._user_cache = {}
return
def send(self, body, title='', notify_type=NotifyType.INFO, attach=None,
@ -293,7 +297,7 @@ class NotifyTwitter(NotifyBase):
continue
self.logger.debug(
'Preparing Twiter attachment {}'.format(
'Preparing Twitter attachment {}'.format(
attachment.url(privacy=True)))
# Upload our image and get our id associated with it
@ -536,16 +540,9 @@ class NotifyTwitter(NotifyBase):
"""
# Prepare a whoami key; this is to prevent conflict with other
# NotifyTwitter declarations that may or may not use a different
# set of authentication keys
whoami_key = '{}{}{}{}'.format(
self.ckey, self.csecret, self.akey, self.asecret)
if lazy and hasattr(NotifyTwitter, '_whoami_cache') \
and whoami_key in getattr(NotifyTwitter, '_whoami_cache'):
if lazy and self._whoami_cache is not None:
# Use cached response
return getattr(NotifyTwitter, '_whoami_cache')[whoami_key]
return self._whoami_cache
# Contains a mapping of screen_name to id
results = {}
@ -560,22 +557,11 @@ class NotifyTwitter(NotifyBase):
if postokay:
try:
results[response['screen_name']] = response['id']
self._whoami_cache = {
response['screen_name']: response['id'],
}
if lazy:
# Cache our response for future references
if not hasattr(NotifyTwitter, '_whoami_cache'):
setattr(
NotifyTwitter, '_whoami_cache',
{whoami_key: results})
else:
getattr(NotifyTwitter, '_whoami_cache')\
.update({whoami_key: results})
# Update our user cache as well
if not hasattr(NotifyTwitter, '_user_cache'):
setattr(NotifyTwitter, '_user_cache', results)
else:
getattr(NotifyTwitter, '_user_cache').update(results)
self._user_cache.update(results)
except (TypeError, KeyError):
pass
@ -595,10 +581,10 @@ class NotifyTwitter(NotifyBase):
# Build a unique set of names
names = parse_list(screen_name)
if lazy and hasattr(NotifyTwitter, '_user_cache'):
if lazy and self._user_cache:
# Use cached response
results = {k: v for k, v in getattr(
NotifyTwitter, '_user_cache').items() if k in names}
results = {
k: v for k, v in self._user_cache.items() if k in names}
# limit our names if they already exist in our cache
names = [name for name in names if name not in results]
@ -612,7 +598,7 @@ class NotifyTwitter(NotifyBase):
# https://developer.twitter.com/en/docs/accounts-and-users/\
# follow-search-get-users/api-reference/get-users-lookup
for i in range(0, len(names), 100):
# Send Twitter DM
# Look up our names by their screen_name
postokay, response = self._fetch(
self.twitter_lookup,
payload={
@ -635,11 +621,7 @@ class NotifyTwitter(NotifyBase):
# Cache our response for future use; this saves on un-nessisary extra
# hits against the Twitter API when we already know the answer
if lazy:
if not hasattr(NotifyTwitter, '_user_cache'):
setattr(NotifyTwitter, '_user_cache', results)
else:
getattr(NotifyTwitter, '_user_cache').update(results)
self._user_cache.update(results)
return results
@ -686,7 +668,7 @@ class NotifyTwitter(NotifyBase):
# Determine how long we should wait for or if we should wait at
# all. This isn't fool-proof because we can't be sure the client
# time (calling this script) is completely synced up with the
# Gitter server. One would hope we're on NTP and our clocks are
# Twitter server. One would hope we're on NTP and our clocks are
# the same allowing this to role smoothly:
now = datetime.utcnow()
@ -810,7 +792,7 @@ class NotifyTwitter(NotifyBase):
return '{schema}://{ckey}/{csecret}/{akey}/{asecret}' \
'/{targets}/?{params}'.format(
schema=self.secure_protocol,
schema=self.secure_protocol[0],
ckey=self.pprint(self.ckey, privacy, safe=''),
csecret=self.pprint(
self.csecret, privacy, mode=PrivacyMode.Secret, safe=''),
@ -818,7 +800,7 @@ class NotifyTwitter(NotifyBase):
asecret=self.pprint(
self.asecret, privacy, mode=PrivacyMode.Secret, safe=''),
targets='/'.join(
[NotifyTwitter.quote('@{}'.format(target), safe='')
[NotifyTwitter.quote('@{}'.format(target), safe='@')
for target in self.targets]),
params=NotifyTwitter.urlencode(params))
@ -862,6 +844,9 @@ class NotifyTwitter(NotifyBase):
results['mode'] = \
NotifyTwitter.unquote(results['qsd']['mode'])
elif results['schema'].startswith('tweet'):
results['mode'] = TwitterMessageMode.TWEET
results['targets'] = []
# if a user has been defined, add it to the list of targets

59
test/test_plugin_twitter.py
Unescape View File

@ -64,11 +64,11 @@ apprise_url_tests = (
# Missing Keys
'instance': TypeError,
}),
('twitter://consumer_key/consumer_secret/access_token/', {
('twitter://consumer_key/consumer_secret/atoken1/', {
# Missing Access Secret
'instance': TypeError,
}),
('twitter://consumer_key/consumer_secret/access_token/access_secret', {
('twitter://consumer_key/consumer_secret/atoken2/access_secret', {
# No user mean's we message ourselves
'instance': NotifyTwitter,
# Expected notify() response False (because we won't be able
@ -76,9 +76,9 @@ apprise_url_tests = (
'notify_response': False,
# Our expected url(privacy=True) startswith() response:
'privacy_url': 'twitter://c...y/****/a...n/****',
'privacy_url': 'twitter://c...y/****/a...2/****',
}),
('twitter://consumer_key/consumer_secret/access_token/access_secret'
('twitter://consumer_key/consumer_secret/atoken3/access_secret'
'?cache=no', {
# No user mean's we message ourselves
'instance': NotifyTwitter,
@ -90,7 +90,7 @@ apprise_url_tests = (
'media_id': 123,
},
}),
('twitter://consumer_key/consumer_secret/access_token/access_secret', {
('twitter://consumer_key/consumer_secret/atoken4/access_secret', {
# No user mean's we message ourselves
'instance': NotifyTwitter,
# However we'll be okay if we return a proper response
@ -102,7 +102,7 @@ apprise_url_tests = (
},
}),
# A duplicate of the entry above, this will cause cache to be referenced
('twitter://consumer_key/consumer_secret/access_token/access_secret', {
('twitter://consumer_key/consumer_secret/atoken5/access_secret', {
# No user mean's we message ourselves
'instance': NotifyTwitter,
# However we'll be okay if we return a proper response
@ -115,7 +115,7 @@ apprise_url_tests = (
}),
# 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', {
('twitter://consumer_key/consumer_secret2/atoken6/access_secret', {
# No user mean's we message ourselves
'instance': NotifyTwitter,
# However we'll be okay if we return a proper response
@ -127,14 +127,14 @@ apprise_url_tests = (
# due to a mangled response_text we'll fail
'notify_response': False,
}),
('twitter://user@consumer_key/csecret2/access_token/access_secret/-/%/', {
('twitter://user@consumer_key/csecret2/atoken7/access_secret/-/%/', {
# One Invalid User
'instance': NotifyTwitter,
# 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'
('twitter://user@consumer_key/csecret/atoken8/access_secret'
'?cache=No&batch=No', {
# No Cache & No Batch
'instance': NotifyTwitter,
@ -143,7 +143,7 @@ apprise_url_tests = (
'screen_name': 'user'
}],
}),
('twitter://user@consumer_key/csecret/access_token/access_secret', {
('twitter://user@consumer_key/csecret/atoken9/access_secret', {
# We're good!
'instance': NotifyTwitter,
'requests_response_text': [{
@ -151,21 +151,20 @@ apprise_url_tests = (
'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', {
('twitter://user@consumer_key/csecret/atoken11/access_secret', {
# We're identifying the same user we already sent to
'instance': NotifyTwitter,
'notify_response': False,
}),
('twitter://ckey/csecret/access_token/access_secret?mode=tweet', {
('tweet://ckey/csecret/atoken12/access_secret', {
# A Public Tweet
'instance': NotifyTwitter,
}),
('twitter://user@ckey/csecret/access_token/access_secret?mode=invalid', {
('twitter://user@ckey/csecret/atoken13/access_secret?mode=invalid', {
# An invalid mode
'instance': TypeError,
}),
('twitter://usera@consumer_key/consumer_secret/access_token/'
('twitter://usera@consumer_key/consumer_secret/atoken14/'
'access_secret/user/?to=userb', {
# We're good!
'instance': NotifyTwitter,
@ -180,19 +179,19 @@ apprise_url_tests = (
'id': 123,
}],
}),
('twitter://ckey/csecret/access_token/access_secret', {
('twitter://ckey/csecret/atoken15/access_secret', {
'instance': NotifyTwitter,
# 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', {
('twitter://ckey/csecret/atoken16/access_secret', {
'instance': NotifyTwitter,
# 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', {
('twitter://ckey/csecret/atoken17/access_secret?mode=tweet', {
'instance': NotifyTwitter,
# Throws a series of connection and transfer exceptions when this flag
# is set and tests that we gracfully handle them
@ -320,7 +319,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
del NotifyTwitter._user_cache
NotifyTwitter._user_cache = {}
assert obj.send(body="test") is True
# Cause content response to be None
@ -350,7 +349,7 @@ def test_plugin_twitter_general(mock_post, mock_get):
# Set ourselves up to handle whoami calls
# Flush out our cache
del NotifyTwitter._user_cache
NotifyTwitter._user_cache = {}
response_obj = {
'screen_name': screen_name,
@ -367,12 +366,12 @@ def test_plugin_twitter_general(mock_post, mock_get):
assert obj.send(body="test") is True
# Alter the key forcing us to look up a new value of ourselves again
del NotifyTwitter._user_cache
del NotifyTwitter._whoami_cache
NotifyTwitter._user_cache = {}
NotifyTwitter._whoami_cache = None
obj.ckey = 'different.then.it.was'
assert obj.send(body="test") is True
del NotifyTwitter._whoami_cache
NotifyTwitter._whoami_cache = None
obj.ckey = 'different.again'
assert obj.send(body="test") is True
@ -440,10 +439,17 @@ def test_plugin_twitter_dm_attachments(mock_get, mock_post):
'id': 9876,
}
# Epoch time:
epoch = datetime.utcfromtimestamp(0)
# 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,
}
# Prepare bad response
bad_response = mock.Mock()
@ -540,7 +546,10 @@ def test_plugin_twitter_dm_attachments(mock_get, mock_post):
body='body', title='title', notify_type=NotifyType.INFO,
attach=attach) is False
assert mock_get.call_count == 0
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
assert mock_post.call_args_list[0][0][0] == \

Write Preview
Loading…
Cancel
Save