From 9a21de2e56f9d2a39595f3625cad1f0faacd0d41 Mon Sep 17 00:00:00 2001 From: Chris Caron Date: Thu, 9 Jun 2022 17:54:57 -0400 Subject: [PATCH] Refactored token parsing for YAML config (#599) --- apprise/config/ConfigBase.py | 11 ++- apprise/plugins/NotifyDBus.py | 116 +++++++++++++++++------------- apprise/plugins/NotifyDapnet.py | 75 +++++++++---------- apprise/plugins/NotifyGnome.py | 83 ++++++++++++--------- apprise/plugins/NotifyGotify.py | 73 ++++++++++++------- apprise/plugins/NotifyGrowl.py | 88 +++++++++++------------ apprise/plugins/NotifyJoin.py | 76 ++++++++++---------- apprise/plugins/NotifyNtfy.py | 79 ++++++++++---------- apprise/plugins/NotifyOpsgenie.py | 99 ++++++++++++------------- apprise/plugins/NotifyProwl.py | 90 +++++++++++------------ apprise/plugins/NotifyPushover.py | 80 ++++++++++----------- test/test_apprise_config.py | 6 +- test/test_plugin_dapnet.py | 72 +++++++++++++++++++ test/test_plugin_glib.py | 80 +++++++++++++++++++-- test/test_plugin_gnome.py | 70 ++++++++++++++++++ test/test_plugin_gotify.py | 73 +++++++++++++++++++ test/test_plugin_growl.py | 70 ++++++++++++++++++ test/test_plugin_join.py | 72 +++++++++++++++++++ test/test_plugin_ntfy.py | 97 +++++++++++++++++++++---- test/test_plugin_opsgenie.py | 112 ++++++++++++++++++++++++----- test/test_plugin_prowl.py | 70 ++++++++++++++++++ test/test_plugin_pushover.py | 111 ++++++++++++++++++++++------ 22 files changed, 1226 insertions(+), 477 deletions(-) diff --git a/apprise/config/ConfigBase.py b/apprise/config/ConfigBase.py index f2b958ed..d5136faf 100644 --- a/apprise/config/ConfigBase.py +++ b/apprise/config/ConfigBase.py @@ -927,6 +927,14 @@ class ConfigBase(URLBase): # Grab our first item _results = results.pop(0) + if _results['schema'] not in plugins.SCHEMA_MAP: + # the arguments are invalid or can not be used. + ConfigBase.logger.warning( + 'An invalid Apprise schema ({}) in YAML configuration ' + 'entry #{}, item #{}' + .format(_results['schema'], no + 1, entry)) + continue + # tag is a special keyword that is managed by Apprise object. # The below ensures our tags are set correctly if 'tag' in _results: @@ -958,6 +966,7 @@ class ConfigBase(URLBase): # Prepare our Asset Object _results['asset'] = asset + # Now we generate our plugin try: # Attempt to create an instance of our plugin using the # parsed URL information @@ -1088,7 +1097,7 @@ class ConfigBase(URLBase): # Detect if we're dealign with a list or not is_list = re.search( - r'^(list|choice):.*', + r'^list:.*', meta.get('type'), re.IGNORECASE) diff --git a/apprise/plugins/NotifyDBus.py b/apprise/plugins/NotifyDBus.py index 145e1c05..cdae2da3 100644 --- a/apprise/plugins/NotifyDBus.py +++ b/apprise/plugins/NotifyDBus.py @@ -115,12 +115,30 @@ class DBusUrgency(object): HIGH = 2 -# Define our urgency levels -DBUS_URGENCIES = ( - DBusUrgency.LOW, - DBusUrgency.NORMAL, - DBusUrgency.HIGH, -) +DBUS_URGENCIES = { + # Note: This also acts as a reverse lookup mapping + DBusUrgency.LOW: 'low', + DBusUrgency.NORMAL: 'normal', + DBusUrgency.HIGH: 'high', +} + +DBUS_URGENCY_MAP = { + # Maps against string 'low' + 'l': DBusUrgency.LOW, + # Maps against string 'moderate' + 'm': DBusUrgency.LOW, + # Maps against string 'normal' + 'n': DBusUrgency.NORMAL, + # Maps against string 'high' + 'h': DBusUrgency.HIGH, + # Maps against string 'emergency' + 'e': DBusUrgency.HIGH, + + # Entries to additionally support (so more like DBus's API) + '0': DBusUrgency.LOW, + '1': DBusUrgency.NORMAL, + '2': DBusUrgency.HIGH, +} class NotifyDBus(NotifyBase): @@ -182,6 +200,12 @@ class NotifyDBus(NotifyBase): 'values': DBUS_URGENCIES, 'default': DBusUrgency.NORMAL, }, + 'priority': { + # Apprise uses 'priority' everywhere; it's just a nice consistent + # feel to be able to use it here as well. Just map the + # value back to 'priority' + 'alias_of': 'urgency', + }, 'x': { 'name': _('X-Axis'), 'type': 'int', @@ -223,15 +247,29 @@ class NotifyDBus(NotifyBase): raise TypeError(msg) # The urgency of the message - if urgency not in DBUS_URGENCIES: - self.urgency = DBusUrgency.NORMAL - - else: - self.urgency = urgency + self.urgency = int( + NotifyDBus.template_args['urgency']['default'] + if urgency is None else + next(( + v for k, v in DBUS_URGENCY_MAP.items() + if str(urgency).lower().startswith(k)), + NotifyDBus.template_args['urgency']['default'])) # Our x/y axis settings - self.x_axis = x_axis if isinstance(x_axis, int) else None - self.y_axis = y_axis if isinstance(y_axis, int) else None + if x_axis or y_axis: + try: + self.x_axis = int(x_axis) + self.y_axis = int(y_axis) + + except (TypeError, ValueError): + # Invalid x/y values specified + msg = 'The x,y coordinates specified ({},{}) are invalid.'\ + .format(x_axis, y_axis) + self.logger.warning(msg) + raise TypeError(msg) + else: + self.x_axis = None + self.y_axis = None # Track whether or not we want to send an image with our notification # or not. @@ -343,17 +381,13 @@ class NotifyDBus(NotifyBase): Returns the URL built dynamically based on specified arguments. """ - _map = { - DBusUrgency.LOW: 'low', - DBusUrgency.NORMAL: 'normal', - DBusUrgency.HIGH: 'high', - } - # Define any URL parameters params = { 'image': 'yes' if self.include_image else 'no', - 'urgency': 'normal' if self.urgency not in _map - else _map[self.urgency], + 'urgency': + DBUS_URGENCIES[self.template_args['urgency']['default']] + if self.urgency not in DBUS_URGENCIES + else DBUS_URGENCIES[self.urgency], } # Extend our parameters @@ -389,38 +423,20 @@ class NotifyDBus(NotifyBase): # DBus supports urgency, but we we also support the keyword priority # so that it is consistent with some of the other plugins - urgency = results['qsd'].get('urgency', results['qsd'].get('priority')) - if urgency and len(urgency): - _map = { - '0': DBusUrgency.LOW, - 'l': DBusUrgency.LOW, - 'n': DBusUrgency.NORMAL, - '1': DBusUrgency.NORMAL, - 'h': DBusUrgency.HIGH, - '2': DBusUrgency.HIGH, - } + if 'priority' in results['qsd'] and len(results['qsd']['priority']): + # We intentionally store the priority in the urgency section + results['urgency'] = \ + NotifyDBus.unquote(results['qsd']['priority']) - try: - # Attempt to index/retrieve our urgency - results['urgency'] = _map[urgency[0].lower()] - - except KeyError: - # No priority was set - pass + if 'urgency' in results['qsd'] and len(results['qsd']['urgency']): + results['urgency'] = \ + NotifyDBus.unquote(results['qsd']['urgency']) # handle x,y coordinates - try: - results['x_axis'] = int(results['qsd'].get('x')) + if 'x' in results['qsd'] and len(results['qsd']['x']): + results['x_axis'] = NotifyDBus.unquote(results['qsd'].get('x')) - except (TypeError, ValueError): - # No x was set - pass - - try: - results['y_axis'] = int(results['qsd'].get('y')) - - except (TypeError, ValueError): - # No y was set - pass + if 'y' in results['qsd'] and len(results['qsd']['y']): + results['y_axis'] = NotifyDBus.unquote(results['qsd'].get('y')) return results diff --git a/apprise/plugins/NotifyDapnet.py b/apprise/plugins/NotifyDapnet.py index 2e0389db..8fd4a229 100644 --- a/apprise/plugins/NotifyDapnet.py +++ b/apprise/plugins/NotifyDapnet.py @@ -63,10 +63,22 @@ class DapnetPriority(object): EMERGENCY = 1 -DAPNET_PRIORITIES = ( - DapnetPriority.NORMAL, - DapnetPriority.EMERGENCY, -) +DAPNET_PRIORITIES = { + DapnetPriority.NORMAL: 'normal', + DapnetPriority.EMERGENCY: 'emergency', +} + + +DAPNET_PRIORITY_MAP = { + # Maps against string 'normal' + 'n': DapnetPriority.NORMAL, + # Maps against string 'emergency' + 'e': DapnetPriority.EMERGENCY, + + # Entries to additionally support (so more like Dapnet's API) + '0': DapnetPriority.NORMAL, + '1': DapnetPriority.EMERGENCY, +} class NotifyDapnet(NotifyBase): @@ -172,11 +184,14 @@ class NotifyDapnet(NotifyBase): # Parse our targets self.targets = list() - # get the emergency prio setting - if priority not in DAPNET_PRIORITIES: - self.priority = self.template_args['priority']['default'] - else: - self.priority = priority + # The Priority of the message + self.priority = int( + NotifyDapnet.template_args['priority']['default'] + if priority is None else + next(( + v for k, v in DAPNET_PRIORITY_MAP.items() + if str(priority).lower().startswith(k)), + NotifyDapnet.template_args['priority']['default'])) if not (self.user and self.password): msg = 'A Dapnet user/pass was not provided.' @@ -201,8 +216,7 @@ class NotifyDapnet(NotifyBase): ) continue - # Store callsign without SSID and - # ignore duplicates + # Store callsign without SSID and ignore duplicates if result['callsign'] not in self.targets: self.targets.append(result['callsign']) @@ -230,10 +244,6 @@ class NotifyDapnet(NotifyBase): # error tracking (used for function return) has_error = False - # prepare the emergency mode - emergency_mode = True \ - if self.priority == DapnetPriority.EMERGENCY else False - # Create a copy of the targets list targets = list(self.targets) @@ -244,7 +254,7 @@ class NotifyDapnet(NotifyBase): 'text': body, 'callSignNames': targets[index:index + batch_size], 'transmitterGroupNames': self.txgroups, - 'emergency': emergency_mode, + 'emergency': (self.priority == DapnetPriority.EMERGENCY), } self.logger.debug('DAPNET POST URL: %s' % self.notify_url) @@ -304,16 +314,12 @@ class NotifyDapnet(NotifyBase): Returns the URL built dynamically based on specified arguments. """ - # Define any URL parameters - _map = { - DapnetPriority.NORMAL: 'normal', - DapnetPriority.EMERGENCY: 'emergency', - } - # Define any URL parameters params = { - 'priority': 'normal' if self.priority not in _map - else _map[self.priority], + 'priority': + DAPNET_PRIORITIES[self.template_args['priority']['default']] + if self.priority not in DAPNET_PRIORITIES + else DAPNET_PRIORITIES[self.priority], 'batch': 'yes' if self.batch else 'no', 'txgroups': ','.join(self.txgroups), } @@ -361,25 +367,10 @@ class NotifyDapnet(NotifyBase): results['targets'] += \ NotifyDapnet.parse_list(results['qsd']['to']) - # Check for priority + # Set our priority if 'priority' in results['qsd'] and len(results['qsd']['priority']): - _map = { - # Letter Assignments - 'n': DapnetPriority.NORMAL, - 'e': DapnetPriority.EMERGENCY, - 'no': DapnetPriority.NORMAL, - 'em': DapnetPriority.EMERGENCY, - # Numeric assignments - '0': DapnetPriority.NORMAL, - '1': DapnetPriority.EMERGENCY, - } - try: - results['priority'] = \ - _map[results['qsd']['priority'][0:2].lower()] - - except KeyError: - # No priority was set - pass + results['priority'] = \ + NotifyDapnet.unquote(results['qsd']['priority']) # Check for one or multiple transmitter groups (comma separated) # and split them up, when necessary diff --git a/apprise/plugins/NotifyGnome.py b/apprise/plugins/NotifyGnome.py index 6317c0d5..1a4ec149 100644 --- a/apprise/plugins/NotifyGnome.py +++ b/apprise/plugins/NotifyGnome.py @@ -66,11 +66,30 @@ class GnomeUrgency(object): HIGH = 2 -GNOME_URGENCIES = ( - GnomeUrgency.LOW, - GnomeUrgency.NORMAL, - GnomeUrgency.HIGH, -) +GNOME_URGENCIES = { + GnomeUrgency.LOW: 'low', + GnomeUrgency.NORMAL: 'normal', + GnomeUrgency.HIGH: 'high', +} + + +GNOME_URGENCY_MAP = { + # Maps against string 'low' + 'l': GnomeUrgency.LOW, + # Maps against string 'moderate' + 'm': GnomeUrgency.LOW, + # Maps against string 'normal' + 'n': GnomeUrgency.NORMAL, + # Maps against string 'high' + 'h': GnomeUrgency.HIGH, + # Maps against string 'emergency' + 'e': GnomeUrgency.HIGH, + + # Entries to additionally support (so more like Gnome's API) + '0': GnomeUrgency.LOW, + '1': GnomeUrgency.NORMAL, + '2': GnomeUrgency.HIGH, +} class NotifyGnome(NotifyBase): @@ -126,6 +145,12 @@ class NotifyGnome(NotifyBase): 'values': GNOME_URGENCIES, 'default': GnomeUrgency.NORMAL, }, + 'priority': { + # Apprise uses 'priority' everywhere; it's just a nice consistent + # feel to be able to use it here as well. Just map the + # value back to 'priority' + 'alias_of': 'urgency', + }, 'image': { 'name': _('Include Image'), 'type': 'bool', @@ -142,11 +167,13 @@ class NotifyGnome(NotifyBase): super(NotifyGnome, self).__init__(**kwargs) # The urgency of the message - if urgency not in GNOME_URGENCIES: - self.urgency = self.template_args['urgency']['default'] - - else: - self.urgency = urgency + self.urgency = int( + NotifyGnome.template_args['urgency']['default'] + if urgency is None else + next(( + v for k, v in GNOME_URGENCY_MAP.items() + if str(urgency).lower().startswith(k)), + NotifyGnome.template_args['urgency']['default'])) # Track whether or not we want to send an image with our notification # or not. @@ -205,17 +232,13 @@ class NotifyGnome(NotifyBase): Returns the URL built dynamically based on specified arguments. """ - _map = { - GnomeUrgency.LOW: 'low', - GnomeUrgency.NORMAL: 'normal', - GnomeUrgency.HIGH: 'high', - } - # Define any URL parameters params = { 'image': 'yes' if self.include_image else 'no', - 'urgency': 'normal' if self.urgency not in _map - else _map[self.urgency], + 'urgency': + GNOME_URGENCIES[self.template_args['urgency']['default']] + if self.urgency not in GNOME_URGENCIES + else GNOME_URGENCIES[self.urgency], } # Extend our parameters @@ -243,23 +266,13 @@ class NotifyGnome(NotifyBase): # Gnome supports urgency, but we we also support the keyword priority # so that it is consistent with some of the other plugins - urgency = results['qsd'].get('urgency', results['qsd'].get('priority')) - if urgency and len(urgency): - _map = { - '0': GnomeUrgency.LOW, - 'l': GnomeUrgency.LOW, - 'n': GnomeUrgency.NORMAL, - '1': GnomeUrgency.NORMAL, - 'h': GnomeUrgency.HIGH, - '2': GnomeUrgency.HIGH, - } + if 'priority' in results['qsd'] and len(results['qsd']['priority']): + # We intentionally store the priority in the urgency section + results['urgency'] = \ + NotifyGnome.unquote(results['qsd']['priority']) - try: - # Attempt to index/retrieve our urgency - results['urgency'] = _map[urgency[0].lower()] - - except KeyError: - # No priority was set - pass + if 'urgency' in results['qsd'] and len(results['qsd']['urgency']): + results['urgency'] = \ + NotifyGnome.unquote(results['qsd']['urgency']) return results diff --git a/apprise/plugins/NotifyGotify.py b/apprise/plugins/NotifyGotify.py index d064c07a..8941e40a 100644 --- a/apprise/plugins/NotifyGotify.py +++ b/apprise/plugins/NotifyGotify.py @@ -49,13 +49,37 @@ class GotifyPriority(object): EMERGENCY = 10 -GOTIFY_PRIORITIES = ( - GotifyPriority.LOW, - GotifyPriority.MODERATE, - GotifyPriority.NORMAL, - GotifyPriority.HIGH, - GotifyPriority.EMERGENCY, -) +GOTIFY_PRIORITIES = { + # Note: This also acts as a reverse lookup mapping + GotifyPriority.LOW: 'low', + GotifyPriority.MODERATE: 'moderate', + GotifyPriority.NORMAL: 'normal', + GotifyPriority.HIGH: 'high', + GotifyPriority.EMERGENCY: 'emergency', +} + +GOTIFY_PRIORITY_MAP = { + # Maps against string 'low' + 'l': GotifyPriority.LOW, + # Maps against string 'moderate' + 'm': GotifyPriority.MODERATE, + # Maps against string 'normal' + 'n': GotifyPriority.NORMAL, + # Maps against string 'high' + 'h': GotifyPriority.HIGH, + # Maps against string 'emergency' + 'e': GotifyPriority.EMERGENCY, + + # Entries to additionally support (so more like Gotify's API) + '10': GotifyPriority.EMERGENCY, + # ^ 10 needs to be checked before '1' below or it will match the wrong + # priority + '0': GotifyPriority.LOW, '1': GotifyPriority.LOW, '2': GotifyPriority.LOW, + '3': GotifyPriority.MODERATE, '4': GotifyPriority.MODERATE, + '5': GotifyPriority.NORMAL, '6': GotifyPriority.NORMAL, + '7': GotifyPriority.NORMAL, + '8': GotifyPriority.HIGH, '9': GotifyPriority.HIGH, +} class NotifyGotify(NotifyBase): @@ -144,11 +168,14 @@ class NotifyGotify(NotifyBase): # prepare our fullpath self.fullpath = kwargs.get('fullpath', '/') - if priority not in GOTIFY_PRIORITIES: - self.priority = GotifyPriority.NORMAL - - else: - self.priority = priority + # The Priority of the message + self.priority = int( + NotifyGotify.template_args['priority']['default'] + if priority is None else + next(( + v for k, v in GOTIFY_PRIORITY_MAP.items() + if str(priority).lower().startswith(k)), + NotifyGotify.template_args['priority']['default'])) if self.secure: self.schema = 'https' @@ -246,7 +273,10 @@ class NotifyGotify(NotifyBase): # Define any URL parameters params = { - 'priority': self.priority, + 'priority': + GOTIFY_PRIORITIES[self.template_args['priority']['default']] + if self.priority not in GOTIFY_PRIORITIES + else GOTIFY_PRIORITIES[self.priority], } # Extend our parameters @@ -294,20 +324,9 @@ class NotifyGotify(NotifyBase): results['fullpath'] = \ '/' if not entries else '/{}/'.format('/'.join(entries)) + # Set our priority if 'priority' in results['qsd'] and len(results['qsd']['priority']): - _map = { - 'l': GotifyPriority.LOW, - 'm': GotifyPriority.MODERATE, - 'n': GotifyPriority.NORMAL, - 'h': GotifyPriority.HIGH, - 'e': GotifyPriority.EMERGENCY, - } - try: - results['priority'] = \ - _map[results['qsd']['priority'][0].lower()] - - except KeyError: - # No priority was set - pass + results['priority'] = \ + NotifyGotify.unquote(results['qsd']['priority']) return results diff --git a/apprise/plugins/NotifyGrowl.py b/apprise/plugins/NotifyGrowl.py index 446ad660..76ac266a 100644 --- a/apprise/plugins/NotifyGrowl.py +++ b/apprise/plugins/NotifyGrowl.py @@ -54,13 +54,34 @@ class GrowlPriority(object): EMERGENCY = 2 -GROWL_PRIORITIES = ( - GrowlPriority.LOW, - GrowlPriority.MODERATE, - GrowlPriority.NORMAL, - GrowlPriority.HIGH, - GrowlPriority.EMERGENCY, -) +GROWL_PRIORITIES = { + # Note: This also acts as a reverse lookup mapping + GrowlPriority.LOW: 'low', + GrowlPriority.MODERATE: 'moderate', + GrowlPriority.NORMAL: 'normal', + GrowlPriority.HIGH: 'high', + GrowlPriority.EMERGENCY: 'emergency', +} + +GROWL_PRIORITY_MAP = { + # Maps against string 'low' + 'l': GrowlPriority.LOW, + # Maps against string 'moderate' + 'm': GrowlPriority.MODERATE, + # Maps against string 'normal' + 'n': GrowlPriority.NORMAL, + # Maps against string 'high' + 'h': GrowlPriority.HIGH, + # Maps against string 'emergency' + 'e': GrowlPriority.EMERGENCY, + + # Entries to additionally support (so more like Growl's API) + '-2': GrowlPriority.LOW, + '-1': GrowlPriority.MODERATE, + '0': GrowlPriority.NORMAL, + '1': GrowlPriority.HIGH, + '2': GrowlPriority.EMERGENCY, +} class NotifyGrowl(NotifyBase): @@ -172,11 +193,12 @@ class NotifyGrowl(NotifyBase): self.port = self.default_port # The Priority of the message - if priority not in GROWL_PRIORITIES: - self.priority = GrowlPriority.NORMAL - - else: - self.priority = priority + self.priority = NotifyGrowl.template_args['priority']['default'] \ + if not priority else \ + next(( + v for k, v in GROWL_PRIORITY_MAP.items() + if str(priority).lower().startswith(k)), + NotifyGrowl.template_args['priority']['default']) # Our Registered object self.growl = None @@ -318,21 +340,14 @@ class NotifyGrowl(NotifyBase): Returns the URL built dynamically based on specified arguments. """ - _map = { - GrowlPriority.LOW: 'low', - GrowlPriority.MODERATE: 'moderate', - GrowlPriority.NORMAL: 'normal', - GrowlPriority.HIGH: 'high', - GrowlPriority.EMERGENCY: 'emergency', - } - # Define any URL parameters params = { 'image': 'yes' if self.include_image else 'no', 'sticky': 'yes' if self.sticky else 'no', 'priority': - _map[GrowlPriority.NORMAL] if self.priority not in _map - else _map[self.priority], + GROWL_PRIORITIES[self.template_args['priority']['default']] + if self.priority not in GROWL_PRIORITIES + else GROWL_PRIORITIES[self.priority], 'version': self.version, } @@ -384,33 +399,10 @@ class NotifyGrowl(NotifyBase): ) pass + # Set our priority if 'priority' in results['qsd'] and len(results['qsd']['priority']): - _map = { - # Letter Assignments - 'l': GrowlPriority.LOW, - 'm': GrowlPriority.MODERATE, - 'n': GrowlPriority.NORMAL, - 'h': GrowlPriority.HIGH, - 'e': GrowlPriority.EMERGENCY, - 'lo': GrowlPriority.LOW, - 'me': GrowlPriority.MODERATE, - 'no': GrowlPriority.NORMAL, - 'hi': GrowlPriority.HIGH, - 'em': GrowlPriority.EMERGENCY, - # Support 3rd Party Documented Scale - '-2': GrowlPriority.LOW, - '-1': GrowlPriority.MODERATE, - '0': GrowlPriority.NORMAL, - '1': GrowlPriority.HIGH, - '2': GrowlPriority.EMERGENCY, - } - try: - results['priority'] = \ - _map[results['qsd']['priority'][0:2].lower()] - - except KeyError: - # No priority was set - pass + results['priority'] = \ + NotifyGrowl.unquote(results['qsd']['priority']) # Because of the URL formatting, the password is actually where the # username field is. For this reason, we just preform this small hack diff --git a/apprise/plugins/NotifyJoin.py b/apprise/plugins/NotifyJoin.py index d5e3f947..87731424 100644 --- a/apprise/plugins/NotifyJoin.py +++ b/apprise/plugins/NotifyJoin.py @@ -71,13 +71,34 @@ class JoinPriority(object): EMERGENCY = 2 -JOIN_PRIORITIES = ( - JoinPriority.LOW, - JoinPriority.MODERATE, - JoinPriority.NORMAL, - JoinPriority.HIGH, - JoinPriority.EMERGENCY, -) +JOIN_PRIORITIES = { + # Note: This also acts as a reverse lookup mapping + JoinPriority.LOW: 'low', + JoinPriority.MODERATE: 'moderate', + JoinPriority.NORMAL: 'normal', + JoinPriority.HIGH: 'high', + JoinPriority.EMERGENCY: 'emergency', +} + +JOIN_PRIORITY_MAP = { + # Maps against string 'low' + 'l': JoinPriority.LOW, + # Maps against string 'moderate' + 'm': JoinPriority.MODERATE, + # Maps against string 'normal' + 'n': JoinPriority.NORMAL, + # Maps against string 'high' + 'h': JoinPriority.HIGH, + # Maps against string 'emergency' + 'e': JoinPriority.EMERGENCY, + + # Entries to additionally support (so more like Join's API) + '-2': JoinPriority.LOW, + '-1': JoinPriority.MODERATE, + '0': JoinPriority.NORMAL, + '1': JoinPriority.HIGH, + '2': JoinPriority.EMERGENCY, +} class NotifyJoin(NotifyBase): @@ -189,11 +210,13 @@ class NotifyJoin(NotifyBase): raise TypeError(msg) # The Priority of the message - if priority not in JOIN_PRIORITIES: - self.priority = self.template_args['priority']['default'] - - else: - self.priority = priority + self.priority = int( + NotifyJoin.template_args['priority']['default'] + if priority is None else + next(( + v for k, v in JOIN_PRIORITY_MAP.items() + if str(priority).lower().startswith(k)), + NotifyJoin.template_args['priority']['default'])) # Prepare a list of targets to store entries into self.targets = list() @@ -324,19 +347,12 @@ class NotifyJoin(NotifyBase): """ Returns the URL built dynamically based on specified arguments. """ - _map = { - JoinPriority.LOW: 'low', - JoinPriority.MODERATE: 'moderate', - JoinPriority.NORMAL: 'normal', - JoinPriority.HIGH: 'high', - JoinPriority.EMERGENCY: 'emergency', - } - # Define any URL parameters params = { 'priority': - _map[self.template_args['priority']['default']] - if self.priority not in _map else _map[self.priority], + JOIN_PRIORITIES[self.template_args['priority']['default']] + if self.priority not in JOIN_PRIORITIES + else JOIN_PRIORITIES[self.priority], 'image': 'yes' if self.include_image else 'no', } @@ -371,20 +387,8 @@ class NotifyJoin(NotifyBase): # Set our priority if 'priority' in results['qsd'] and len(results['qsd']['priority']): - _map = { - 'l': JoinPriority.LOW, - 'm': JoinPriority.MODERATE, - 'n': JoinPriority.NORMAL, - 'h': JoinPriority.HIGH, - 'e': JoinPriority.EMERGENCY, - } - try: - results['priority'] = \ - _map[results['qsd']['priority'][0].lower()] - - except KeyError: - # No priority was set - pass + results['priority'] = \ + NotifyJoin.unquote(results['qsd']['priority']) # Our Devices results['targets'] = list() diff --git a/apprise/plugins/NotifyNtfy.py b/apprise/plugins/NotifyNtfy.py index 61c8f5dc..74a030ab 100644 --- a/apprise/plugins/NotifyNtfy.py +++ b/apprise/plugins/NotifyNtfy.py @@ -86,6 +86,39 @@ NTFY_PRIORITIES = ( NtfyPriority.MIN, ) +NTFY_PRIORITY_MAP = { + # Maps against string 'low' but maps to Moderate to avoid + # conflicting with actual ntfy mappings + 'l': NtfyPriority.LOW, + # Maps against string 'moderate' + 'mo': NtfyPriority.LOW, + # Maps against string 'normal' + 'n': NtfyPriority.NORMAL, + # Maps against string 'high' + 'h': NtfyPriority.HIGH, + # Maps against string 'emergency' + 'e': NtfyPriority.MAX, + + # Entries to additionally support (so more like Ntfy's API) + # Maps against string 'min' + 'mi': NtfyPriority.MIN, + # Maps against string 'max' + 'ma': NtfyPriority.MAX, + # Maps against string 'default' + 'd': NtfyPriority.NORMAL, + + # support 1-5 values as well + '1': NtfyPriority.MIN, + # Maps against string 'moderate' + '2': NtfyPriority.LOW, + # Maps against string 'normal' + '3': NtfyPriority.NORMAL, + # Maps against string 'high' + '4': NtfyPriority.HIGH, + # Maps against string 'emergency' + '5': NtfyPriority.MAX, +} + class NotifyNtfy(NotifyBase): """ @@ -237,18 +270,13 @@ class NotifyNtfy(NotifyBase): # An email to forward notifications to self.email = email - # The priority of the message - - if priority is None: - self.priority = self.template_args['priority']['default'] - else: - self.priority = priority - - if self.priority not in NTFY_PRIORITIES: - msg = 'An invalid ntfy Priority ({}) was specified.'.format( - priority) - self.logger.warning(msg) - raise TypeError(msg) + # The Priority of the message + self.priority = NotifyNtfy.template_args['priority']['default'] \ + if not priority else \ + next(( + v for k, v in NTFY_PRIORITY_MAP.items() + if str(priority).lower().startswith(k)), + NotifyNtfy.template_args['priority']['default']) # Any optional tags to attach to the notification self.__tags = parse_list(tags) @@ -565,31 +593,10 @@ class NotifyNtfy(NotifyBase): # We're done early as we couldn't load the results return results + # Set our priority if 'priority' in results['qsd'] and len(results['qsd']['priority']): - _map = { - # Supported lookups - 'mi': NtfyPriority.MIN, - '1': NtfyPriority.MIN, - 'l': NtfyPriority.LOW, - '2': NtfyPriority.LOW, - 'n': NtfyPriority.NORMAL, # support normal keyword - 'd': NtfyPriority.NORMAL, # default keyword - '3': NtfyPriority.NORMAL, - 'h': NtfyPriority.HIGH, - '4': NtfyPriority.HIGH, - 'ma': NtfyPriority.MAX, - '5': NtfyPriority.MAX, - } - try: - # pretty-format (and update short-format) - results['priority'] = \ - _map[results['qsd']['priority'][0:2].lower()] - - except KeyError: - # Pass along what was set so it can be handed during - # initialization - results['priority'] = str(results['qsd']['priority']) - pass + results['priority'] = \ + NotifyNtfy.unquote(results['qsd']['priority']) if 'attach' in results['qsd'] and len(results['qsd']['attach']): results['attach'] = NotifyNtfy.unquote(results['qsd']['attach']) diff --git a/apprise/plugins/NotifyOpsgenie.py b/apprise/plugins/NotifyOpsgenie.py index da63a1d8..9ae9ed79 100644 --- a/apprise/plugins/NotifyOpsgenie.py +++ b/apprise/plugins/NotifyOpsgenie.py @@ -101,13 +101,40 @@ class OpsgeniePriority(object): EMERGENCY = 5 -OPSGENIE_PRIORITIES = ( - OpsgeniePriority.LOW, - OpsgeniePriority.MODERATE, - OpsgeniePriority.NORMAL, - OpsgeniePriority.HIGH, - OpsgeniePriority.EMERGENCY, -) +OPSGENIE_PRIORITIES = { + # Note: This also acts as a reverse lookup mapping + OpsgeniePriority.LOW: 'low', + OpsgeniePriority.MODERATE: 'moderate', + OpsgeniePriority.NORMAL: 'normal', + OpsgeniePriority.HIGH: 'high', + OpsgeniePriority.EMERGENCY: 'emergency', +} + +OPSGENIE_PRIORITY_MAP = { + # Maps against string 'low' + 'l': OpsgeniePriority.LOW, + # Maps against string 'moderate' + 'm': OpsgeniePriority.MODERATE, + # Maps against string 'normal' + 'n': OpsgeniePriority.NORMAL, + # Maps against string 'high' + 'h': OpsgeniePriority.HIGH, + # Maps against string 'emergency' + 'e': OpsgeniePriority.EMERGENCY, + + # Entries to additionally support (so more like Opsgenie's API) + '1': OpsgeniePriority.LOW, + '2': OpsgeniePriority.MODERATE, + '3': OpsgeniePriority.NORMAL, + '4': OpsgeniePriority.HIGH, + '5': OpsgeniePriority.EMERGENCY, + # Support p-prefix + 'p1': OpsgeniePriority.LOW, + 'p2': OpsgeniePriority.MODERATE, + 'p3': OpsgeniePriority.NORMAL, + 'p4': OpsgeniePriority.HIGH, + 'p5': OpsgeniePriority.EMERGENCY, +} class NotifyOpsgenie(NotifyBase): @@ -246,11 +273,12 @@ class NotifyOpsgenie(NotifyBase): raise TypeError(msg) # The Priority of the message - if priority not in OPSGENIE_PRIORITIES: - self.priority = OpsgeniePriority.NORMAL - - else: - self.priority = priority + self.priority = NotifyOpsgenie.template_args['priority']['default'] \ + if not priority else \ + next(( + v for k, v in OPSGENIE_PRIORITY_MAP.items() + if str(priority).lower().startswith(k)), + NotifyOpsgenie.template_args['priority']['default']) # Store our region try: @@ -450,20 +478,13 @@ class NotifyOpsgenie(NotifyBase): Returns the URL built dynamically based on specified arguments. """ - _map = { - OpsgeniePriority.LOW: 'low', - OpsgeniePriority.MODERATE: 'moderate', - OpsgeniePriority.NORMAL: 'normal', - OpsgeniePriority.HIGH: 'high', - OpsgeniePriority.EMERGENCY: 'emergency', - } - # Define any URL parameters params = { 'region': self.region_name, 'priority': - _map[OpsgeniePriority.NORMAL] if self.priority not in _map - else _map[self.priority], + OPSGENIE_PRIORITIES[self.template_args['priority']['default']] + if self.priority not in OPSGENIE_PRIORITIES + else OPSGENIE_PRIORITIES[self.priority], 'batch': 'yes' if self.batch_size > 1 else 'no', } @@ -530,38 +551,10 @@ class NotifyOpsgenie(NotifyBase): results['details'] = {NotifyBase.unquote(x): NotifyBase.unquote(y) for x, y in results['qsd+'].items()} + # Set our priority if 'priority' in results['qsd'] and len(results['qsd']['priority']): - _map = { - # Letter Assignnments - 'l': OpsgeniePriority.LOW, - 'm': OpsgeniePriority.MODERATE, - 'n': OpsgeniePriority.NORMAL, - 'h': OpsgeniePriority.HIGH, - 'e': OpsgeniePriority.EMERGENCY, - 'lo': OpsgeniePriority.LOW, - 'me': OpsgeniePriority.MODERATE, - 'no': OpsgeniePriority.NORMAL, - 'hi': OpsgeniePriority.HIGH, - 'em': OpsgeniePriority.EMERGENCY, - # Support 3rd Party API Documented Scale - '1': OpsgeniePriority.LOW, - '2': OpsgeniePriority.MODERATE, - '3': OpsgeniePriority.NORMAL, - '4': OpsgeniePriority.HIGH, - '5': OpsgeniePriority.EMERGENCY, - 'p1': OpsgeniePriority.LOW, - 'p2': OpsgeniePriority.MODERATE, - 'p3': OpsgeniePriority.NORMAL, - 'p4': OpsgeniePriority.HIGH, - 'p5': OpsgeniePriority.EMERGENCY, - } - try: - results['priority'] = \ - _map[results['qsd']['priority'][0:2].lower()] - - except KeyError: - # No priority was set - pass + results['priority'] = \ + NotifyOpsgenie.unquote(results['qsd']['priority']) # Get Batch Boolean (if set) results['batch'] = \ diff --git a/apprise/plugins/NotifyProwl.py b/apprise/plugins/NotifyProwl.py index 95673b3a..3564d9a6 100644 --- a/apprise/plugins/NotifyProwl.py +++ b/apprise/plugins/NotifyProwl.py @@ -40,13 +40,34 @@ class ProwlPriority(object): EMERGENCY = 2 -PROWL_PRIORITIES = ( - ProwlPriority.LOW, - ProwlPriority.MODERATE, - ProwlPriority.NORMAL, - ProwlPriority.HIGH, - ProwlPriority.EMERGENCY, -) +PROWL_PRIORITIES = { + # Note: This also acts as a reverse lookup mapping + ProwlPriority.LOW: 'low', + ProwlPriority.MODERATE: 'moderate', + ProwlPriority.NORMAL: 'normal', + ProwlPriority.HIGH: 'high', + ProwlPriority.EMERGENCY: 'emergency', +} + +PROWL_PRIORITY_MAP = { + # Maps against string 'low' + 'l': ProwlPriority.LOW, + # Maps against string 'moderate' + 'm': ProwlPriority.MODERATE, + # Maps against string 'normal' + 'n': ProwlPriority.NORMAL, + # Maps against string 'high' + 'h': ProwlPriority.HIGH, + # Maps against string 'emergency' + 'e': ProwlPriority.EMERGENCY, + + # Entries to additionally support (so more like Prowl's API) + '-2': ProwlPriority.LOW, + '-1': ProwlPriority.MODERATE, + '0': ProwlPriority.NORMAL, + '1': ProwlPriority.HIGH, + '2': ProwlPriority.EMERGENCY, +} # Provide some known codes Prowl uses and what they translate to: PROWL_HTTP_ERROR_MAP = { @@ -124,11 +145,13 @@ class NotifyProwl(NotifyBase): """ super(NotifyProwl, self).__init__(**kwargs) - if priority not in PROWL_PRIORITIES: - self.priority = self.template_args['priority']['default'] - - else: - self.priority = priority + # The Priority of the message + self.priority = NotifyProwl.template_args['priority']['default'] \ + if not priority else \ + next(( + v for k, v in PROWL_PRIORITY_MAP.items() + if str(priority).lower().startswith(k)), + NotifyProwl.template_args['priority']['default']) # API Key (associated with project) self.apikey = validate_regex( @@ -229,18 +252,12 @@ class NotifyProwl(NotifyBase): Returns the URL built dynamically based on specified arguments. """ - _map = { - ProwlPriority.LOW: 'low', - ProwlPriority.MODERATE: 'moderate', - ProwlPriority.NORMAL: 'normal', - ProwlPriority.HIGH: 'high', - ProwlPriority.EMERGENCY: 'emergency', - } - # Define any URL parameters params = { - 'priority': 'normal' if self.priority not in _map - else _map[self.priority], + 'priority': + PROWL_PRIORITIES[self.template_args['priority']['default']] + if self.priority not in PROWL_PRIORITIES + else PROWL_PRIORITIES[self.priority], } # Extend our parameters @@ -276,32 +293,9 @@ class NotifyProwl(NotifyBase): except IndexError: pass + # Set our priority if 'priority' in results['qsd'] and len(results['qsd']['priority']): - _map = { - # Letter Assignments - 'l': ProwlPriority.LOW, - 'm': ProwlPriority.MODERATE, - 'n': ProwlPriority.NORMAL, - 'h': ProwlPriority.HIGH, - 'e': ProwlPriority.EMERGENCY, - 'lo': ProwlPriority.LOW, - 'me': ProwlPriority.MODERATE, - 'no': ProwlPriority.NORMAL, - 'hi': ProwlPriority.HIGH, - 'em': ProwlPriority.EMERGENCY, - # Support 3rd Party Documented Scale - '-2': ProwlPriority.LOW, - '-1': ProwlPriority.MODERATE, - '0': ProwlPriority.NORMAL, - '1': ProwlPriority.HIGH, - '2': ProwlPriority.EMERGENCY, - } - try: - results['priority'] = \ - _map[results['qsd']['priority'][0:2].lower()] - - except KeyError: - # No priority was set - pass + results['priority'] = \ + NotifyProwl.unquote(results['qsd']['priority']) return results diff --git a/apprise/plugins/NotifyPushover.py b/apprise/plugins/NotifyPushover.py index f1d1e2a0..ef694d24 100644 --- a/apprise/plugins/NotifyPushover.py +++ b/apprise/plugins/NotifyPushover.py @@ -102,13 +102,34 @@ PUSHOVER_SOUNDS = ( PushoverSound.NONE, ) -PUSHOVER_PRIORITIES = ( - PushoverPriority.LOW, - PushoverPriority.MODERATE, - PushoverPriority.NORMAL, - PushoverPriority.HIGH, - PushoverPriority.EMERGENCY, -) +PUSHOVER_PRIORITIES = { + # Note: This also acts as a reverse lookup mapping + PushoverPriority.LOW: 'low', + PushoverPriority.MODERATE: 'moderate', + PushoverPriority.NORMAL: 'normal', + PushoverPriority.HIGH: 'high', + PushoverPriority.EMERGENCY: 'emergency', +} + +PUSHOVER_PRIORITY_MAP = { + # Maps against string 'low' + 'l': PushoverPriority.LOW, + # Maps against string 'moderate' + 'm': PushoverPriority.MODERATE, + # Maps against string 'normal' + 'n': PushoverPriority.NORMAL, + # Maps against string 'high' + 'h': PushoverPriority.HIGH, + # Maps against string 'emergency' + 'e': PushoverPriority.EMERGENCY, + + # Entries to additionally support (so more like Pushover's API) + '-2': PushoverPriority.LOW, + '-1': PushoverPriority.MODERATE, + '0': PushoverPriority.NORMAL, + '1': PushoverPriority.HIGH, + '2': PushoverPriority.EMERGENCY, +} # Extend HTTP Error Messages PUSHOVER_HTTP_ERROR_MAP = { @@ -265,11 +286,13 @@ class NotifyPushover(NotifyBase): raise TypeError(msg) # The Priority of the message - if priority not in PUSHOVER_PRIORITIES: - self.priority = self.template_args['priority']['default'] - - else: - self.priority = priority + self.priority = int( + NotifyPushover.template_args['priority']['default'] + if priority is None else + next(( + v for k, v in PUSHOVER_PRIORITY_MAP.items() + if str(priority).lower().startswith(k)), + NotifyPushover.template_args['priority']['default'])) # The following are for emergency alerts if self.priority == PushoverPriority.EMERGENCY: @@ -510,19 +533,12 @@ class NotifyPushover(NotifyBase): Returns the URL built dynamically based on specified arguments. """ - _map = { - PushoverPriority.LOW: 'low', - PushoverPriority.MODERATE: 'moderate', - PushoverPriority.NORMAL: 'normal', - PushoverPriority.HIGH: 'high', - PushoverPriority.EMERGENCY: 'emergency', - } - # Define any URL parameters params = { 'priority': - _map[self.template_args['priority']['default']] - if self.priority not in _map else _map[self.priority], + PUSHOVER_PRIORITIES[self.template_args['priority']['default']] + if self.priority not in PUSHOVER_PRIORITIES + else PUSHOVER_PRIORITIES[self.priority], } # Only add expire and retry for emergency messages, @@ -563,26 +579,8 @@ class NotifyPushover(NotifyBase): # Set our priority if 'priority' in results['qsd'] and len(results['qsd']['priority']): - _map = { - # Keep for backwards compatibility - 'l': PushoverPriority.LOW, - 'm': PushoverPriority.MODERATE, - 'n': PushoverPriority.NORMAL, - 'h': PushoverPriority.HIGH, - 'e': PushoverPriority.EMERGENCY, - - # Entries to additionally support (so more like PushOver's API) - '-2': PushoverPriority.LOW, - '-1': PushoverPriority.MODERATE, - '0': PushoverPriority.NORMAL, - '1': PushoverPriority.HIGH, - '2': PushoverPriority.EMERGENCY, - } - priority = results['qsd']['priority'].lower() results['priority'] = \ - next(( - v for k, v in _map.items() if priority.startswith(k)), - NotifyPushover.template_args['priority']['default']) + NotifyPushover.unquote(results['qsd']['priority']) # Retrieve all of our targets results['targets'] = NotifyPushover.split_path(results['fullpath']) diff --git a/test/test_apprise_config.py b/test/test_apprise_config.py index b565f19b..d5c3aecd 100644 --- a/test/test_apprise_config.py +++ b/test/test_apprise_config.py @@ -31,9 +31,9 @@ import pytest from apprise import NotifyFormat from apprise import ConfigFormat from apprise import ContentIncludeMode -from apprise.Apprise import Apprise -from apprise.AppriseConfig import AppriseConfig -from apprise.AppriseAsset import AppriseAsset +from apprise import Apprise +from apprise import AppriseConfig +from apprise import AppriseAsset from apprise.config.ConfigBase import ConfigBase from apprise.plugins.NotifyBase import NotifyBase diff --git a/test/test_plugin_dapnet.py b/test/test_plugin_dapnet.py index 9f63c8f2..885dd831 100644 --- a/test/test_plugin_dapnet.py +++ b/test/test_plugin_dapnet.py @@ -25,7 +25,10 @@ # Disable logging for a cleaner testing output import logging import requests +import mock +import apprise +from apprise.plugins.NotifyDapnet import DapnetPriority from apprise import plugins from helpers import AppriseURLTester @@ -129,3 +132,72 @@ def test_plugin_dapnet_urls(): # Run our general tests AppriseURLTester(tests=apprise_url_tests).run_all() + + +@mock.patch('requests.post') +def test_plugin_dapnet_config_files(mock_post): + """ + NotifyDapnet() Config File Cases + """ + content = """ + urls: + - dapnet://user:pass@DF1ABC: + - priority: 0 + tag: dapnet_int normal + - priority: "0" + tag: dapnet_str_int normal + - priority: normal + tag: dapnet_str normal + + # This will take on normal (default) priority + - priority: invalid + tag: dapnet_invalid + + - dapnet://user1:pass2@DF1ABC: + - priority: 1 + tag: dapnet_int emerg + - priority: "1" + tag: dapnet_str_int emerg + - priority: emergency + tag: dapnet_str emerg + """ + + # Disable Throttling to speed testing + plugins.NotifyDapnet.request_rate_per_sec = 0 + + # Prepare Mock + mock_post.return_value = requests.Request() + mock_post.return_value.status_code = requests.codes.created + + # Create ourselves a config object + ac = apprise.AppriseConfig() + assert ac.add_config(content=content) is True + + aobj = apprise.Apprise() + + # Add our configuration + aobj.add(ac) + + # We should be able to read our 7 servers from that + # 4x normal (invalid + 3 exclusivly specified to be so) + # 3x emerg + assert len(ac.servers()) == 7 + assert len(aobj) == 7 + assert len([x for x in aobj.find(tag='normal')]) == 3 + for s in aobj.find(tag='normal'): + assert s.priority == DapnetPriority.NORMAL + + assert len([x for x in aobj.find(tag='emerg')]) == 3 + for s in aobj.find(tag='emerg'): + assert s.priority == DapnetPriority.EMERGENCY + + assert len([x for x in aobj.find(tag='dapnet_str')]) == 2 + assert len([x for x in aobj.find(tag='dapnet_str_int')]) == 2 + assert len([x for x in aobj.find(tag='dapnet_int')]) == 2 + + assert len([x for x in aobj.find(tag='dapnet_invalid')]) == 1 + assert next(aobj.find(tag='dapnet_invalid')).priority == \ + DapnetPriority.NORMAL + + # Notifications work + assert aobj.notify(title="title", body="body") is True diff --git a/test/test_plugin_glib.py b/test/test_plugin_glib.py index e5ba0386..ed823563 100644 --- a/test/test_plugin_glib.py +++ b/test/test_plugin_glib.py @@ -51,6 +51,7 @@ if 'dbus' not in sys.modules: pytest.skip("Skipping dbus-python based tests", allow_module_level=True) from dbus import DBusException # noqa E402 +from apprise.plugins.NotifyDBus import DBusUrgency # noqa E402 @mock.patch('dbus.SessionBus') @@ -265,13 +266,78 @@ def test_plugin_dbus_general(mock_mainloop, mock_byte, mock_bytearray, title='title', body='body', notify_type=apprise.NotifyType.INFO) is True - obj = apprise.Apprise.instantiate( - 'dbus://_/?x=invalid&y=invalid', suppress_exceptions=False) - assert isinstance(obj, apprise.plugins.NotifyDBus) is True - assert isinstance(obj.url(), six.string_types) is True - assert obj.notify( - title='title', body='body', - notify_type=apprise.NotifyType.INFO) is True + with pytest.raises(TypeError): + obj = apprise.Apprise.instantiate( + 'dbus://_/?x=invalid&y=invalid', suppress_exceptions=False) + + # Test configuration parsing + content = """ + urls: + - dbus://: + - priority: 0 + tag: dbus_int low + - priority: "0" + tag: dbus_str_int low + - priority: low + tag: dbus_str low + - urgency: 0 + tag: dbus_int low + - urgency: "0" + tag: dbus_str_int low + - urgency: low + tag: dbus_str low + + # These will take on normal (default) urgency + - priority: invalid + tag: dbus_invalid + - urgency: invalid + tag: dbus_invalid + + - dbus://: + - priority: 2 + tag: dbus_int high + - priority: "2" + tag: dbus_str_int high + - priority: high + tag: dbus_str high + - urgency: 2 + tag: dbus_int high + - urgency: "2" + tag: dbus_str_int high + - urgency: high + tag: dbus_str high + """ + + # Create ourselves a config object + ac = apprise.AppriseConfig() + assert ac.add_config(content=content) is True + + aobj = apprise.Apprise() + + # Add our configuration + aobj.add(ac) + + # We should be able to read our 14 servers from that + # 6x low + # 6x high + # 2x invalid (so takes on normal urgency) + assert len(ac.servers()) == 14 + assert len(aobj) == 14 + assert len([x for x in aobj.find(tag='low')]) == 6 + for s in aobj.find(tag='low'): + assert s.urgency == DBusUrgency.LOW + + assert len([x for x in aobj.find(tag='high')]) == 6 + for s in aobj.find(tag='high'): + assert s.urgency == DBusUrgency.HIGH + + assert len([x for x in aobj.find(tag='dbus_str')]) == 4 + assert len([x for x in aobj.find(tag='dbus_str_int')]) == 4 + assert len([x for x in aobj.find(tag='dbus_int')]) == 4 + + assert len([x for x in aobj.find(tag='dbus_invalid')]) == 2 + for s in aobj.find(tag='dbus_invalid'): + assert s.urgency == DBusUrgency.NORMAL # If our underlining object throws for whatever rea on, we will # gracefully fail diff --git a/test/test_plugin_gnome.py b/test/test_plugin_gnome.py index a47ae114..39b73b84 100644 --- a/test/test_plugin_gnome.py +++ b/test/test_plugin_gnome.py @@ -29,6 +29,7 @@ import sys import types import pytest import apprise +from apprise.plugins.NotifyGnome import GnomeUrgency try: # Python v3.4+ @@ -189,6 +190,75 @@ def test_plugin_gnome_general(): assert obj.notify(title='title', body='body', notify_type=apprise.NotifyType.INFO) is True + # Test configuration parsing + content = """ + urls: + - gnome://: + - priority: 0 + tag: gnome_int low + - priority: "0" + tag: gnome_str_int low + - priority: low + tag: gnome_str low + - urgency: 0 + tag: gnome_int low + - urgency: "0" + tag: gnome_str_int low + - urgency: low + tag: gnome_str low + + # These will take on normal (default) urgency + - priority: invalid + tag: gnome_invalid + - urgency: invalid + tag: gnome_invalid + + - gnome://: + - priority: 2 + tag: gnome_int high + - priority: "2" + tag: gnome_str_int high + - priority: high + tag: gnome_str high + - urgency: 2 + tag: gnome_int high + - urgency: "2" + tag: gnome_str_int high + - urgency: high + tag: gnome_str high + """ + + # Create ourselves a config object + ac = apprise.AppriseConfig() + assert ac.add_config(content=content) is True + + aobj = apprise.Apprise() + + # Add our configuration + aobj.add(ac) + + # We should be able to read our 14 servers from that + # 6x low + # 6x high + # 2x invalid (so takes on normal urgency) + assert len(ac.servers()) == 14 + assert len(aobj) == 14 + assert len([x for x in aobj.find(tag='low')]) == 6 + for s in aobj.find(tag='low'): + assert s.urgency == GnomeUrgency.LOW + + assert len([x for x in aobj.find(tag='high')]) == 6 + for s in aobj.find(tag='high'): + assert s.urgency == GnomeUrgency.HIGH + + assert len([x for x in aobj.find(tag='gnome_str')]) == 4 + assert len([x for x in aobj.find(tag='gnome_str_int')]) == 4 + assert len([x for x in aobj.find(tag='gnome_int')]) == 4 + + assert len([x for x in aobj.find(tag='gnome_invalid')]) == 2 + for s in aobj.find(tag='gnome_invalid'): + assert s.urgency == GnomeUrgency.NORMAL + # Test our loading of our icon exception; it will still allow the # notification to be sent mock_pixbuf.new_from_file.side_effect = AttributeError() diff --git a/test/test_plugin_gotify.py b/test/test_plugin_gotify.py index 62bd074c..9fb4213c 100644 --- a/test/test_plugin_gotify.py +++ b/test/test_plugin_gotify.py @@ -22,9 +22,12 @@ # 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. +import mock import pytest import requests +import apprise from apprise import plugins +from apprise.plugins.NotifyGotify import GotifyPriority from helpers import AppriseURLTester # Disable logging for a cleaner testing output @@ -119,3 +122,73 @@ def test_plugin_gotify_edge_cases(): # Whitespace also acts as an invalid token value with pytest.raises(TypeError): plugins.NotifyGotify(token=" ") + + +@mock.patch('requests.post') +def test_plugin_gotify_config_files(mock_post): + """ + NotifyGotify() Config File Cases + """ + content = """ + urls: + - gotify://hostname/%s: + - priority: 0 + tag: gotify_int low + - priority: "0" + tag: gotify_str_int low + # We want to make sure our '1' does not match the '10' entry + - priority: "1" + tag: gotify_str_int low + - priority: low + tag: gotify_str low + + # This will take on moderate (default) priority + - priority: invalid + tag: gotify_invalid + + - gotify://hostname/%s: + - priority: 10 + tag: gotify_int emerg + - priority: "10" + tag: gotify_str_int emerg + - priority: emergency + tag: gotify_str emerg + """ % ('a' * 16, 'b' * 16) + + # Disable Throttling to speed testing + plugins.NotifyGotify.request_rate_per_sec = 0 + + # Prepare Mock + mock_post.return_value = requests.Request() + mock_post.return_value.status_code = requests.codes.ok + + # Create ourselves a config object + ac = apprise.AppriseConfig() + assert ac.add_config(content=content) is True + + aobj = apprise.Apprise() + + # Add our configuration + aobj.add(ac) + + # We should be able to read our 8 servers from that + # 4x low + # 3x emerg + # 1x invalid (so takes on normal priority) + assert len(ac.servers()) == 8 + assert len(aobj) == 8 + assert len([x for x in aobj.find(tag='low')]) == 4 + for s in aobj.find(tag='low'): + assert s.priority == GotifyPriority.LOW + + assert len([x for x in aobj.find(tag='emerg')]) == 3 + for s in aobj.find(tag='emerg'): + assert s.priority == GotifyPriority.EMERGENCY + + assert len([x for x in aobj.find(tag='gotify_str')]) == 2 + assert len([x for x in aobj.find(tag='gotify_str_int')]) == 3 + assert len([x for x in aobj.find(tag='gotify_int')]) == 2 + + assert len([x for x in aobj.find(tag='gotify_invalid')]) == 1 + assert next(aobj.find(tag='gotify_invalid')).priority == \ + GotifyPriority.NORMAL diff --git a/test/test_plugin_growl.py b/test/test_plugin_growl.py index 1e08cc5d..8fbe9638 100644 --- a/test/test_plugin_growl.py +++ b/test/test_plugin_growl.py @@ -28,6 +28,7 @@ import mock import six import pytest import apprise +from apprise.plugins.NotifyGrowl import GrowlPriority try: from gntp import errors @@ -317,3 +318,72 @@ def test_plugin_growl_general(mock_gntp): print('%s / %s' % (url, str(e))) assert exception is not None assert isinstance(e, exception) + + +@pytest.mark.skipif( + 'gntp' not in sys.modules, reason="Requires gntp") +@mock.patch('gntp.notifier.GrowlNotifier') +def test_plugin_growl_config_files(mock_gntp): + """ + NotifyGrowl() Config File Cases + """ + content = """ + urls: + - growl://pass@growl.server: + - priority: -2 + tag: growl_int low + - priority: "-2" + tag: growl_str_int low + - priority: low + tag: growl_str low + + # This will take on moderate (default) priority + - priority: invalid + tag: growl_invalid + + - growl://pass@growl.server: + - priority: 2 + tag: growl_int emerg + - priority: "2" + tag: growl_str_int emerg + - priority: emergency + tag: growl_str emerg + """ + + # Disable Throttling to speed testing + apprise.plugins.NotifyGrowl.request_rate_per_sec = 0 + + mock_notifier = mock.Mock() + mock_gntp.return_value = mock_notifier + mock_notifier.notify.return_value = True + + # Create ourselves a config object + ac = apprise.AppriseConfig() + assert ac.add_config(content=content) is True + + aobj = apprise.Apprise() + + # Add our configuration + aobj.add(ac) + + # We should be able to read our 7 servers from that + # 3x low + # 3x emerg + # 1x invalid (so takes on normal priority) + assert len(ac.servers()) == 7 + assert len(aobj) == 7 + assert len([x for x in aobj.find(tag='low')]) == 3 + for s in aobj.find(tag='low'): + assert s.priority == GrowlPriority.LOW + + assert len([x for x in aobj.find(tag='emerg')]) == 3 + for s in aobj.find(tag='emerg'): + assert s.priority == GrowlPriority.EMERGENCY + + assert len([x for x in aobj.find(tag='growl_str')]) == 2 + assert len([x for x in aobj.find(tag='growl_str_int')]) == 2 + assert len([x for x in aobj.find(tag='growl_int')]) == 2 + + assert len([x for x in aobj.find(tag='growl_invalid')]) == 1 + assert next(aobj.find(tag='growl_invalid')).priority == \ + GrowlPriority.NORMAL diff --git a/test/test_plugin_join.py b/test/test_plugin_join.py index fa2dbed9..bd8a69c7 100644 --- a/test/test_plugin_join.py +++ b/test/test_plugin_join.py @@ -25,9 +25,11 @@ import mock import pytest import requests +import apprise from apprise import plugins from apprise import NotifyType from helpers import AppriseURLTester +from apprise.plugins.NotifyJoin import JoinPriority # Disable logging for a cleaner testing output import logging @@ -166,3 +168,73 @@ def test_plugin_join_edge_cases(mock_post, mock_get): # Test notifications without a body or a title; nothing to send # so we return False p.notify(body=None, title=None, notify_type=NotifyType.INFO) is False + + +@mock.patch('requests.post') +def test_plugin_join_config_files(mock_post): + """ + NotifyJoin() Config File Cases + """ + content = """ + urls: + - join://%s@%s: + - priority: -2 + tag: join_int low + - priority: "-2" + tag: join_str_int low + - priority: low + tag: join_str low + + # This will take on normal (default) priority + - priority: invalid + tag: join_invalid + + - join://%s@%s: + - priority: 2 + tag: join_int emerg + - priority: "2" + tag: join_str_int emerg + - priority: emergency + tag: join_str emerg + """ % ('a' * 32, 'b' * 32, 'c' * 32, 'd' * 32) + + # Disable Throttling to speed testing + plugins.NotifyJoin.request_rate_per_sec = 0 + + # Prepare Mock + mock_post.return_value = requests.Request() + mock_post.return_value.status_code = requests.codes.ok + + # Create ourselves a config object + ac = apprise.AppriseConfig() + assert ac.add_config(content=content) is True + + aobj = apprise.Apprise() + + # Add our configuration + aobj.add(ac) + + # We should be able to read our 7 servers from that + # 3x low + # 3x emerg + # 1x invalid (so takes on normal priority) + assert len(ac.servers()) == 7 + assert len(aobj) == 7 + assert len([x for x in aobj.find(tag='low')]) == 3 + for s in aobj.find(tag='low'): + assert s.priority == JoinPriority.LOW + + assert len([x for x in aobj.find(tag='emerg')]) == 3 + for s in aobj.find(tag='emerg'): + assert s.priority == JoinPriority.EMERGENCY + + assert len([x for x in aobj.find(tag='join_str')]) == 2 + assert len([x for x in aobj.find(tag='join_str_int')]) == 2 + assert len([x for x in aobj.find(tag='join_int')]) == 2 + + assert len([x for x in aobj.find(tag='join_invalid')]) == 1 + assert next(aobj.find(tag='join_invalid')).priority == \ + JoinPriority.NORMAL + + # Notifications work + assert aobj.notify(title="title", body="body") is True diff --git a/test/test_plugin_ntfy.py b/test/test_plugin_ntfy.py index 34ab3024..f7a5c063 100644 --- a/test/test_plugin_ntfy.py +++ b/test/test_plugin_ntfy.py @@ -26,11 +26,11 @@ import os import json import mock import requests -from apprise import Apprise from apprise import plugins -from apprise import NotifyType +import apprise from helpers import AppriseURLTester -from apprise import AppriseAttachment + +from apprise.plugins.NotifyNtfy import NtfyPriority # Disable logging for a cleaner testing output import logging @@ -157,10 +157,6 @@ apprise_url_tests = ( 'instance': plugins.NotifyNtfy, 'requests_response_text': GOOD_RESPONSE_TEXT, }), - # Invalid Priority - ('ntfy://localhost/topic1/?priority=invalid', { - 'instance': TypeError, - }), # A topic and port identifier ('ntfy://user:pass@localhost:8080/topic/', { 'instance': plugins.NotifyNtfy, @@ -258,7 +254,7 @@ def test_plugin_ntfy_attachments(mock_post): mock_post.reset_mock() # Prepare our object - obj = Apprise.instantiate( + obj = apprise.Apprise.instantiate( 'ntfy://user:pass@localhost:8080/topic') # Send a good attachment @@ -278,10 +274,11 @@ def test_plugin_ntfy_attachments(mock_post): mock_post.reset_mock() # prepare our attachment - attach = AppriseAttachment(os.path.join(TEST_VAR_DIR, 'apprise-test.gif')) + attach = apprise.AppriseAttachment( + os.path.join(TEST_VAR_DIR, 'apprise-test.gif')) # Prepare our object - obj = Apprise.instantiate( + obj = apprise.Apprise.instantiate( 'ntfy://user:pass@localhost:8084/topic') # Send a good attachment @@ -333,14 +330,15 @@ def test_plugin_ntfy_attachments(mock_post): # An invalid attachment will cause a failure path = os.path.join(TEST_VAR_DIR, '/invalid/path/to/an/invalid/file.jpg') - attach = AppriseAttachment(path) + attach = apprise.AppriseAttachment(path) assert obj.notify(body="test", attach=attach) is False # Test our call count assert mock_post.call_count == 0 # prepare our attachment - attach = AppriseAttachment(os.path.join(TEST_VAR_DIR, 'apprise-test.gif')) + attach = apprise.AppriseAttachment( + os.path.join(TEST_VAR_DIR, 'apprise-test.gif')) # Throw an exception on the first call to requests.post() mock_post.return_value = None @@ -414,7 +412,8 @@ def test_plugin_custom_ntfy_edge_cases(mock_post): assert 'topic1' in instance.topics assert instance.notify( - body='body', title='title', notify_type=NotifyType.INFO) is True + body='body', title='title', + notify_type=apprise.NotifyType.INFO) is True # Test our call count assert mock_post.call_count == 1 @@ -426,3 +425,75 @@ def test_plugin_custom_ntfy_edge_cases(mock_post): assert response['title'] == 'title' assert response['attach'] == 'http://example.com/file.jpg' assert response['filename'] == 'smoke.jpg' + + +@mock.patch('requests.post') +@mock.patch('requests.get') +def test_plugin_ntfy_config_files(mock_post, mock_get): + """ + NotifyNtfy() Config File Cases + """ + content = """ + urls: + - ntfy://localhost/topic1: + - priority: 1 + tag: ntfy_int min + - priority: "1" + tag: ntfy_str_int min + - priority: min + tag: ntfy_str min + + # This will take on normal (default) priority + - priority: invalid + tag: ntfy_invalid + + - ntfy://localhost/topic2: + - priority: 5 + tag: ntfy_int max + - priority: "5" + tag: ntfy_str_int max + - priority: emergency + tag: ntfy_str max + - priority: max + tag: ntfy_str max + """ + + # Disable Throttling to speed testing + plugins.NotifyNtfy.request_rate_per_sec = 0 + + # Prepare Mock + mock_post.return_value = requests.Request() + mock_post.return_value.status_code = requests.codes.ok + mock_get.return_value = requests.Request() + mock_get.return_value.status_code = requests.codes.ok + + # Create ourselves a config object + ac = apprise.AppriseConfig() + assert ac.add_config(content=content) is True + + aobj = apprise.Apprise() + + # Add our configuration + aobj.add(ac) + + # We should be able to read our 8 servers from that + # 3x min + # 4x max + # 1x invalid (so takes on normal priority) + assert len(ac.servers()) == 8 + assert len(aobj) == 8 + assert len([x for x in aobj.find(tag='min')]) == 3 + for s in aobj.find(tag='min'): + assert s.priority == NtfyPriority.MIN + + assert len([x for x in aobj.find(tag='max')]) == 4 + for s in aobj.find(tag='max'): + assert s.priority == NtfyPriority.MAX + + assert len([x for x in aobj.find(tag='ntfy_str')]) == 3 + assert len([x for x in aobj.find(tag='ntfy_str_int')]) == 2 + assert len([x for x in aobj.find(tag='ntfy_int')]) == 2 + + assert len([x for x in aobj.find(tag='ntfy_invalid')]) == 1 + assert next(aobj.find(tag='ntfy_invalid')).priority == \ + NtfyPriority.NORMAL diff --git a/test/test_plugin_opsgenie.py b/test/test_plugin_opsgenie.py index 2b720c24..0748c4c0 100644 --- a/test/test_plugin_opsgenie.py +++ b/test/test_plugin_opsgenie.py @@ -22,7 +22,10 @@ # 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. -from apprise import plugins +import mock +import requests +from apprise.plugins.NotifyOpsgenie import OpsgeniePriority +import apprise from helpers import AppriseURLTester # Disable logging for a cleaner testing output @@ -52,72 +55,72 @@ apprise_url_tests = ( }), ('opsgenie://apikey/', { # No targets specified; this is allowed - 'instance': plugins.NotifyOpsgenie, + 'instance': apprise.plugins.NotifyOpsgenie, }), ('opsgenie://apikey/user', { # Valid user - 'instance': plugins.NotifyOpsgenie, + 'instance': apprise.plugins.NotifyOpsgenie, 'privacy_url': 'opsgenie://a...y/%40user', }), ('opsgenie://apikey/@user?region=eu', { # European Region - 'instance': plugins.NotifyOpsgenie, + 'instance': apprise.plugins.NotifyOpsgenie, }), ('opsgenie://apikey/@user?entity=A%20Entity', { # Assign an entity - 'instance': plugins.NotifyOpsgenie, + 'instance': apprise.plugins.NotifyOpsgenie, }), ('opsgenie://apikey/@user?alias=An%20Alias', { # Assign an alias - 'instance': plugins.NotifyOpsgenie, + 'instance': apprise.plugins.NotifyOpsgenie, }), ('opsgenie://apikey/@user?priority=p3', { # Assign our priority - 'instance': plugins.NotifyOpsgenie, + 'instance': apprise.plugins.NotifyOpsgenie, }), ('opsgenie://apikey/?tags=comma,separated', { # Test our our 'tags' (tag is reserved in Apprise) but not 'tags' # Also test the fact we do not need to define a target - 'instance': plugins.NotifyOpsgenie, + 'instance': apprise.plugins.NotifyOpsgenie, }), ('opsgenie://apikey/@user?priority=invalid', { # Invalid priority (loads using default) - 'instance': plugins.NotifyOpsgenie, + 'instance': apprise.plugins.NotifyOpsgenie, }), ('opsgenie://apikey/user@email.com/#team/*sche/^esc/%20/a', { # Valid user (email), valid schedule, Escalated ID, # an invalid entry (%20), and too short of an entry (a) - 'instance': plugins.NotifyOpsgenie, + 'instance': apprise.plugins.NotifyOpsgenie, }), ('opsgenie://apikey/{}/@{}/#{}/*{}/^{}/'.format( UUID4, UUID4, UUID4, UUID4, UUID4), { # similar to the above, except we use the UUID's - 'instance': plugins.NotifyOpsgenie, + 'instance': apprise.plugins.NotifyOpsgenie, }), ('opsgenie://apikey?to=#team,user&+key=value&+type=override', { # Test to= and details (key/value pair) also override 'type' - 'instance': plugins.NotifyOpsgenie, + 'instance': apprise.plugins.NotifyOpsgenie, }), ('opsgenie://apikey/#team/@user/?batch=yes', { # Test batch= - 'instance': plugins.NotifyOpsgenie, + 'instance': apprise.plugins.NotifyOpsgenie, }), ('opsgenie://apikey/#team/@user/?batch=no', { # Test batch= - 'instance': plugins.NotifyOpsgenie, + 'instance': apprise.plugins.NotifyOpsgenie, }), ('opsgenie://?apikey=abc&to=user', { # Test Kwargs - 'instance': plugins.NotifyOpsgenie, + 'instance': apprise.plugins.NotifyOpsgenie, }), ('opsgenie://apikey/#team/user/', { - 'instance': plugins.NotifyOpsgenie, + 'instance': apprise.plugins.NotifyOpsgenie, # throw a bizzare code forcing us to fail to look it up 'response': False, 'requests_response_code': 999, }), ('opsgenie://apikey/#topic1/device/', { - 'instance': plugins.NotifyOpsgenie, + 'instance': apprise.plugins.NotifyOpsgenie, # Throws a series of connection and transfer exceptions when this flag # is set and tests that we gracfully handle them 'test_requests_exceptions': True, @@ -131,5 +134,80 @@ def test_plugin_opsgenie_urls(): """ + # Disable Throttling to speed testing + apprise.plugins.NotifyOpsgenie.request_rate_per_sec = 0 + # Run our general tests AppriseURLTester(tests=apprise_url_tests).run_all() + + +@mock.patch('requests.post') +def test_plugin_opsgenie_config_files(mock_post): + """ + NotifyOpsgenie() Config File Cases + """ + content = """ + urls: + - opsgenie://apikey/user: + - priority: 1 + tag: opsgenie_int low + - priority: "1" + tag: opsgenie_str_int low + - priority: "p1" + tag: opsgenie_pstr_int low + - priority: low + tag: opsgenie_str low + + # This will take on moderate (default) priority + - priority: invalid + tag: opsgenie_invalid + + - opsgenie://apikey2/user2: + - priority: 5 + tag: opsgenie_int emerg + - priority: "5" + tag: opsgenie_str_int emerg + - priority: "p5" + tag: opsgenie_pstr_int emerg + - priority: emergency + tag: opsgenie_str emerg + """ + + # Disable Throttling to speed testing + apprise.plugins.NotifyOpsgenie.request_rate_per_sec = 0 + + # Prepare Mock + mock_post.return_value = requests.Request() + mock_post.return_value.status_code = requests.codes.ok + + # Create ourselves a config object + ac = apprise.AppriseConfig() + assert ac.add_config(content=content) is True + + aobj = apprise.Apprise() + + # Add our configuration + aobj.add(ac) + + # We should be able to read our 9 servers from that + # 4x low + # 4x emerg + # 1x invalid (so takes on normal priority) + assert len(ac.servers()) == 9 + assert len(aobj) == 9 + assert len([x for x in aobj.find(tag='low')]) == 4 + for s in aobj.find(tag='low'): + assert s.priority == OpsgeniePriority.LOW + + assert len([x for x in aobj.find(tag='emerg')]) == 4 + for s in aobj.find(tag='emerg'): + assert s.priority == OpsgeniePriority.EMERGENCY + + assert len([x for x in aobj.find(tag='opsgenie_str')]) == 2 + assert len([x for x in aobj.find(tag='opsgenie_str_int')]) == 2 + assert len([x for x in aobj.find(tag='opsgenie_pstr_int')]) == 2 + assert len([x for x in aobj.find(tag='opsgenie_int')]) == 2 + + assert len([x for x in aobj.find(tag='opsgenie_invalid')]) == 1 + assert next(aobj.find(tag='opsgenie_invalid')).priority == \ + OpsgeniePriority.NORMAL diff --git a/test/test_plugin_prowl.py b/test/test_plugin_prowl.py index c765093d..d0d69b2a 100644 --- a/test/test_plugin_prowl.py +++ b/test/test_plugin_prowl.py @@ -22,9 +22,12 @@ # 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. +import mock import pytest import requests +from apprise.plugins.NotifyProwl import ProwlPriority from apprise import plugins +import apprise from helpers import AppriseURLTester # Disable logging for a cleaner testing output @@ -140,3 +143,70 @@ def test_plugin_prowl_edge_cases(): plugins.NotifyProwl(apikey='abcd', providerkey=object()) with pytest.raises(TypeError): plugins.NotifyProwl(apikey='abcd', providerkey=' ') + + +@mock.patch('requests.post') +def test_plugin_prowl_config_files(mock_post): + """ + NotifyProwl() Config File Cases + """ + content = """ + urls: + - prowl://%s: + - priority: -2 + tag: prowl_int low + - priority: "-2" + tag: prowl_str_int low + - priority: low + tag: prowl_str low + + # This will take on moderate (default) priority + - priority: invalid + tag: prowl_invalid + + - prowl://%s: + - priority: 2 + tag: prowl_int emerg + - priority: "2" + tag: prowl_str_int emerg + - priority: emergency + tag: prowl_str emerg + """ % ('a' * 40, 'b' * 40) + + # Disable Throttling to speed testing + plugins.NotifyProwl.request_rate_per_sec = 0 + + # Prepare Mock + mock_post.return_value = requests.Request() + mock_post.return_value.status_code = requests.codes.ok + + # Create ourselves a config object + ac = apprise.AppriseConfig() + assert ac.add_config(content=content) is True + + aobj = apprise.Apprise() + + # Add our configuration + aobj.add(ac) + + # We should be able to read our 7 servers from that + # 3x low + # 3x emerg + # 1x invalid (so takes on normal priority) + assert len(ac.servers()) == 7 + assert len(aobj) == 7 + assert len([x for x in aobj.find(tag='low')]) == 3 + for s in aobj.find(tag='low'): + assert s.priority == ProwlPriority.LOW + + assert len([x for x in aobj.find(tag='emerg')]) == 3 + for s in aobj.find(tag='emerg'): + assert s.priority == ProwlPriority.EMERGENCY + + assert len([x for x in aobj.find(tag='prowl_str')]) == 2 + assert len([x for x in aobj.find(tag='prowl_str_int')]) == 2 + assert len([x for x in aobj.find(tag='prowl_int')]) == 2 + + assert len([x for x in aobj.find(tag='prowl_invalid')]) == 1 + assert next(aobj.find(tag='prowl_invalid')).priority == \ + ProwlPriority.NORMAL diff --git a/test/test_plugin_pushover.py b/test/test_plugin_pushover.py index fef109fa..0f60362d 100644 --- a/test/test_plugin_pushover.py +++ b/test/test_plugin_pushover.py @@ -28,10 +28,9 @@ import mock import requests import pytest from json import dumps -from apprise import Apprise -from apprise import NotifyType -from apprise import AppriseAttachment +from apprise.plugins.NotifyPushover import PushoverPriority from apprise import plugins +import apprise from helpers import AppriseURLTester # Disable logging for a cleaner testing output @@ -194,7 +193,7 @@ def test_plugin_pushover_attachments(mock_post, tmpdir): """ # Disable Throttling to speed testing - plugins.NotifyBase.request_rate_per_sec = 0 + plugins.NotifyPushover.request_rate_per_sec = 0 # Initialize some generic (but valid) tokens user_key = 'u' * 30 @@ -216,10 +215,11 @@ def test_plugin_pushover_attachments(mock_post, tmpdir): mock_post.return_value = response # prepare our attachment - attach = AppriseAttachment(os.path.join(TEST_VAR_DIR, 'apprise-test.gif')) + attach = apprise.AppriseAttachment( + os.path.join(TEST_VAR_DIR, 'apprise-test.gif')) # Instantiate our object - obj = Apprise.instantiate( + obj = apprise.Apprise.instantiate( 'pover://{}@{}/'.format(user_key, api_token)) assert isinstance(obj, plugins.NotifyPushover) @@ -251,7 +251,7 @@ def test_plugin_pushover_attachments(mock_post, tmpdir): image = tmpdir.mkdir("pover_image").join("test.jpg") image.write('a' * plugins.NotifyPushover.attach_max_size_bytes) - attach = AppriseAttachment.instantiate(str(image)) + attach = apprise.AppriseAttachment.instantiate(str(image)) assert obj.notify(body="test", attach=attach) is True # Test our call count @@ -263,16 +263,17 @@ def test_plugin_pushover_attachments(mock_post, tmpdir): mock_post.reset_mock() # Add 1 more byte to the file (putting it over the limit) - image.write('a' * (plugins.NotifyPushover.attach_max_size_bytes + 1)) + image.write( + 'a' * (plugins.NotifyPushover.attach_max_size_bytes + 1)) - attach = AppriseAttachment.instantiate(str(image)) + attach = apprise.AppriseAttachment.instantiate(str(image)) assert obj.notify(body="test", attach=attach) is False # Test our call count assert mock_post.call_count == 0 # Test case when file is missing - attach = AppriseAttachment.instantiate( + attach = apprise.AppriseAttachment.instantiate( 'file://{}?cache=False'.format(str(image))) os.unlink(str(image)) assert obj.notify( @@ -284,13 +285,14 @@ def test_plugin_pushover_attachments(mock_post, tmpdir): # Test unsuported files: image = tmpdir.mkdir("pover_unsupported").join("test.doc") image.write('a' * 256) - attach = AppriseAttachment.instantiate(str(image)) + attach = apprise.AppriseAttachment.instantiate(str(image)) # Content is silently ignored assert obj.notify(body="test", attach=attach) is True # prepare our attachment - attach = AppriseAttachment(os.path.join(TEST_VAR_DIR, 'apprise-test.gif')) + attach = apprise.AppriseAttachment( + os.path.join(TEST_VAR_DIR, 'apprise-test.gif')) # Throw an exception on the first call to requests.post() for side_effect in (requests.RequestException(), OSError(), bad_response): @@ -303,15 +305,14 @@ def test_plugin_pushover_attachments(mock_post, tmpdir): assert obj.send(body="test") is False -@mock.patch('requests.get') @mock.patch('requests.post') -def test_plugin_pushover_edge_cases(mock_post, mock_get): +def test_plugin_pushover_edge_cases(mock_post): """ NotifyPushover() Edge Cases """ # Disable Throttling to speed testing - plugins.NotifyBase.request_rate_per_sec = 0 + plugins.NotifyPushover.request_rate_per_sec = 0 # No token with pytest.raises(TypeError): @@ -327,10 +328,8 @@ def test_plugin_pushover_edge_cases(mock_post, mock_get): devices = 'device1,device2,,,,%s' % invalid_device # 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 # No webhook id specified with pytest.raises(TypeError): @@ -344,7 +343,7 @@ def test_plugin_pushover_edge_cases(mock_post, mock_get): # This call fails because there is 1 invalid device assert obj.notify( body='body', title='title', - notify_type=NotifyType.INFO) is False + notify_type=apprise.NotifyType.INFO) is False obj = plugins.NotifyPushover(user_key=user_key, token=token) assert isinstance(obj, plugins.NotifyPushover) is True @@ -354,9 +353,11 @@ def test_plugin_pushover_edge_cases(mock_post, mock_get): # This call succeeds because all of the devices are valid assert obj.notify( - body='body', title='title', notify_type=NotifyType.INFO) is True + body='body', title='title', + notify_type=apprise.NotifyType.INFO) is True - obj = plugins.NotifyPushover(user_key=user_key, token=token, targets=set()) + obj = plugins.NotifyPushover( + user_key=user_key, token=token, targets=set()) assert isinstance(obj, plugins.NotifyPushover) is True # Default is to send to all devices, so there will be a # device defined here @@ -372,3 +373,73 @@ def test_plugin_pushover_edge_cases(mock_post, mock_get): with pytest.raises(TypeError): plugins.NotifyPushover(user_key="abcd", token=" ") + + +@mock.patch('requests.post') +def test_plugin_pushover_config_files(mock_post): + """ + NotifyPushover() Config File Cases + """ + content = """ + urls: + - pover://USER@TOKEN: + - priority: -2 + tag: pushover_int low + - priority: "-2" + tag: pushover_str_int low + - priority: low + tag: pushover_str low + + # This will take on normal (default) priority + - priority: invalid + tag: pushover_invalid + + - pover://USER2@TOKEN2: + - priority: 2 + tag: pushover_int emerg + - priority: "2" + tag: pushover_str_int emerg + - priority: emergency + tag: pushover_str emerg + """ + + # Disable Throttling to speed testing + plugins.NotifyPushover.request_rate_per_sec = 0 + + # Prepare Mock + mock_post.return_value = requests.Request() + mock_post.return_value.status_code = requests.codes.ok + + # Create ourselves a config object + ac = apprise.AppriseConfig() + assert ac.add_config(content=content) is True + + aobj = apprise.Apprise() + + # Add our configuration + aobj.add(ac) + + # We should be able to read our 7 servers from that + # 3x low + # 3x emerg + # 1x invalid (so takes on normal priority) + assert len(ac.servers()) == 7 + assert len(aobj) == 7 + assert len([x for x in aobj.find(tag='low')]) == 3 + for s in aobj.find(tag='low'): + assert s.priority == PushoverPriority.LOW + + assert len([x for x in aobj.find(tag='emerg')]) == 3 + for s in aobj.find(tag='emerg'): + assert s.priority == PushoverPriority.EMERGENCY + + assert len([x for x in aobj.find(tag='pushover_str')]) == 2 + assert len([x for x in aobj.find(tag='pushover_str_int')]) == 2 + assert len([x for x in aobj.find(tag='pushover_int')]) == 2 + + assert len([x for x in aobj.find(tag='pushover_invalid')]) == 1 + assert next(aobj.find(tag='pushover_invalid')).priority == \ + PushoverPriority.NORMAL + + # Notifications work + assert aobj.notify(title="title", body="body") is True