diff --git a/apprise/plugins/NotifyGrowl/__init__.py b/apprise/plugins/NotifyGrowl.py similarity index 74% rename from apprise/plugins/NotifyGrowl/__init__.py rename to apprise/plugins/NotifyGrowl.py index d35ade64..a74c56a4 100644 --- a/apprise/plugins/NotifyGrowl/__init__.py +++ b/apprise/plugins/NotifyGrowl.py @@ -1,6 +1,6 @@ # -*- coding: utf-8 -*- # -# Copyright (C) 2019 Chris Caron +# Copyright (C) 2020 Chris Caron # All rights reserved. # # This code is licensed under the MIT License. @@ -23,14 +23,26 @@ # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN # THE SOFTWARE. -from .gntp import notifier -from .gntp import errors -from ..NotifyBase import NotifyBase -from ...URLBase import PrivacyMode -from ...common import NotifyImageSize -from ...common import NotifyType -from ...utils import parse_bool -from ...AppriseLocale import gettext_lazy as _ +from .NotifyBase import NotifyBase +from ..URLBase import PrivacyMode +from ..common import NotifyImageSize +from ..common import NotifyType +from ..utils import parse_bool +from ..AppriseLocale import gettext_lazy as _ + +# Default our global support flag +NOTIFY_GROWL_SUPPORT_ENABLED = False + +try: + import gntp.notifier + + # We're good to go! + NOTIFY_GROWL_SUPPORT_ENABLED = True + +except ImportError: + # No problem; we just simply can't support this plugin until + # gntp is installed + pass # Priorities @@ -50,8 +62,6 @@ GROWL_PRIORITIES = ( GrowlPriority.EMERGENCY, ) -GROWL_NOTIFICATION_TYPE = "New Messages" - class NotifyGrowl(NotifyBase): """ @@ -74,14 +84,19 @@ class NotifyGrowl(NotifyBase): # Allows the user to specify the NotifyImageSize object image_size = NotifyImageSize.XY_72 + # This entry is a bit hacky, but it allows us to unit-test this library + # in an environment that simply doesn't have the windows packages + # available to us. It also allows us to handle situations where the + # packages actually are present but we need to test that they aren't. + # If anyone is seeing this had knows a better way of testing this + # outside of what is defined in test/test_growl_plugin.py, please + # let me know! :) + _enabled = NOTIFY_GROWL_SUPPORT_ENABLED + # Disable throttle rate for Growl requests since they are normally # local anyway request_rate_per_sec = 0 - # A title can not be used for Growl Messages. Setting this to zero will - # cause any title (if defined) to get placed into the message body. - title_maxlen = 0 - # Limit results to just the first 10 line otherwise there is just to much # content to display body_max_line_count = 2 @@ -89,7 +104,9 @@ class NotifyGrowl(NotifyBase): # Default Growl Port default_port = 23053 - # Define object templates + # The Growl notification type used + growl_notification_type = "New Messages" + # Define object templates templates = ( '{schema}://{host}', @@ -138,9 +155,16 @@ class NotifyGrowl(NotifyBase): 'default': True, 'map_to': 'include_image', }, + 'sticky': { + 'name': _('Sticky'), + 'type': 'bool', + 'default': True, + 'map_to': 'sticky', + }, }) - def __init__(self, priority=None, version=2, include_image=True, **kwargs): + def __init__(self, priority=None, version=2, include_image=True, + sticky=False, **kwargs): """ Initialize Growl Object """ @@ -156,16 +180,29 @@ class NotifyGrowl(NotifyBase): else: self.priority = priority - # Always default the sticky flag to False - self.sticky = False + # Our Registered object + self.growl = None + + # Sticky flag + self.sticky = sticky # Store Version self.version = version + # Track whether or not we want to send an image with our notification + # or not. + self.include_image = include_image + + return + + def register(self): + """ + Registers with the Growl server + """ payload = { 'applicationName': self.app_id, - 'notifications': [GROWL_NOTIFICATION_TYPE, ], - 'defaultNotifications': [GROWL_NOTIFICATION_TYPE, ], + 'notifications': [self.growl_notification_type, ], + 'defaultNotifications': [self.growl_notification_type, ], 'hostname': self.host, 'port': self.port, } @@ -174,43 +211,58 @@ class NotifyGrowl(NotifyBase): payload['password'] = self.password self.logger.debug('Growl Registration Payload: %s' % str(payload)) - self.growl = notifier.GrowlNotifier(**payload) + self.growl = gntp.notifier.GrowlNotifier(**payload) try: self.growl.register() - self.logger.debug( - 'Growl server registration completed successfully.' - ) - except errors.NetworkError: - msg = 'A network error occurred sending Growl ' \ - 'notification to {}.'.format(self.host) + except gntp.errors.NetworkError: + msg = 'A network error error occurred registering ' \ + 'with Growl at {}.'.format(self.host) self.logger.warning(msg) - raise TypeError(msg) + return False - except errors.AuthError: - msg = 'An authentication error occurred sending Growl ' \ - 'notification to {}.'.format(self.host) + except gntp.errors.ParseError: + msg = 'A parsing error error occurred registering ' \ + 'with Growl at {}.'.format(self.host) self.logger.warning(msg) - raise TypeError(msg) + return False - except errors.UnsupportedError: - msg = 'An unsupported error occurred sending Growl ' \ - 'notification to {}.'.format(self.host) + except gntp.errors.AuthError: + msg = 'An authentication error error occurred registering ' \ + 'with Growl at {}.'.format(self.host) self.logger.warning(msg) - raise TypeError(msg) + return False - # Track whether or not we want to send an image with our notification - # or not. - self.include_image = include_image + except gntp.errors.UnsupportedError: + msg = 'An unsupported error occurred registering with ' \ + 'Growl at {}.'.format(self.host) + self.logger.warning(msg) + return False - return + self.logger.debug( + 'Growl server registration completed successfully.' + ) + + # Return our state + return True def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs): """ Perform Growl Notification """ + if not self._enabled: + self.logger.warning( + "Growl Notifications are not supported by this system; " + "`pip install gntp`.") + return False + + # Register ourselves with the server if we haven't done so already + if not self.growl and not self.register(): + # We failed to register + return False + icon = None if self.version >= 2: # URL Based @@ -223,11 +275,11 @@ class NotifyGrowl(NotifyBase): else self.image_raw(notify_type) payload = { - 'noteType': GROWL_NOTIFICATION_TYPE, + 'noteType': self.growl_notification_type, 'title': title, 'description': body, 'icon': icon is not None, - 'sticky': False, + 'sticky': self.sticky, 'priority': self.priority, } self.logger.debug('Growl Payload: %s' % str(payload)) @@ -241,6 +293,7 @@ class NotifyGrowl(NotifyBase): self.throttle() try: + # Perform notification response = self.growl.notify(**payload) if not isinstance(response, bool): self.logger.warning( @@ -251,7 +304,7 @@ class NotifyGrowl(NotifyBase): else: self.logger.info('Sent Growl notification.') - except errors.BaseError as e: + except gntp.errors.BaseError as e: # Since Growl servers listen for UDP broadcasts, it's possible # that you will never get to this part of the code since there is # no acknowledgement as to whether it accepted what was sent to it @@ -285,6 +338,7 @@ class NotifyGrowl(NotifyBase): # 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], @@ -365,7 +419,13 @@ class NotifyGrowl(NotifyBase): # Include images with our message results['include_image'] = \ - parse_bool(results['qsd'].get('image', True)) + parse_bool(results['qsd'].get('image', + NotifyGrowl.template_args['image']['default'])) + + # Include images with our message + results['sticky'] = \ + parse_bool(results['qsd'].get('sticky', + NotifyGrowl.template_args['sticky']['default'])) # Set our version if version: diff --git a/apprise/plugins/NotifyGrowl/gntp/__init__.py b/apprise/plugins/NotifyGrowl/gntp/__init__.py deleted file mode 100644 index e69de29b..00000000 diff --git a/apprise/plugins/NotifyGrowl/gntp/cli.py b/apprise/plugins/NotifyGrowl/gntp/cli.py deleted file mode 100644 index 0dc61d0a..00000000 --- a/apprise/plugins/NotifyGrowl/gntp/cli.py +++ /dev/null @@ -1,141 +0,0 @@ -# Copyright: 2013 Paul Traylor -# These sources are released under the terms of the MIT license: see LICENSE - -import logging -import os -import sys -from optparse import OptionParser, OptionGroup - -from .notifier import GrowlNotifier -from .shim import RawConfigParser -from .version import __version__ - -DEFAULT_CONFIG = os.path.expanduser('~/.gntp') - -config = RawConfigParser({ - 'hostname': 'localhost', - 'password': None, - 'port': 23053, -}) -config.read([DEFAULT_CONFIG]) -if not config.has_section('gntp'): - config.add_section('gntp') - - -class ClientParser(OptionParser): - def __init__(self): - OptionParser.__init__(self, version="%%prog %s" % __version__) - - group = OptionGroup(self, "Network Options") - group.add_option("-H", "--host", - dest="host", default=config.get('gntp', 'hostname'), - help="Specify a hostname to which to send a remote notification. [%default]") - group.add_option("--port", - dest="port", default=config.getint('gntp', 'port'), type="int", - help="port to listen on [%default]") - group.add_option("-P", "--password", - dest='password', default=config.get('gntp', 'password'), - help="Network password") - self.add_option_group(group) - - group = OptionGroup(self, "Notification Options") - group.add_option("-n", "--name", - dest="app", default='Python GNTP Test Client', - help="Set the name of the application [%default]") - group.add_option("-s", "--sticky", - dest='sticky', default=False, action="store_true", - help="Make the notification sticky [%default]") - group.add_option("--image", - dest="icon", default=None, - help="Icon for notification (URL or /path/to/file)") - group.add_option("-m", "--message", - dest="message", default=None, - help="Sets the message instead of using stdin") - group.add_option("-p", "--priority", - dest="priority", default=0, type="int", - help="-2 to 2 [%default]") - group.add_option("-d", "--identifier", - dest="identifier", - help="Identifier for coalescing") - group.add_option("-t", "--title", - dest="title", default=None, - help="Set the title of the notification [%default]") - group.add_option("-N", "--notification", - dest="name", default='Notification', - help="Set the notification name [%default]") - group.add_option("--callback", - dest="callback", - help="URL callback") - self.add_option_group(group) - - # Extra Options - self.add_option('-v', '--verbose', - dest='verbose', default=0, action='count', - help="Verbosity levels") - - def parse_args(self, args=None, values=None): - values, args = OptionParser.parse_args(self, args, values) - - if values.message is None: - print('Enter a message followed by Ctrl-D') - try: - message = sys.stdin.read() - except KeyboardInterrupt: - exit() - else: - message = values.message - - if values.title is None: - values.title = ' '.join(args) - - # If we still have an empty title, use the - # first bit of the message as the title - if values.title == '': - values.title = message[:20] - - values.verbose = logging.WARNING - values.verbose * 10 - - return values, message - - -def main(): - (options, message) = ClientParser().parse_args() - logging.basicConfig(level=options.verbose) - if not os.path.exists(DEFAULT_CONFIG): - logging.info('No config read found at %s', DEFAULT_CONFIG) - - growl = GrowlNotifier( - applicationName=options.app, - notifications=[options.name], - defaultNotifications=[options.name], - hostname=options.host, - password=options.password, - port=options.port, - ) - result = growl.register() - if result is not True: - exit(result) - - # This would likely be better placed within the growl notifier - # class but until I make _checkIcon smarter this is "easier" - if options.icon is not None and not options.icon.startswith('http'): - logging.info('Loading image %s', options.icon) - f = open(options.icon) - options.icon = f.read() - f.close() - - result = growl.notify( - noteType=options.name, - title=options.title, - description=message, - icon=options.icon, - sticky=options.sticky, - priority=options.priority, - callback=options.callback, - identifier=options.identifier, - ) - if result is not True: - exit(result) - -if __name__ == "__main__": - main() diff --git a/apprise/plugins/NotifyGrowl/gntp/config.py b/apprise/plugins/NotifyGrowl/gntp/config.py deleted file mode 100644 index e7dda48a..00000000 --- a/apprise/plugins/NotifyGrowl/gntp/config.py +++ /dev/null @@ -1,77 +0,0 @@ -# Copyright: 2013 Paul Traylor -# These sources are released under the terms of the MIT license: see LICENSE - -""" -The gntp.config module is provided as an extended GrowlNotifier object that takes -advantage of the ConfigParser module to allow us to setup some default values -(such as hostname, password, and port) in a more global way to be shared among -programs using gntp -""" -import logging -import os - -from .gntp import notifier -from .gntp import shim - -__all__ = [ - 'mini', - 'GrowlNotifier' -] - -logger = logging.getLogger('gntp') - - -class GrowlNotifier(notifier.GrowlNotifier): - """ - ConfigParser enhanced GrowlNotifier object - - For right now, we are only interested in letting users overide certain - values from ~/.gntp - - :: - - [gntp] - hostname = ? - password = ? - port = ? - """ - def __init__(self, *args, **kwargs): - config = shim.RawConfigParser({ - 'hostname': kwargs.get('hostname', 'localhost'), - 'password': kwargs.get('password'), - 'port': kwargs.get('port', 23053), - }) - - config.read([os.path.expanduser('~/.gntp')]) - - # If the file does not exist, then there will be no gntp section defined - # and the config.get() lines below will get confused. Since we are not - # saving the config, it should be safe to just add it here so the - # code below doesn't complain - if not config.has_section('gntp'): - logger.info('Error reading ~/.gntp config file') - config.add_section('gntp') - - kwargs['password'] = config.get('gntp', 'password') - kwargs['hostname'] = config.get('gntp', 'hostname') - kwargs['port'] = config.getint('gntp', 'port') - - super(GrowlNotifier, self).__init__(*args, **kwargs) - - -def mini(description, **kwargs): - """Single notification function - - Simple notification function in one line. Has only one required parameter - and attempts to use reasonable defaults for everything else - :param string description: Notification message - """ - kwargs['notifierFactory'] = GrowlNotifier - notifier.mini(description, **kwargs) - - -if __name__ == '__main__': - # If we're running this module directly we're likely running it as a test - # so extra debugging is useful - logging.basicConfig(level=logging.INFO) - mini('Testing mini notification') diff --git a/apprise/plugins/NotifyGrowl/gntp/core.py b/apprise/plugins/NotifyGrowl/gntp/core.py deleted file mode 100644 index 99db7570..00000000 --- a/apprise/plugins/NotifyGrowl/gntp/core.py +++ /dev/null @@ -1,511 +0,0 @@ -# Copyright: 2013 Paul Traylor -# These sources are released under the terms of the MIT license: see LICENSE - -import hashlib -import re -import time - -from . import shim -from . import errors as errors - -__all__ = [ - 'GNTPRegister', - 'GNTPNotice', - 'GNTPSubscribe', - 'GNTPOK', - 'GNTPError', - 'parse_gntp', -] - -#GNTP/ [:][ :.] -GNTP_INFO_LINE = re.compile( - r'GNTP/(?P\d+\.\d+) (?PREGISTER|NOTIFY|SUBSCRIBE|\-OK|\-ERROR)' + - r' (?P[A-Z0-9]+(:(?P[A-F0-9]+))?) ?' + - r'((?P[A-Z0-9]+):(?P[A-F0-9]+).(?P[A-F0-9]+))?\r\n', - re.IGNORECASE -) - -GNTP_INFO_LINE_SHORT = re.compile( - r'GNTP/(?P\d+\.\d+) (?PREGISTER|NOTIFY|SUBSCRIBE|\-OK|\-ERROR)', - re.IGNORECASE -) - -GNTP_HEADER = re.compile(r'([\w-]+):(.+)') - -GNTP_EOL = shim.b('\r\n') -GNTP_SEP = shim.b(': ') - - -class _GNTPBuffer(shim.StringIO): - """GNTP Buffer class""" - def writeln(self, value=None): - if value: - self.write(shim.b(value)) - self.write(GNTP_EOL) - - def writeheader(self, key, value): - if not isinstance(value, str): - value = str(value) - self.write(shim.b(key)) - self.write(GNTP_SEP) - self.write(shim.b(value)) - self.write(GNTP_EOL) - - -class _GNTPBase(object): - """Base initilization - - :param string messagetype: GNTP Message type - :param string version: GNTP Protocol version - :param string encription: Encryption protocol - """ - def __init__(self, messagetype=None, version='1.0', encryption=None): - self.info = { - 'version': version, - 'messagetype': messagetype, - 'encryptionAlgorithmID': encryption - } - self.hash_algo = { - 'MD5': hashlib.md5, - 'SHA1': hashlib.sha1, - 'SHA256': hashlib.sha256, - 'SHA512': hashlib.sha512, - } - self.headers = {} - self.resources = {} - - def __str__(self): - return self.encode() - - def _parse_info(self, data): - """Parse the first line of a GNTP message to get security and other info values - - :param string data: GNTP Message - :return dict: Parsed GNTP Info line - """ - - match = GNTP_INFO_LINE.match(data) - - if not match: - raise errors.ParseError('ERROR_PARSING_INFO_LINE') - - info = match.groupdict() - if info['encryptionAlgorithmID'] == 'NONE': - info['encryptionAlgorithmID'] = None - - return info - - def set_password(self, password, encryptAlgo='MD5'): - """Set a password for a GNTP Message - - :param string password: Null to clear password - :param string encryptAlgo: Supports MD5, SHA1, SHA256, SHA512 - """ - if not password: - self.info['encryptionAlgorithmID'] = None - self.info['keyHashAlgorithm'] = None - return - - self.password = shim.b(password) - self.encryptAlgo = encryptAlgo.upper() - - if not self.encryptAlgo in self.hash_algo: - raise errors.UnsupportedError('INVALID HASH "%s"' % self.encryptAlgo) - - hashfunction = self.hash_algo.get(self.encryptAlgo) - - password = password.encode('utf8') - seed = time.ctime().encode('utf8') - salt = hashfunction(seed).hexdigest() - saltHash = hashfunction(seed).digest() - keyBasis = password + saltHash - key = hashfunction(keyBasis).digest() - keyHash = hashfunction(key).hexdigest() - - self.info['keyHashAlgorithmID'] = self.encryptAlgo - self.info['keyHash'] = keyHash.upper() - self.info['salt'] = salt.upper() - - def _decode_hex(self, value): - """Helper function to decode hex string to `proper` hex string - - :param string value: Human readable hex string - :return string: Hex string - """ - result = '' - for i in range(0, len(value), 2): - tmp = int(value[i:i + 2], 16) - result += chr(tmp) - return result - - def _decode_binary(self, rawIdentifier, identifier): - rawIdentifier += '\r\n\r\n' - dataLength = int(identifier['Length']) - pointerStart = self.raw.find(rawIdentifier) + len(rawIdentifier) - pointerEnd = pointerStart + dataLength - data = self.raw[pointerStart:pointerEnd] - if not len(data) == dataLength: - raise errors.ParseError('INVALID_DATA_LENGTH Expected: %s Recieved %s' % (dataLength, len(data))) - return data - - def _validate_password(self, password): - """Validate GNTP Message against stored password""" - self.password = password - if password is None: - raise errors.AuthError('Missing password') - keyHash = self.info.get('keyHash', None) - if keyHash is None and self.password is None: - return True - if keyHash is None: - raise errors.AuthError('Invalid keyHash') - if self.password is None: - raise errors.AuthError('Missing password') - - keyHashAlgorithmID = self.info.get('keyHashAlgorithmID','MD5') - - password = self.password.encode('utf8') - saltHash = self._decode_hex(self.info['salt']) - - keyBasis = password + saltHash - self.key = self.hash_algo[keyHashAlgorithmID](keyBasis).digest() - keyHash = self.hash_algo[keyHashAlgorithmID](self.key).hexdigest() - - if not keyHash.upper() == self.info['keyHash'].upper(): - raise errors.AuthError('Invalid Hash') - return True - - def validate(self): - """Verify required headers""" - for header in self._requiredHeaders: - if not self.headers.get(header, False): - raise errors.ParseError('Missing Notification Header: ' + header) - - def _format_info(self): - """Generate info line for GNTP Message - - :return string: - """ - info = 'GNTP/%s %s' % ( - self.info.get('version'), - self.info.get('messagetype'), - ) - if self.info.get('encryptionAlgorithmID', None): - info += ' %s:%s' % ( - self.info.get('encryptionAlgorithmID'), - self.info.get('ivValue'), - ) - else: - info += ' NONE' - - if self.info.get('keyHashAlgorithmID', None): - info += ' %s:%s.%s' % ( - self.info.get('keyHashAlgorithmID'), - self.info.get('keyHash'), - self.info.get('salt') - ) - - return info - - def _parse_dict(self, data): - """Helper function to parse blocks of GNTP headers into a dictionary - - :param string data: - :return dict: Dictionary of parsed GNTP Headers - """ - d = {} - for line in data.split('\r\n'): - match = GNTP_HEADER.match(line) - if not match: - continue - - key = match.group(1).strip() - val = match.group(2).strip() - d[key] = val - return d - - def add_header(self, key, value): - self.headers[key] = value - - def add_resource(self, data): - """Add binary resource - - :param string data: Binary Data - """ - data = shim.b(data) - identifier = hashlib.md5(data).hexdigest() - self.resources[identifier] = data - return 'x-growl-resource://%s' % identifier - - def decode(self, data, password=None): - """Decode GNTP Message - - :param string data: - """ - self.password = password - self.raw = shim.u(data) - parts = self.raw.split('\r\n\r\n') - self.info = self._parse_info(self.raw) - self.headers = self._parse_dict(parts[0]) - - def encode(self): - """Encode a generic GNTP Message - - :return string: GNTP Message ready to be sent. Returned as a byte string - """ - - buff = _GNTPBuffer() - - buff.writeln(self._format_info()) - - #Headers - for k, v in self.headers.items(): - buff.writeheader(k, v) - buff.writeln() - - #Resources - for resource, data in self.resources.items(): - buff.writeheader('Identifier', resource) - buff.writeheader('Length', len(data)) - buff.writeln() - buff.write(data) - buff.writeln() - buff.writeln() - - return buff.getvalue() - - -class GNTPRegister(_GNTPBase): - """Represents a GNTP Registration Command - - :param string data: (Optional) See decode() - :param string password: (Optional) Password to use while encoding/decoding messages - """ - _requiredHeaders = [ - 'Application-Name', - 'Notifications-Count' - ] - _requiredNotificationHeaders = ['Notification-Name'] - - def __init__(self, data=None, password=None): - _GNTPBase.__init__(self, 'REGISTER') - self.notifications = [] - - if data: - self.decode(data, password) - else: - self.set_password(password) - self.add_header('Application-Name', 'pygntp') - self.add_header('Notifications-Count', 0) - - def validate(self): - '''Validate required headers and validate notification headers''' - for header in self._requiredHeaders: - if not self.headers.get(header, False): - raise errors.ParseError('Missing Registration Header: ' + header) - for notice in self.notifications: - for header in self._requiredNotificationHeaders: - if not notice.get(header, False): - raise errors.ParseError('Missing Notification Header: ' + header) - - def decode(self, data, password): - """Decode existing GNTP Registration message - - :param string data: Message to decode - """ - self.raw = shim.u(data) - parts = self.raw.split('\r\n\r\n') - self.info = self._parse_info(self.raw) - self._validate_password(password) - self.headers = self._parse_dict(parts[0]) - - for i, part in enumerate(parts): - if i == 0: - continue # Skip Header - if part.strip() == '': - continue - notice = self._parse_dict(part) - if notice.get('Notification-Name', False): - self.notifications.append(notice) - elif notice.get('Identifier', False): - notice['Data'] = self._decode_binary(part, notice) - #open('register.png','wblol').write(notice['Data']) - self.resources[notice.get('Identifier')] = notice - - def add_notification(self, name, enabled=True): - """Add new Notification to Registration message - - :param string name: Notification Name - :param boolean enabled: Enable this notification by default - """ - notice = {} - notice['Notification-Name'] = name - notice['Notification-Enabled'] = enabled - - self.notifications.append(notice) - self.add_header('Notifications-Count', len(self.notifications)) - - def encode(self): - """Encode a GNTP Registration Message - - :return string: Encoded GNTP Registration message. Returned as a byte string - """ - - buff = _GNTPBuffer() - - buff.writeln(self._format_info()) - - #Headers - for k, v in self.headers.items(): - buff.writeheader(k, v) - buff.writeln() - - #Notifications - if len(self.notifications) > 0: - for notice in self.notifications: - for k, v in notice.items(): - buff.writeheader(k, v) - buff.writeln() - - #Resources - for resource, data in self.resources.items(): - buff.writeheader('Identifier', resource) - buff.writeheader('Length', len(data)) - buff.writeln() - buff.write(data) - buff.writeln() - buff.writeln() - - return buff.getvalue() - - -class GNTPNotice(_GNTPBase): - """Represents a GNTP Notification Command - - :param string data: (Optional) See decode() - :param string app: (Optional) Set Application-Name - :param string name: (Optional) Set Notification-Name - :param string title: (Optional) Set Notification Title - :param string password: (Optional) Password to use while encoding/decoding messages - """ - _requiredHeaders = [ - 'Application-Name', - 'Notification-Name', - 'Notification-Title' - ] - - def __init__(self, data=None, app=None, name=None, title=None, password=None): - _GNTPBase.__init__(self, 'NOTIFY') - - if data: - self.decode(data, password) - else: - self.set_password(password) - if app: - self.add_header('Application-Name', app) - if name: - self.add_header('Notification-Name', name) - if title: - self.add_header('Notification-Title', title) - - def decode(self, data, password): - """Decode existing GNTP Notification message - - :param string data: Message to decode. - """ - self.raw = shim.u(data) - parts = self.raw.split('\r\n\r\n') - self.info = self._parse_info(self.raw) - self._validate_password(password) - self.headers = self._parse_dict(parts[0]) - - for i, part in enumerate(parts): - if i == 0: - continue # Skip Header - if part.strip() == '': - continue - notice = self._parse_dict(part) - if notice.get('Identifier', False): - notice['Data'] = self._decode_binary(part, notice) - #open('notice.png','wblol').write(notice['Data']) - self.resources[notice.get('Identifier')] = notice - - -class GNTPSubscribe(_GNTPBase): - """Represents a GNTP Subscribe Command - - :param string data: (Optional) See decode() - :param string password: (Optional) Password to use while encoding/decoding messages - """ - _requiredHeaders = [ - 'Subscriber-ID', - 'Subscriber-Name', - ] - - def __init__(self, data=None, password=None): - _GNTPBase.__init__(self, 'SUBSCRIBE') - if data: - self.decode(data, password) - else: - self.set_password(password) - - -class GNTPOK(_GNTPBase): - """Represents a GNTP OK Response - - :param string data: (Optional) See _GNTPResponse.decode() - :param string action: (Optional) Set type of action the OK Response is for - """ - _requiredHeaders = ['Response-Action'] - - def __init__(self, data=None, action=None): - _GNTPBase.__init__(self, '-OK') - if data: - self.decode(data) - if action: - self.add_header('Response-Action', action) - - -class GNTPError(_GNTPBase): - """Represents a GNTP Error response - - :param string data: (Optional) See _GNTPResponse.decode() - :param string errorcode: (Optional) Error code - :param string errordesc: (Optional) Error Description - """ - _requiredHeaders = ['Error-Code', 'Error-Description'] - - def __init__(self, data=None, errorcode=None, errordesc=None): - _GNTPBase.__init__(self, '-ERROR') - if data: - self.decode(data) - if errorcode: - self.add_header('Error-Code', errorcode) - self.add_header('Error-Description', errordesc) - - def error(self): - return (self.headers.get('Error-Code', None), - self.headers.get('Error-Description', None)) - - -def parse_gntp(data, password=None): - """Attempt to parse a message as a GNTP message - - :param string data: Message to be parsed - :param string password: Optional password to be used to verify the message - """ - data = shim.u(data) - match = GNTP_INFO_LINE_SHORT.match(data) - if not match: - raise errors.ParseError('INVALID_GNTP_INFO') - info = match.groupdict() - if info['messagetype'] == 'REGISTER': - return GNTPRegister(data, password=password) - elif info['messagetype'] == 'NOTIFY': - return GNTPNotice(data, password=password) - elif info['messagetype'] == 'SUBSCRIBE': - return GNTPSubscribe(data, password=password) - elif info['messagetype'] == '-OK': - return GNTPOK(data) - elif info['messagetype'] == '-ERROR': - return GNTPError(data) - raise errors.ParseError('INVALID_GNTP_MESSAGE') diff --git a/apprise/plugins/NotifyGrowl/gntp/errors.py b/apprise/plugins/NotifyGrowl/gntp/errors.py deleted file mode 100644 index c006fd68..00000000 --- a/apprise/plugins/NotifyGrowl/gntp/errors.py +++ /dev/null @@ -1,25 +0,0 @@ -# Copyright: 2013 Paul Traylor -# These sources are released under the terms of the MIT license: see LICENSE - -class BaseError(Exception): - pass - - -class ParseError(BaseError): - errorcode = 500 - errordesc = 'Error parsing the message' - - -class AuthError(BaseError): - errorcode = 400 - errordesc = 'Error with authorization' - - -class UnsupportedError(BaseError): - errorcode = 500 - errordesc = 'Currently unsupported by gntp.py' - - -class NetworkError(BaseError): - errorcode = 500 - errordesc = "Error connecting to growl server" diff --git a/apprise/plugins/NotifyGrowl/gntp/notifier.py b/apprise/plugins/NotifyGrowl/gntp/notifier.py deleted file mode 100644 index 38d8328f..00000000 --- a/apprise/plugins/NotifyGrowl/gntp/notifier.py +++ /dev/null @@ -1,265 +0,0 @@ -# Copyright: 2013 Paul Traylor -# These sources are released under the terms of the MIT license: see LICENSE - -""" -The gntp.notifier module is provided as a simple way to send notifications -using GNTP - -.. note:: - This class is intended to mostly mirror the older Python bindings such - that you should be able to replace instances of the old bindings with - this class. - `Original Python bindings `_ - -""" -import logging -import platform -import socket -import sys - -from .version import __version__ -from . import core -from . import errors as errors -from . import shim - -__all__ = [ - 'mini', - 'GrowlNotifier', -] - -logger = logging.getLogger('gntp') - - -class GrowlNotifier(object): - """Helper class to simplfy sending Growl messages - - :param string applicationName: Sending application name - :param list notification: List of valid notifications - :param list defaultNotifications: List of notifications that should be enabled - by default - :param string applicationIcon: Icon URL - :param string hostname: Remote host - :param integer port: Remote port - """ - - passwordHash = 'MD5' - socketTimeout = 3 - - def __init__(self, applicationName='Python GNTP', notifications=[], - defaultNotifications=None, applicationIcon=None, hostname='localhost', - password=None, port=23053): - - self.applicationName = applicationName - self.notifications = list(notifications) - if defaultNotifications: - self.defaultNotifications = list(defaultNotifications) - else: - self.defaultNotifications = self.notifications - self.applicationIcon = applicationIcon - - self.password = password - self.hostname = hostname - self.port = int(port) - - def _checkIcon(self, data): - ''' - Check the icon to see if it's valid - - If it's a simple URL icon, then we return True. If it's a data icon - then we return False - ''' - logger.info('Checking icon') - return shim.u(data).startswith('http') - - def register(self): - """Send GNTP Registration - - .. warning:: - Before sending notifications to Growl, you need to have - sent a registration message at least once - """ - logger.info('Sending registration to %s:%s', self.hostname, self.port) - register = core.GNTPRegister() - register.add_header('Application-Name', self.applicationName) - for notification in self.notifications: - enabled = notification in self.defaultNotifications - register.add_notification(notification, enabled) - if self.applicationIcon: - if self._checkIcon(self.applicationIcon): - register.add_header('Application-Icon', self.applicationIcon) - else: - resource = register.add_resource(self.applicationIcon) - register.add_header('Application-Icon', resource) - if self.password: - register.set_password(self.password, self.passwordHash) - self.add_origin_info(register) - self.register_hook(register) - return self._send('register', register) - - def notify(self, noteType, title, description, icon=None, sticky=False, - priority=None, callback=None, identifier=None, custom={}): - """Send a GNTP notifications - - .. warning:: - Must have registered with growl beforehand or messages will be ignored - - :param string noteType: One of the notification names registered earlier - :param string title: Notification title (usually displayed on the notification) - :param string description: The main content of the notification - :param string icon: Icon URL path - :param boolean sticky: Sticky notification - :param integer priority: Message priority level from -2 to 2 - :param string callback: URL callback - :param dict custom: Custom attributes. Key names should be prefixed with X- - according to the spec but this is not enforced by this class - - .. warning:: - For now, only URL callbacks are supported. In the future, the - callback argument will also support a function - """ - logger.info('Sending notification [%s] to %s:%s', noteType, self.hostname, self.port) - assert noteType in self.notifications - notice = core.GNTPNotice() - notice.add_header('Application-Name', self.applicationName) - notice.add_header('Notification-Name', noteType) - notice.add_header('Notification-Title', title) - if self.password: - notice.set_password(self.password, self.passwordHash) - if sticky: - notice.add_header('Notification-Sticky', sticky) - if priority: - notice.add_header('Notification-Priority', priority) - if icon: - if self._checkIcon(icon): - notice.add_header('Notification-Icon', icon) - else: - resource = notice.add_resource(icon) - notice.add_header('Notification-Icon', resource) - - if description: - notice.add_header('Notification-Text', description) - if callback: - notice.add_header('Notification-Callback-Target', callback) - if identifier: - notice.add_header('Notification-Coalescing-ID', identifier) - - for key in custom: - notice.add_header(key, custom[key]) - - self.add_origin_info(notice) - self.notify_hook(notice) - - return self._send('notify', notice) - - def subscribe(self, id, name, port): - """Send a Subscribe request to a remote machine""" - sub = core.GNTPSubscribe() - sub.add_header('Subscriber-ID', id) - sub.add_header('Subscriber-Name', name) - sub.add_header('Subscriber-Port', port) - if self.password: - sub.set_password(self.password, self.passwordHash) - - self.add_origin_info(sub) - self.subscribe_hook(sub) - - return self._send('subscribe', sub) - - def add_origin_info(self, packet): - """Add optional Origin headers to message""" - packet.add_header('Origin-Machine-Name', platform.node()) - packet.add_header('Origin-Software-Name', 'gntp.py') - packet.add_header('Origin-Software-Version', __version__) - packet.add_header('Origin-Platform-Name', platform.system()) - packet.add_header('Origin-Platform-Version', platform.platform()) - - def register_hook(self, packet): - pass - - def notify_hook(self, packet): - pass - - def subscribe_hook(self, packet): - pass - - def _send(self, messagetype, packet): - """Send the GNTP Packet""" - - packet.validate() - data = packet.encode() - - logger.debug('To : %s:%s <%s>\n%s', self.hostname, self.port, packet.__class__, data) - - s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - s.settimeout(self.socketTimeout) - try: - s.connect((self.hostname, self.port)) - s.send(data) - recv_data = s.recv(1024) - while not recv_data.endswith(shim.b("\r\n\r\n")): - recv_data += s.recv(1024) - except socket.error: - # Python2.5 and Python3 compatibile exception - exc = sys.exc_info()[1] - raise errors.NetworkError(exc) - - response = core.parse_gntp(recv_data) - s.close() - - logger.debug('From : %s:%s <%s>\n%s', self.hostname, self.port, response.__class__, response) - - if type(response) == core.GNTPOK: - return True - logger.error('Invalid response: %s', response.error()) - return response.error() - - -def mini(description, applicationName='PythonMini', noteType="Message", - title="Mini Message", applicationIcon=None, hostname='localhost', - password=None, port=23053, sticky=False, priority=None, - callback=None, notificationIcon=None, identifier=None, - notifierFactory=GrowlNotifier): - """Single notification function - - Simple notification function in one line. Has only one required parameter - and attempts to use reasonable defaults for everything else - :param string description: Notification message - - .. warning:: - For now, only URL callbacks are supported. In the future, the - callback argument will also support a function - """ - try: - growl = notifierFactory( - applicationName=applicationName, - notifications=[noteType], - defaultNotifications=[noteType], - applicationIcon=applicationIcon, - hostname=hostname, - password=password, - port=port, - ) - result = growl.register() - if result is not True: - return result - - return growl.notify( - noteType=noteType, - title=title, - description=description, - icon=notificationIcon, - sticky=sticky, - priority=priority, - callback=callback, - identifier=identifier, - ) - except Exception: - # We want the "mini" function to be simple and swallow Exceptions - # in order to be less invasive - logger.exception("Growl error") - -if __name__ == '__main__': - # If we're running this module directly we're likely running it as a test - # so extra debugging is useful - logging.basicConfig(level=logging.INFO) - mini('Testing mini notification') diff --git a/apprise/plugins/NotifyGrowl/gntp/shim.py b/apprise/plugins/NotifyGrowl/gntp/shim.py deleted file mode 100644 index 46952f06..00000000 --- a/apprise/plugins/NotifyGrowl/gntp/shim.py +++ /dev/null @@ -1,45 +0,0 @@ -# Copyright: 2013 Paul Traylor -# These sources are released under the terms of the MIT license: see LICENSE - -""" -Python2.5 and Python3.3 compatibility shim - -Heavily inspirted by the "six" library. -https://pypi.python.org/pypi/six -""" - -import sys - -PY3 = sys.version_info[0] == 3 - -if PY3: - def b(s): - if isinstance(s, bytes): - return s - return s.encode('utf8', 'replace') - - def u(s): - if isinstance(s, bytes): - return s.decode('utf8', 'replace') - return s - - from io import BytesIO as StringIO - from configparser import RawConfigParser -else: - def b(s): - if isinstance(s, unicode): # noqa - return s.encode('utf8', 'replace') - return s - - def u(s): - if isinstance(s, unicode): # noqa - return s - if isinstance(s, int): - s = str(s) - return unicode(s, "utf8", "replace") # noqa - - from StringIO import StringIO - from ConfigParser import RawConfigParser - -b.__doc__ = "Ensure we have a byte string" -u.__doc__ = "Ensure we have a unicode string" diff --git a/apprise/plugins/NotifyGrowl/gntp/version.py b/apprise/plugins/NotifyGrowl/gntp/version.py deleted file mode 100644 index 2166aaca..00000000 --- a/apprise/plugins/NotifyGrowl/gntp/version.py +++ /dev/null @@ -1,4 +0,0 @@ -# Copyright: 2013 Paul Traylor -# These sources are released under the terms of the MIT license: see LICENSE - -__version__ = '1.0.2' diff --git a/apprise/plugins/__init__.py b/apprise/plugins/__init__.py index 9c0ca2ea..cdf8836a 100644 --- a/apprise/plugins/__init__.py +++ b/apprise/plugins/__init__.py @@ -33,7 +33,6 @@ from os.path import abspath # Used for testing from . import NotifyEmail as NotifyEmailBase -from .NotifyGrowl import gntp from .NotifyXMPP import SleekXmppAdapter # NotifyBase object is passed in as a module not class @@ -63,9 +62,6 @@ __all__ = [ # Tokenizer 'url_to_dict', - # gntp (used for NotifyGrowl Testing) - 'gntp', - # sleekxmpp access points (used for NotifyXMPP Testing) 'SleekXmppAdapter', ] diff --git a/dev-requirements.txt b/dev-requirements.txt index 9745a8ce..00bde0dd 100644 --- a/dev-requirements.txt +++ b/dev-requirements.txt @@ -9,3 +9,4 @@ cryptography # Plugin Dependencies sleekxmpp +gntp diff --git a/packaging/redhat/python-apprise.spec b/packaging/redhat/python-apprise.spec index b432bad8..c317505a 100644 --- a/packaging/redhat/python-apprise.spec +++ b/packaging/redhat/python-apprise.spec @@ -84,10 +84,12 @@ BuildRequires: python-markdown %if 0%{?rhel} && 0%{?rhel} <= 7 BuildRequires: python-cryptography BuildRequires: python-babel +BuildRequires: python-gntp BuildRequires: python-yaml %else BuildRequires: python2-cryptography BuildRequires: python2-babel +BuildRequires: python2-gntp BuildRequires: python2-yaml %endif @@ -96,8 +98,10 @@ Requires: python2-requests-oauthlib Requires: python-six Requires: python-markdown %if 0%{?rhel} && 0%{?rhel} <= 7 +Requires: python-gntp Requires: python-yaml %else +Requires: python2-gntp Requires: python2-yaml %endif @@ -140,6 +144,7 @@ BuildRequires: python%{python3_pkgversion}-requests-oauthlib BuildRequires: python%{python3_pkgversion}-six BuildRequires: python%{python3_pkgversion}-click >= 5.0 BuildRequires: python%{python3_pkgversion}-markdown +BuildRequires: python%{python3_pkgversion}-gntp BuildRequires: python%{python3_pkgversion}-yaml BuildRequires: python%{python3_pkgversion}-babel BuildRequires: python%{python3_pkgversion}-cryptography @@ -147,6 +152,7 @@ Requires: python%{python3_pkgversion}-requests Requires: python%{python3_pkgversion}-requests-oauthlib Requires: python%{python3_pkgversion}-six Requires: python%{python3_pkgversion}-markdown +Requires: python%{python3_pkgversion}-gntp Requires: python%{python3_pkgversion}-yaml %if %{with tests} diff --git a/setup.cfg b/setup.cfg index 2a2015d7..e4225772 100644 --- a/setup.cfg +++ b/setup.cfg @@ -7,7 +7,7 @@ license_file = LICENSE [flake8] # We exclude packages we don't maintain -exclude = .eggs,.tox,gntp +exclude = .eggs,.tox ignore = E741,E722,W503,W504,W605 statistics = true builtins = _ diff --git a/test/test_growl_plugin.py b/test/test_growl_plugin.py index 4c520a78..f3375d53 100644 --- a/test/test_growl_plugin.py +++ b/test/test_growl_plugin.py @@ -23,127 +23,287 @@ # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN # THE SOFTWARE. +import os +import sys import mock import six -from apprise import plugins -from apprise import NotifyType -from apprise import Apprise +import pytest +import apprise # Disable logging for a cleaner testing output import logging logging.disable(logging.CRITICAL) - -TEST_URLS = ( - ################################## - # NotifyGrowl - ################################## - ('growl://', { - 'instance': None, - }), - ('growl://:@/', { - 'instance': None - }), - - ('growl://pass@growl.server', { - 'instance': plugins.NotifyGrowl, - }), - ('growl://ignored:pass@growl.server', { - 'instance': plugins.NotifyGrowl, - }), - ('growl://growl.server', { - 'instance': plugins.NotifyGrowl, - # don't include an image by default - 'include_image': False, - }), - ('growl://growl.server?version=1', { - 'instance': plugins.NotifyGrowl, - }), - # Force a failure - ('growl://growl.server?version=1', { - 'instance': plugins.NotifyGrowl, - 'growl_response': None, - }), - ('growl://growl.server?version=2', { - # don't include an image by default - 'include_image': False, - 'instance': plugins.NotifyGrowl, - }), - ('growl://growl.server?version=2', { - # don't include an image by default - 'include_image': False, - 'instance': plugins.NotifyGrowl, - 'growl_response': None, - }), - - # Priorities - ('growl://pass@growl.server?priority=low', { - 'instance': plugins.NotifyGrowl, - }), - ('growl://pass@growl.server?priority=moderate', { - 'instance': plugins.NotifyGrowl, - }), - ('growl://pass@growl.server?priority=normal', { - 'instance': plugins.NotifyGrowl, - }), - ('growl://pass@growl.server?priority=high', { - 'instance': plugins.NotifyGrowl, - }), - ('growl://pass@growl.server?priority=emergency', { - 'instance': plugins.NotifyGrowl, - }), - - # Invalid Priorities - ('growl://pass@growl.server?priority=invalid', { - 'instance': plugins.NotifyGrowl, - }), - ('growl://pass@growl.server?priority=', { - 'instance': plugins.NotifyGrowl, - }), - - # invalid version - ('growl://growl.server?version=', { - 'instance': plugins.NotifyGrowl, - }), - ('growl://growl.server?version=crap', { - 'instance': plugins.NotifyGrowl, - }), - - # Ports - ('growl://growl.changeport:2000', { - 'instance': plugins.NotifyGrowl, - }), - ('growl://growl.garbageport:garbage', { - 'instance': plugins.NotifyGrowl, - }), - ('growl://growl.colon:', { - 'instance': plugins.NotifyGrowl, - }), - # Exceptions - ('growl://growl.exceptions01', { - 'instance': plugins.NotifyGrowl, - # Throws a series of connection and transfer exceptions when this flag - # is set and tests that we gracfully handle them - 'test_growl_notify_exceptions': True, - }), - ('growl://growl.exceptions02', { - 'instance': plugins.NotifyGrowl, - # Throws a series of connection and transfer exceptions when this flag - # is set and tests that we gracfully handle them - 'test_growl_register_exceptions': True, - }), -) +try: + # Python v3.4+ + from importlib import reload +except ImportError: + try: + # Python v3.0-v3.3 + from imp import reload + except ImportError: + # Python v2.7 + pass -@mock.patch('apprise.plugins.gntp.notifier.GrowlNotifier') +try: + from gntp import errors + + TEST_GROWL_EXCEPTIONS = ( + errors.NetworkError( + 0, 'gntp.ParseError() not handled'), + errors.AuthError( + 0, 'gntp.AuthError() not handled'), + errors.ParseError( + 0, 'gntp.ParseError() not handled'), + errors.UnsupportedError( + 0, 'gntp.UnsupportedError() not handled'), + ) +except ImportError: + # no problem; these tests will be skipped at this point + TEST_GROWL_EXCEPTIONS = tuple() + + +@pytest.mark.skipif('gntp' not in sys.modules, reason="requires gntp") +def test_growl_plugin_import_error(tmpdir): + """ + API: NotifyGrowl Plugin() Import Error + + """ + # This is a really confusing test case; it can probably be done better, + # but this was all I could come up with. Effectively Apprise is will + # still work flawlessly without the gntp dependancy. Since + # gntp is actually required to be installed to run these unit tests + # we need to do some hacky tricks into fooling our test cases that the + # package isn't available. + + # So we create a temporary directory called gntp (simulating the + # library itself) and writing an __init__.py in it that does nothing + # but throw an ImportError exception (simulating that the library + # isn't found). + suite = tmpdir.mkdir("gntp") + suite.join("__init__.py").write('') + module_name = 'gntp' + suite.join("{}.py".format(module_name)).write('raise ImportError()') + + # The second part of the test is to update our PYTHON_PATH to look + # into this new directory first (before looking where the actual + # valid paths are). This will allow us to override 'JUST' the sleekxmpp + # path. + + # Update our path to point to our new test suite + sys.path.insert(0, str(suite)) + + # We need to remove the gntp modules that have already been loaded + # in memory otherwise they'll just be used instead. Python is smart and + # won't go try and reload everything again if it doesn't have to. + for name in list(sys.modules.keys()): + if name.startswith('{}.'.format(module_name)): + del sys.modules[name] + del sys.modules[module_name] + + # The following libraries need to be reloaded to prevent + # TypeError: super(type, obj): obj must be an instance or subtype of type + # This is better explained in this StackOverflow post: + # https://stackoverflow.com/questions/31363311/\ + # any-way-to-manually-fix-operation-of-\ + # super-after-ipython-reload-avoiding-ty + # + reload(sys.modules['apprise.plugins.NotifyGrowl']) + reload(sys.modules['apprise.plugins']) + reload(sys.modules['apprise.Apprise']) + reload(sys.modules['apprise']) + + # This tests that Apprise still works without gntp. + obj = apprise.Apprise.instantiate('growl://growl.server') + + # Growl objects can still be instantiated however + assert obj is not None + + # Notifications won't work because gntp did not load + assert obj.notify( + title='test', body='body', + notify_type=apprise.NotifyType.INFO) is False + + # Tidy-up / restore things to how they were + # Remove our garbage library + os.unlink(str(suite.join("{}.py".format(module_name)))) + + # Remove our custom entry into the path + sys.path.remove(str(suite)) + + # Reload the libraries we care about + reload(sys.modules['apprise.plugins.NotifyGrowl']) + reload(sys.modules['apprise.plugins']) + reload(sys.modules['apprise.Apprise']) + reload(sys.modules['apprise']) + + +@pytest.mark.skipif('gntp' not in sys.modules, reason="requires gntp") +@mock.patch('gntp.notifier.GrowlNotifier') +def test_growl_exception_handling(mock_gntp): + """ + API: NotifyGrowl Exception Handling + """ + + from gntp import errors + + TEST_GROWL_EXCEPTIONS = ( + errors.NetworkError( + 0, 'gntp.ParseError() not handled'), + errors.AuthError( + 0, 'gntp.AuthError() not handled'), + errors.ParseError( + 0, 'gntp.ParseError() not handled'), + errors.UnsupportedError( + 0, 'gntp.UnsupportedError() not handled'), + ) + + mock_notifier = mock.Mock() + mock_gntp.return_value = mock_notifier + mock_notifier.notify.return_value = True + + # First we test the growl.register() function + for exception in TEST_GROWL_EXCEPTIONS: + mock_notifier.register.side_effect = exception + + # instantiate our object + obj = apprise.Apprise.instantiate( + 'growl://growl.server.hostname', suppress_exceptions=False) + + # Verify Growl object was instantiated + assert obj is not None + + # We will fail to send the notification because our registration + # would have failed + assert obj.notify( + title='test', body='body', + notify_type=apprise.NotifyType.INFO) is False + + # Now we test the growl.notify() function + mock_notifier.register.side_effect = None + for exception in TEST_GROWL_EXCEPTIONS: + mock_notifier.notify.side_effect = exception + + # instantiate our object + obj = apprise.Apprise.instantiate( + 'growl://growl.server.hostname', suppress_exceptions=False) + + # Verify Growl object was instantiated + assert obj is not None + + # We will fail to send the notification because of the underlining + # notify() call throws an exception + assert obj.notify( + title='test', body='body', + notify_type=apprise.NotifyType.INFO) is False + + +@pytest.mark.skipif( + 'gntp' not in sys.modules, reason="requires gntp") +@mock.patch('gntp.notifier.GrowlNotifier') def test_growl_plugin(mock_gntp): """ API: NotifyGrowl Plugin() """ + urls = ( + ################################## + # NotifyGrowl + ################################## + ('growl://', { + 'instance': None, + }), + ('growl://:@/', { + 'instance': None + }), + + ('growl://pass@growl.server', { + 'instance': apprise.plugins.NotifyGrowl, + }), + ('growl://ignored:pass@growl.server', { + 'instance': apprise.plugins.NotifyGrowl, + }), + ('growl://growl.server', { + 'instance': apprise.plugins.NotifyGrowl, + # don't include an image by default + 'include_image': False, + }), + ('growl://growl.server?version=1', { + 'instance': apprise.plugins.NotifyGrowl, + }), + # Test sticky flag + ('growl://growl.server?sticky=yes', { + 'instance': apprise.plugins.NotifyGrowl, + }), + ('growl://growl.server?sticky=no', { + 'instance': apprise.plugins.NotifyGrowl, + }), + # Force a failure + ('growl://growl.server?version=1', { + 'instance': apprise.plugins.NotifyGrowl, + 'growl_response': None, + }), + ('growl://growl.server?version=2', { + # don't include an image by default + 'include_image': False, + 'instance': apprise.plugins.NotifyGrowl, + }), + ('growl://growl.server?version=2', { + # don't include an image by default + 'include_image': False, + 'instance': apprise.plugins.NotifyGrowl, + 'growl_response': None, + }), + + # Priorities + ('growl://pass@growl.server?priority=low', { + 'instance': apprise.plugins.NotifyGrowl, + }), + ('growl://pass@growl.server?priority=moderate', { + 'instance': apprise.plugins.NotifyGrowl, + }), + ('growl://pass@growl.server?priority=normal', { + 'instance': apprise.plugins.NotifyGrowl, + }), + ('growl://pass@growl.server?priority=high', { + 'instance': apprise.plugins.NotifyGrowl, + }), + ('growl://pass@growl.server?priority=emergency', { + 'instance': apprise.plugins.NotifyGrowl, + }), + + # Invalid Priorities + ('growl://pass@growl.server?priority=invalid', { + 'instance': apprise.plugins.NotifyGrowl, + }), + ('growl://pass@growl.server?priority=', { + 'instance': apprise.plugins.NotifyGrowl, + }), + + # invalid version + ('growl://growl.server?version=', { + 'instance': apprise.plugins.NotifyGrowl, + }), + ('growl://growl.server?version=crap', { + 'instance': apprise.plugins.NotifyGrowl, + }), + + # Ports + ('growl://growl.changeport:2000', { + 'instance': apprise.plugins.NotifyGrowl, + }), + ('growl://growl.garbageport:garbage', { + 'instance': apprise.plugins.NotifyGrowl, + }), + ('growl://growl.colon:', { + 'instance': apprise.plugins.NotifyGrowl, + }), + ) + # iterate over our dictionary and test it out - for (url, meta) in TEST_URLS: + for (url, meta) in urls: # Our expected instance instance = meta.get('instance', None) @@ -162,54 +322,15 @@ def test_growl_plugin(mock_gntp): growl_response = meta.get( 'growl_response', True if response else False) - test_growl_notify_exceptions = meta.get( - 'test_growl_notify_exceptions', False) - - test_growl_register_exceptions = meta.get( - 'test_growl_register_exceptions', False) - mock_notifier = mock.Mock() mock_gntp.return_value = mock_notifier + mock_notifier.notify.side_effect = None - test_growl_exceptions = ( - plugins.gntp.errors.NetworkError( - 0, 'gntp.ParseError() not handled'), - plugins.gntp.errors.AuthError( - 0, 'gntp.AuthError() not handled'), - plugins.gntp.errors.UnsupportedError( - 'gntp.UnsupportedError() not handled'), - ) - - if test_growl_notify_exceptions is True: - # Store oure exceptions - test_growl_notify_exceptions = test_growl_exceptions - - elif test_growl_register_exceptions is True: - # Store oure exceptions - test_growl_register_exceptions = test_growl_exceptions - - for exception in test_growl_register_exceptions: - mock_notifier.register.side_effect = exception - try: - obj = Apprise.instantiate(url, suppress_exceptions=False) - - except TypeError: - # This is the response we expect - assert True - - except Exception: - # We can't handle this exception type - assert False - - # We're done this part of the test - continue - - else: - # Store our response - mock_notifier.notify.return_value = growl_response + # Store our response + mock_notifier.notify.return_value = growl_response try: - obj = Apprise.instantiate(url, suppress_exceptions=False) + obj = apprise.Apprise.instantiate(url, suppress_exceptions=False) assert exception is None @@ -223,7 +344,7 @@ def test_growl_plugin(mock_gntp): assert isinstance(obj, instance) is True - if isinstance(obj, plugins.NotifyBase): + if isinstance(obj, apprise.plugins.NotifyBase): # We loaded okay; now lets make sure we can reverse this url assert isinstance(obj.url(), six.string_types) is True @@ -233,11 +354,11 @@ def test_growl_plugin(mock_gntp): # Instantiate the exact same object again using the URL from # the one that was already created properly - obj_cmp = Apprise.instantiate(obj.url()) + obj_cmp = apprise.Apprise.instantiate(obj.url()) # Our object should be the same instance as what we had # originally expected above. - if not isinstance(obj_cmp, plugins.NotifyBase): + if not isinstance(obj_cmp, apprise.plugins.NotifyBase): # Assert messages are hard to trace back with the way # these tests work. Just printing before throwing our # assertion failure makes things easier to debug later on @@ -253,32 +374,10 @@ def test_growl_plugin(mock_gntp): assert getattr(key, obj) == val try: - if test_growl_notify_exceptions is False: - # check that we're as expected - assert obj.notify( - title='test', body='body', - notify_type=NotifyType.INFO) == response - - else: - for exception in test_growl_notify_exceptions: - mock_notifier.notify.side_effect = exception - 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 - print('%s / %s' % (url, str(e))) - assert False - - except AssertionError: - # Don't mess with these entries - raise + # check that we're as expected + assert obj.notify( + title='test', body='body', + notify_type=apprise.NotifyType.INFO) == response except Exception as e: # Check that we were expecting this exception to happen