From a03525a85946a5d52d37e34f4018f13f14903597 Mon Sep 17 00:00:00 2001 From: Chris Caron Date: Mon, 25 Dec 2017 15:07:41 -0500 Subject: [PATCH] 100% test coverage --- .../plugins/NotifyPushjet/NotifyPushjet.py | 7 +- apprise/plugins/NotifyRocketChat.py | 118 ++++--- .../plugins/NotifyTwitter/NotifyTwitter.py | 44 ++- apprise/plugins/__init__.py | 16 +- test/test_pushjet_plugin.py | 157 +++++++++ test/test_rest_plugins.py | 313 ++++++++++++++++-- test/test_twitter_plugin.py | 189 +++++++++++ 7 files changed, 736 insertions(+), 108 deletions(-) create mode 100644 test/test_pushjet_plugin.py create mode 100644 test/test_twitter_plugin.py diff --git a/apprise/plugins/NotifyPushjet/NotifyPushjet.py b/apprise/plugins/NotifyPushjet/NotifyPushjet.py index 78790867..356da3fb 100644 --- a/apprise/plugins/NotifyPushjet/NotifyPushjet.py +++ b/apprise/plugins/NotifyPushjet/NotifyPushjet.py @@ -15,12 +15,17 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Lesser General Public License for more details. - +import re from .pushjet import errors from .pushjet import pushjet from ..NotifyBase import NotifyBase +PUBLIC_KEY_RE = re.compile( + r'^[a-z0-9]{4}-[a-z0-9]{6}-[a-z0-9]{12}-[a-z0-9]{5}-[a-z0-9]{9}$', re.I) + +SECRET_KEY_RE = re.compile(r'^[a-z0-9]{32}$', re.I) + class NotifyPushjet(NotifyBase): """ diff --git a/apprise/plugins/NotifyRocketChat.py b/apprise/plugins/NotifyRocketChat.py index 399575c6..c8813998 100644 --- a/apprise/plugins/NotifyRocketChat.py +++ b/apprise/plugins/NotifyRocketChat.py @@ -74,16 +74,16 @@ class NotifyRocketChat(NotifyBase): # Initialize channels list self.channels = list() - # Initialize room_id list - self.room_ids = list() + # Initialize room list + self.rooms = list() if recipients is None: recipients = [] elif compat_is_basestring(recipients): - recipients = filter(bool, LIST_DELIM.split( + recipients = [x for x in filter(bool, LIST_DELIM.split( recipients, - )) + ))] elif not isinstance(recipients, (set, tuple, list)): recipients = [] @@ -98,62 +98,88 @@ class NotifyRocketChat(NotifyBase): result = IS_ROOM_ID.match(recipient) if result: - # store valid room_id - self.channels.append(result.group('name')) + # store valid room + self.rooms.append(result.group('name')) continue self.logger.warning( - 'Dropped invalid channel/room_id ' + + 'Dropped invalid channel/room ' + '(%s) specified.' % recipient, ) - if len(self.room_ids) == 0 and len(self.channels) == 0: + if len(self.rooms) == 0 and len(self.channels) == 0: raise TypeError( - 'No Rocket.Chat room_id and/or channels specified to notify.' + 'No Rocket.Chat room and/or channels specified to notify.' ) # Used to track token headers upon authentication (if successful) self.headers = {} - # Track whether we authenticated okay - self.authenticated = self.login() - - if not self.authenticated: - raise TypeError( - 'Authentication to Rocket.Chat server failed.' - ) - def notify(self, title, body, notify_type, **kwargs): """ wrapper to send_notification since we can alert more then one channel """ + # Track whether we authenticated okay + + if not self.login(): + return False + # Prepare our message text = '*%s*\r\n%s' % (title.replace('*', '\*'), body) - # Send all our defined channels - for channel in self.channels: - self.send_notification({ - 'text': text, - 'channel': channel, - }, notify_type=notify_type, **kwargs) + # Initiaize our error tracking + has_error = False + + # Create a copy of our rooms and channels to notify against + channels = list(self.channels) + rooms = list(self.rooms) + + while len(channels) > 0: + # Get Channel + channel = channels.pop(0) + + if not self.send_notification( + { + 'text': text, + 'channel': channel, + }, notify_type=notify_type, **kwargs): + + # toggle flag + has_error = True + + if len(channels) + len(rooms) > 0: + # Prevent thrashing requests + self.throttle() # Send all our defined room id's - for room_id in self.room_ids: - self.send_notification({ - 'text': text, - 'roomId': room_id, - }, notify_type=notify_type, **kwargs) + while len(rooms): + # Get Room + room = rooms.pop(0) + + if not self.send_notification( + { + 'text': text, + 'roomId': room, + }, notify_type=notify_type, **kwargs): + + # toggle flag + has_error = True + + if len(rooms) > 0: + # Prevent thrashing requests + self.throttle() + + # logout + self.logout() + + return not has_error def send_notification(self, payload, notify_type, **kwargs): """ Perform Notify Rocket.Chat Notification """ - if not self.authenticated: - # We couldn't authenticate; we're done - return False - self.logger.debug('Rocket.Chat POST URL: %s (cert_verify=%r)' % ( self.api_url + 'chat.postMessage', self.verify_certificate, )) @@ -173,6 +199,7 @@ class NotifyRocketChat(NotifyBase): '%s (error=%s).' % ( RC_HTTP_ERROR_MAP[r.status_code], r.status_code)) + except KeyError: self.logger.warning( 'Failed to send Rocket.Chat notification ' + @@ -200,6 +227,7 @@ class NotifyRocketChat(NotifyBase): def login(self): """ login to our server + """ payload = { 'username': self.user, @@ -220,7 +248,8 @@ class NotifyRocketChat(NotifyBase): '%s (error=%s).' % ( RC_HTTP_ERROR_MAP[r.status_code], r.status_code)) - except IndexError: + + except KeyError: self.logger.warning( 'Failed to authenticate with Rocket.Chat server ' + '(error=%s).' % ( @@ -238,15 +267,12 @@ class NotifyRocketChat(NotifyBase): return False # Set our headers for further communication - self.headers['X-Auth-Token'] = \ - response.get('data').get('authToken') - self.headers['X-User-Id'] = \ - response.get('data').get('userId') + self.headers['X-Auth-Token'] = response.get( + 'data', {'authToken': None}).get('authToken') + self.headers['X-User-Id'] = response.get( + 'data', {'userId': None}).get('userId') - # We're authenticated now - self.authenticated = True - - except requests.ConnectionError as e: + except requests.RequestException as e: self.logger.warning( 'A Connection error occured authenticating to the ' + 'Rocket.Chat server.') @@ -259,10 +285,6 @@ class NotifyRocketChat(NotifyBase): """ logout of our server """ - if not self.authenticated: - # Nothing to do - return True - try: r = requests.post( self.api_url + 'logout', @@ -278,7 +300,7 @@ class NotifyRocketChat(NotifyBase): RC_HTTP_ERROR_MAP[r.status_code], r.status_code)) - except IndexError: + except KeyError: self.logger.warning( 'Failed to log off Rocket.Chat server ' + '(error=%s).' % ( @@ -292,15 +314,13 @@ class NotifyRocketChat(NotifyBase): 'Rocket.Chat log off successful; response %s.' % ( r.text)) - except requests.ConnectionError as e: + except requests.RequestException as e: self.logger.warning( 'A Connection error occured logging off the ' + 'Rocket.Chat server') self.logger.debug('Socket Exception: %s' % str(e)) return False - # We're no longer authenticated now - self.authenticated = False return True @staticmethod diff --git a/apprise/plugins/NotifyTwitter/NotifyTwitter.py b/apprise/plugins/NotifyTwitter/NotifyTwitter.py index 7a511109..c134ad3b 100644 --- a/apprise/plugins/NotifyTwitter/NotifyTwitter.py +++ b/apprise/plugins/NotifyTwitter/NotifyTwitter.py @@ -68,18 +68,11 @@ class NotifyTwitter(NotifyBase): 'No user was specified.' ) - try: - # Attempt to Establish a connection to Twitter - self.auth = tweepy.OAuthHandler(ckey, csecret) - - # Apply our Access Tokens - self.auth.set_access_token(akey, asecret) - - except Exception: - raise TypeError( - 'Twitter authentication failed; ' - 'please verify your configuration.' - ) + # Store our data + self.ckey = ckey + self.csecret = csecret + self.akey = akey + self.asecret = asecret return @@ -88,6 +81,20 @@ class NotifyTwitter(NotifyBase): Perform Twitter Notification """ + try: + # Attempt to Establish a connection to Twitter + self.auth = tweepy.OAuthHandler(self.ckey, self.csecret) + + # Apply our Access Tokens + self.auth.set_access_token(self.akey, self.asecret) + + except Exception: + self.logger.warning( + 'Twitter authentication failed; ' + 'please verify your configuration.' + ) + return False + text = '%s\r\n%s' % (title, body) try: # Get our API @@ -128,18 +135,19 @@ class NotifyTwitter(NotifyBase): # Now fetch the remaining tokens try: consumer_secret, access_token_key, access_token_secret = \ - filter(bool, NotifyBase.split_path(results['fullpath']))[0:3] + [x for x in filter(bool, NotifyBase.split_path( + results['fullpath']))][0:3] - except (AttributeError, IndexError): + except (ValueError, AttributeError, IndexError): # Force some bad values that will get caught # in parsing later consumer_secret = None access_token_key = None access_token_secret = None - results['ckey'] = consumer_key, - results['csecret'] = consumer_secret, - results['akey'] = access_token_key, - results['asecret'] = access_token_secret, + results['ckey'] = consumer_key + results['csecret'] = consumer_secret + results['akey'] = access_token_key + results['asecret'] = access_token_secret return results diff --git a/apprise/plugins/__init__.py b/apprise/plugins/__init__.py index b730fa85..b0213a56 100644 --- a/apprise/plugins/__init__.py +++ b/apprise/plugins/__init__.py @@ -26,21 +26,23 @@ from .NotifyFaast import NotifyFaast from .NotifyGrowl.NotifyGrowl import NotifyGrowl from .NotifyGrowl import gntp from .NotifyJSON import NotifyJSON +from .NotifyMatterMost import NotifyMatterMost from .NotifyMyAndroid import NotifyMyAndroid from .NotifyProwl import NotifyProwl from .NotifyPushalot import NotifyPushalot from .NotifyPushBullet import NotifyPushBullet +from .NotifyPushjet.NotifyPushjet import NotifyPushjet +from .NotifyPushjet import pushjet from .NotifyPushover import NotifyPushover from .NotifyRocketChat import NotifyRocketChat +from .NotifyTelegram import NotifyTelegram from .NotifyToasty import NotifyToasty from .NotifyTwitter.NotifyTwitter import NotifyTwitter +from .NotifyTwitter import tweepy from .NotifyXBMC import NotifyXBMC from .NotifyXML import NotifyXML from .NotifySlack import NotifySlack from .NotifyJoin import NotifyJoin -from .NotifyTelegram import NotifyTelegram -from .NotifyMatterMost import NotifyMatterMost -from .NotifyPushjet.NotifyPushjet import NotifyPushjet from ..common import NotifyImageSize from ..common import NOTIFY_IMAGE_SIZES @@ -61,6 +63,12 @@ __all__ = [ # NotifyEmail Base References (used for Testing) 'NotifyEmailBase', - # gntp (used for Testing) + # gntp (used for NotifyGrowl Testing) 'gntp', + + # pushjet (used for NotifyPushjet Testing) + 'pushjet', + + # tweepy (used for NotifyTwitter Testing) + 'tweepy', ] diff --git a/test/test_pushjet_plugin.py b/test/test_pushjet_plugin.py new file mode 100644 index 00000000..5a23d640 --- /dev/null +++ b/test/test_pushjet_plugin.py @@ -0,0 +1,157 @@ +# -*- coding: utf-8 -*- +# +# NotifyPushjet - Unit Tests +# +# Copyright (C) 2017 Chris Caron +# +# This file is part of apprise. +# +# This program is free software; you can redistribute it and/or modify it +# under the terms of the GNU Lesser General Public License as published by +# the Free Software Foundation; either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Lesser General Public License for more details. + +from apprise import plugins +from apprise import NotifyType +from apprise import Apprise +import mock + +TEST_URLS = ( + ################################## + # NotifyPushjet + ################################## + ('pjet://', { + 'instance': None, + }), + ('pjets://', { + 'instance': None, + }), + # Default query (uses pushjet server) + ('pjet://%s' % ('a' * 32), { + 'instance': plugins.NotifyPushjet, + }), + # Specify your own server + ('pjet://%s@localhost' % ('a' * 32), { + 'instance': plugins.NotifyPushjet, + }), + # Specify your own server with port + ('pjets://%s@localhost:8080' % ('a' * 32), { + 'instance': plugins.NotifyPushjet, + }), + ('pjet://:@/', { + 'instance': None, + }), + ('pjet://%s@localhost:8081' % ('a' * 32), { + 'instance': plugins.NotifyPushjet, + # Throws a series of connection and transfer exceptions when this flag + # is set and tests that we gracfully handle them + 'test_notify_exceptions': True, + }), +) + + +@mock.patch('apprise.plugins.pushjet.pushjet.Service.send') +@mock.patch('apprise.plugins.pushjet.pushjet.Service.refresh') +def test_plugin(mock_refresh, mock_send): + """ + API: NotifyPushjet Plugin() (pt1) + + """ + + # iterate over our dictionary and test it out + for (url, meta) in TEST_URLS: + + # Our expected instance + instance = meta.get('instance', None) + + # Our expected server objects + self = meta.get('self', None) + + # Our expected Query response (True, False, or exception type) + response = meta.get('response', True) + + # Allow us to force the server response code to be something other then + # the defaults + response = meta.get( + 'response', True if response else False) + + test_notify_exceptions = meta.get( + 'test_notify_exceptions', False) + + test_exceptions = ( + plugins.pushjet.errors.AccessError( + 0, 'pushjet.AccessError() not handled'), + plugins.pushjet.errors.NonexistentError( + 0, 'pushjet.NonexistentError() not handled'), + plugins.pushjet.errors.SubscriptionError( + 0, 'gntp.SubscriptionError() not handled'), + plugins.pushjet.errors.RequestError( + 'pushjet.RequestError() not handled'), + ) + + try: + obj = Apprise.instantiate(url, suppress_exceptions=False) + + if instance is None: + # Check that we got what we came for + assert obj is instance + continue + + assert(isinstance(obj, instance)) + + if self: + # Iterate over our expected entries inside of our object + for key, val in self.items(): + # Test that our object has the desired key + assert(hasattr(key, obj)) + assert(getattr(key, obj) == val) + + try: + if test_notify_exceptions is False: + # Store our response + mock_send.return_value = response + mock_send.side_effect = None + + # check that we're as expected + assert obj.notify( + title='test', body='body', + notify_type=NotifyType.INFO) == response + + else: + for exception in test_exceptions: + mock_send.side_effect = exception + mock_send.return_value = None + try: + assert obj.notify( + title='test', body='body', + notify_type=NotifyType.INFO) is False + + except AssertionError: + # Don't mess with these entries + raise + + except Exception as e: + # We can't handle this exception type + assert False + + except AssertionError: + # Don't mess with these entries + raise + + except Exception as e: + # Check that we were expecting this exception to happen + assert isinstance(e, response) + + except AssertionError: + # Don't mess with these entries + raise + + except Exception as e: + # Handle our exception + assert(instance is not None) + assert(isinstance(e, instance)) diff --git a/test/test_rest_plugins.py b/test/test_rest_plugins.py index f239f9a2..d11ee2a5 100644 --- a/test/test_rest_plugins.py +++ b/test/test_rest_plugins.py @@ -20,6 +20,7 @@ from apprise import plugins from apprise import NotifyType from apprise import Apprise from apprise import AppriseAsset +from apprise.utils import compat_is_basestring from json import dumps import requests import mock @@ -37,9 +38,8 @@ TEST_URLS = ( }), # An invalid access and secret key specified ('boxcar://access.key/secret.key/', { - 'instance': plugins.NotifyBoxcar, # Thrown because there were no recipients specified - 'exception': TypeError, + 'instance': TypeError, }), # Provide both an access and a secret ('boxcar://%s/%s' % ('a' * 64, 'b' * 64), { @@ -136,9 +136,8 @@ TEST_URLS = ( }), # Invalid APIKey ('join://%s' % ('a' * 24), { - 'instance': None, # Missing a channel - 'exception': TypeError, + 'instance': TypeError, }), # APIKey + device ('join://%s/%s' % ('a' * 32, 'd' * 32), { @@ -336,14 +335,12 @@ TEST_URLS = ( 'instance': plugins.NotifyMatterMost, }), ('mmosts://localhost', { - 'instance': plugins.NotifyMatterMost, # Thrown because there was no webhook id specified - 'exception': TypeError, + 'instance': TypeError, }), ('mmost://localhost/bad-web-hook', { - 'instance': plugins.NotifyMatterMost, # Thrown because the webhook is not in a valid format - 'exception': TypeError, + 'instance': TypeError, }), ('mmost://:@/', { 'instance': None, @@ -379,7 +376,7 @@ TEST_URLS = ( }), # Invalid APIKey ('nma://%s' % ('a' * 24), { - 'exception': TypeError, + 'instance': TypeError, }), # APIKey ('nma://%s' % ('a' * 48), { @@ -401,7 +398,7 @@ TEST_URLS = ( }), # APIKey + Invalid DevAPI Key ('nma://%s/%s' % ('a' * 48, 'b' * 24), { - 'exception': TypeError, + 'instance': TypeError, }), # APIKey + DevAPI Key ('nma://%s/%s' % ('a' * 48, 'b' * 48), { @@ -462,7 +459,7 @@ TEST_URLS = ( }), # Invalid APIKey ('prowl://%s' % ('a' * 24), { - 'exception': TypeError, + 'instance': TypeError, }), # APIKey ('prowl://%s' % ('a' * 40), { @@ -484,7 +481,7 @@ TEST_URLS = ( }), # APIKey + Invalid Provider Key ('prowl://%s/%s' % ('a' * 40, 'b' * 24), { - 'exception': TypeError, + 'instance': TypeError, }), # APIKey + No Provider Key (empty) ('prowl://%s///' % ('a' * 40), { @@ -539,9 +536,8 @@ TEST_URLS = ( }), # Invalid AuthToken ('palot://%s' % ('a' * 24), { - 'instance': None, # Missing a channel - 'exception': TypeError, + 'instance': TypeError, }), # AuthToken + bad url ('palot://:@/', { @@ -635,15 +631,15 @@ TEST_URLS = ( }), # APIkey; no user ('pover://%s' % ('a' * 30), { - 'exception': TypeError, + 'instance': TypeError, }), # APIkey; invalid user ('pover://%s@%s' % ('u' * 20, 'a' * 30), { - 'exception': TypeError, + 'instance': TypeError, }), # Invalid APIKey; valid User ('pover://%s@%s' % ('u' * 30, 'a' * 24), { - 'exception': TypeError, + 'instance': TypeError, }), # APIKey + Valid User ('pover://%s@%s' % ('u' * 30, 'a' * 30), { @@ -706,6 +702,122 @@ TEST_URLS = ( 'test_requests_exceptions': True, }), + ################################## + # NotifyRocketChat + ################################## + ('rocket://', { + 'instance': None, + }), + ('rockets://', { + 'instance': None, + }), + # No username or pass + ('rocket://localhost', { + 'instance': TypeError, + }), + # No room or channel + ('rocket://user:pass@localhost', { + 'instance': TypeError, + }), + # No valid rooms or channels + ('rocket://user:pass@localhost/#/!/@', { + 'instance': TypeError, + }), + # A room and port identifier + ('rocket://user:pass@localhost:8080/room/', { + 'instance': plugins.NotifyRocketChat, + # The response text is expected to be the following on a success + 'requests_response_text': { + 'status': 'success', + 'data': { + 'authToken': 'abcd', + 'userId': 'user', + }, + }, + }), + # A channel + ('rockets://user:pass@localhost/#channel', { + 'instance': plugins.NotifyRocketChat, + # The response text is expected to be the following on a success + 'requests_response_text': { + 'status': 'success', + 'data': { + 'authToken': 'abcd', + 'userId': 'user', + }, + }, + }), + # Several channels + ('rocket://user:pass@localhost/#channel1/#channel2/', { + 'instance': plugins.NotifyRocketChat, + # The response text is expected to be the following on a success + 'requests_response_text': { + 'status': 'success', + 'data': { + 'authToken': 'abcd', + 'userId': 'user', + }, + }, + }), + # Several Rooms + ('rocket://user:pass@localhost/room1/room2', { + 'instance': plugins.NotifyRocketChat, + # The response text is expected to be the following on a success + 'requests_response_text': { + 'status': 'success', + 'data': { + 'authToken': 'abcd', + 'userId': 'user', + }, + }, + }), + # A room and channel + ('rocket://user:pass@localhost/room/#channel', { + 'instance': plugins.NotifyRocketChat, + # The response text is expected to be the following on a success + 'requests_response_text': { + 'status': 'success', + 'data': { + 'authToken': 'abcd', + 'userId': 'user', + }, + }, + }), + ('rocket://:@/', { + 'instance': None, + }), + # A room and channel + ('rockets://user:pass@localhost/rooma/#channela', { + # The response text is expected to be the following on a success + 'requests_response_code': requests.codes.ok, + 'requests_response_text': { + # return something other then a success message type + 'status': 'failure', + }, + # Exception is thrown in this case + 'instance': plugins.NotifyRocketChat, + # Notifications will fail in this event + 'response': False, + }), + ('rocket://user:pass@localhost:8081/room1/room2', { + 'instance': plugins.NotifyRocketChat, + # force a failure + 'response': False, + 'requests_response_code': requests.codes.internal_server_error, + }), + ('rocket://user:pass@localhost:8082/#channel', { + 'instance': plugins.NotifyRocketChat, + # throw a bizzare code forcing us to fail to look it up + 'response': False, + 'requests_response_code': 999, + }), + ('rocket://user:pass@localhost:8083/#chan1/#chan2/room', { + 'instance': plugins.NotifyRocketChat, + # Throws a series of connection and transfer exceptions when this flag + # is set and tests that we gracfully handle them + 'test_requests_exceptions': True, + }), + ################################## # NotifySlack ################################## @@ -741,19 +853,19 @@ TEST_URLS = ( }), ('slack://username@T1JJ3T3L2/A1BRTD4JD/TIiajkdnlazkcOXrIdevi7FQ', { # Missing a channel - 'exception': TypeError, + 'instance': TypeError, }), ('slack://username@INVALID/A1BRTD4JD/TIiajkdnlazkcOXrIdevi7FQ/#cool', { # invalid 1st Token - 'exception': TypeError, + 'instance': TypeError, }), ('slack://username@T1JJ3T3L2/INVALID/TIiajkdnlazkcOXrIdevi7FQ/#great', { # invalid 2rd Token - 'exception': TypeError, + 'instance': TypeError, }), ('slack://username@T1JJ3T3L2/A1BRTD4JD/INVALID/#channel', { # invalid 3rd Token - 'exception': TypeError, + 'instance': TypeError, }), ('slack://l2g@T1JJ3T3L2/A1BRTD4JD/TIiajkdnlazkcOXrIdevi7FQ/#usenet', { 'instance': plugins.NotifySlack, @@ -906,7 +1018,7 @@ TEST_URLS = ( }), # No username specified but contains a device ('toasty://%s' % ('d' * 32), { - 'exception': TypeError, + 'instance': TypeError, }), # User + 1 device ('toasty://user@device', { @@ -1067,9 +1179,6 @@ def test_rest_plugins(mock_post, mock_get): # Our expected instance instance = meta.get('instance', None) - # Our expected exception - exception = meta.get('exception', None) - # Our expected server objects self = meta.get('self', None) @@ -1083,6 +1192,13 @@ def test_rest_plugins(mock_post, mock_get): requests.codes.ok if response else requests.codes.not_found, ) + # Allow us to force the server response text to be something other then + # the defaults + requests_response_text = meta.get('requests_response_text') + if not compat_is_basestring(requests_response_text): + # Convert to string + requests_response_text = dumps(requests_response_text) + # Allow notification type override, otherwise default to INFO notify_type = meta.get('notify_type', NotifyType.INFO) @@ -1112,6 +1228,12 @@ def test_rest_plugins(mock_post, mock_get): # Handle our default response mock_post.return_value.status_code = requests_response_code mock_get.return_value.status_code = requests_response_code + + # Handle our default text response + mock_get.return_value.text = requests_response_text + mock_post.return_value.text = requests_response_text + + # Ensure there is no side effect set mock_post.side_effect = None mock_get.side_effect = None @@ -1135,20 +1257,11 @@ def test_rest_plugins(mock_post, mock_get): obj = Apprise.instantiate( url, asset=asset, suppress_exceptions=False) - # Make sure we weren't expecting an exception and just didn't get - # one. - assert exception is None - if obj is None: # We're done (assuming this is what we were expecting) assert instance is None continue - if instance is None: - # Expected None but didn't get it - print('%s instantiated %s' % (url, str(obj))) - assert(False) - assert(isinstance(obj, instance)) # Disable throttling to speed up unit tests @@ -1172,6 +1285,7 @@ def test_rest_plugins(mock_post, mock_get): for _exception in test_requests_exceptions: mock_post.side_effect = _exception mock_get.side_effect = _exception + try: assert obj.notify( title='test', body='body', @@ -1203,8 +1317,8 @@ def test_rest_plugins(mock_post, mock_get): except Exception as e: # Handle our exception print('%s / %s' % (url, str(e))) - assert(exception is not None) - assert(isinstance(e, exception)) + assert(instance is not None) + assert(isinstance(e, instance)) @mock.patch('requests.get') @@ -1466,6 +1580,133 @@ def test_notify_pushover_plugin(mock_post, mock_get): assert(plugins.NotifyPushover.parse_url(42) is None) +@mock.patch('requests.get') +@mock.patch('requests.post') +def test_notify_rocketchat_plugin(mock_post, mock_get): + """ + API: NotifyRocketChat() Extra Checks + + """ + # Chat ID + recipients = 'l2g, lead2gold, #channel, #channel2' + + # Prepare Mock + mock_get.return_value = requests.Request() + mock_post.return_value = requests.Request() + mock_post.return_value.status_code = requests.codes.ok + mock_get.return_value.status_code = requests.codes.ok + mock_post.return_value.text = '' + mock_get.return_value.text = '' + + try: + obj = plugins.NotifyRocketChat(recipients=None) + # invalid recipients list (None) + assert(False) + + except TypeError: + # Exception should be thrown about the fact no recipients were + # specified + assert(True) + + try: + obj = plugins.NotifyRocketChat(recipients=object()) + # invalid recipients list (object) + assert(False) + + except TypeError: + # Exception should be thrown about the fact no recipients were + # specified + assert(True) + + try: + obj = plugins.NotifyRocketChat(recipients=set()) + # invalid recipient list/set (no entries) + assert(False) + + except TypeError: + # Exception should be thrown about the fact no recipients were + # specified + assert(True) + + obj = plugins.NotifyRocketChat(recipients=recipients) + assert(isinstance(obj, plugins.NotifyRocketChat)) + assert(len(obj.channels) == 2) + assert(len(obj.rooms) == 2) + + # Disable throttling to speed up unit tests + obj.throttle_attempt = 0 + + # + # Logout + # + assert obj.logout() is True + + # Support the handling of an empty and invalid URL strings + assert plugins.NotifyRocketChat.parse_url(None) is None + assert plugins.NotifyRocketChat.parse_url('') is None + assert plugins.NotifyRocketChat.parse_url(42) is None + + # Prepare Mock to fail + mock_post.return_value.status_code = requests.codes.internal_server_error + mock_get.return_value.status_code = requests.codes.internal_server_error + mock_post.return_value.text = '' + mock_get.return_value.text = '' + + # + # Send Notification + # + assert obj.notify( + title='title', body='body', notify_type=NotifyType.INFO) is False + assert obj.send_notification( + payload='test', notify_type=NotifyType.INFO) is False + + # + # Logout + # + assert obj.logout() is False + + # KeyError handling + mock_post.return_value.status_code = 999 + mock_get.return_value.status_code = 999 + + # + # Send Notification + # + assert obj.notify( + title='title', body='body', notify_type=NotifyType.INFO) is False + assert obj.send_notification( + payload='test', notify_type=NotifyType.INFO) is False + + # + # Logout + # + assert obj.logout() is False + + mock_post.return_value.text = '' + # Generate exceptions + mock_get.side_effect = requests.ConnectionError( + 0, 'requests.ConnectionError() not handled') + mock_post.side_effect = mock_get.side_effect + mock_get.return_value.text = '' + mock_post.return_value.text = '' + + # + # Send Notification + # + assert obj.send_notification( + payload='test', notify_type=NotifyType.INFO) is False + + # Attempt the check again but fake a successful login + obj.login = mock.Mock() + obj.login.return_value = True + assert obj.notify( + title='title', body='body', notify_type=NotifyType.INFO) is False + # + # Logout + # + assert obj.logout() is False + + @mock.patch('requests.get') @mock.patch('requests.post') def test_notify_toasty_plugin(mock_post, mock_get): diff --git a/test/test_twitter_plugin.py b/test/test_twitter_plugin.py new file mode 100644 index 00000000..d4973d66 --- /dev/null +++ b/test/test_twitter_plugin.py @@ -0,0 +1,189 @@ +# -*- coding: utf-8 -*- +# +# NotifyTwitter - Unit Tests +# +# Copyright (C) 2017 Chris Caron +# +# This file is part of apprise. +# +# This program is free software; you can redistribute it and/or modify it +# under the terms of the GNU Lesser General Public License as published by +# the Free Software Foundation; either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Lesser General Public License for more details. + +from apprise import plugins +from apprise import NotifyType +from apprise import Apprise +import mock + + +TEST_URLS = ( + ################################## + # NotifyPushjet + ################################## + ('tweet://', { + 'instance': None, + }), + ('tweet://consumer_key', { + # Missing Keys + 'instance': TypeError, + }), + ('tweet://consumer_key/consumer_key/', { + # Missing Keys + 'instance': TypeError, + }), + ('tweet://consumer_key/consumer_key/access_token/', { + # Missing Access Secret + 'instance': TypeError, + }), + ('tweet://consumer_key/consumer_key/access_token/access_secret', { + # Missing User + 'instance': TypeError, + }), + ('tweet://user@consumer_key/consumer_key/access_token/access_secret', { + # We're good! + 'instance': plugins.NotifyTwitter, + }), + ('tweet://:@/', { + 'instance': None, + }), +) + + +@mock.patch('apprise.plugins.tweepy.API') +@mock.patch('apprise.plugins.tweepy.OAuthHandler') +def test_plugin(mock_oauth, mock_api): + """ + API: NotifyTwitter Plugin() (pt1) + + """ + + # iterate over our dictionary and test it out + for (url, meta) in TEST_URLS: + + # Our expected instance + instance = meta.get('instance', None) + + # Our expected server objects + self = meta.get('self', None) + + # Our expected Query response (True, False, or exception type) + response = meta.get('response', True) + + # Allow us to force the server response code to be something other then + # the defaults + response = meta.get( + 'response', True if response else False) + + try: + obj = Apprise.instantiate(url, suppress_exceptions=False) + + if instance is None: + # Check that we got what we came for + assert obj is instance + continue + + assert(isinstance(obj, instance)) + + if self: + # Iterate over our expected entries inside of our object + for key, val in self.items(): + # Test that our object has the desired key + assert(hasattr(key, obj)) + assert(getattr(key, obj) == val) + + # check that we're as expected + assert obj.notify( + title='test', body='body', + notify_type=NotifyType.INFO) == response + + except AssertionError: + # Don't mess with these entries + raise + + except Exception as e: + # Handle our exception + assert(instance is not None) + assert(isinstance(e, instance)) + + +@mock.patch('apprise.plugins.tweepy.API.send_direct_message') +@mock.patch('apprise.plugins.tweepy.OAuthHandler.set_access_token') +def test_twitter_plugin_init(set_access_token, send_direct_message): + """ + API: NotifyTwitter Plugin() (pt2) + + """ + + try: + plugins.NotifyTwitter( + ckey=None, csecret=None, akey=None, asecret=None) + assert False + except TypeError: + # All keys set to none + assert True + + try: + plugins.NotifyTwitter( + ckey='value', csecret=None, akey=None, asecret=None) + assert False + except TypeError: + # csecret not set + assert True + + try: + plugins.NotifyTwitter( + ckey='value', csecret='value', akey=None, asecret=None) + assert False + except TypeError: + # akey not set + assert True + + try: + plugins.NotifyTwitter( + ckey='value', csecret='value', akey='value', asecret=None) + assert False + except TypeError: + # asecret not set + assert True + + try: + plugins.NotifyTwitter( + ckey='value', csecret='value', akey='value', asecret='value') + assert False + except TypeError: + # user not set + assert True + + try: + obj = plugins.NotifyTwitter( + ckey='value', csecret='value', akey='value', asecret='value', + user='l2g') + # We should initialize properly + assert True + + except TypeError: + # We should not reach here + assert False + + set_access_token.side_effect = TypeError('Invalid') + + assert obj.notify( + title='test', body='body', + notify_type=NotifyType.INFO) is False + + # Make it so we can pass authentication, but fail on message + # delivery + set_access_token.side_effect = None + set_access_token.return_value = True + send_direct_message.side_effect = plugins.tweepy.error.TweepError( + 0, 'pushjet.TweepyError() not handled'), + + assert obj.notify( + title='test', body='body', + notify_type=NotifyType.INFO) is False