mirror of
https://github.com/caronc/apprise.git
synced 2025-01-03 20:49:19 +01:00
Growl notification library rewrite; gntp now external dependency (#252)
This commit is contained in:
Chris Caron
GitHub
parent
f3993b18ae
commit
479218da6a
@ -1,6 +1,6 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
#
|
||||
# Copyright (C) 2019 Chris Caron <lead2gold@gmail.com>
|
||||
# Copyright (C) 2020 Chris Caron <lead2gold@gmail.com>
|
||||
# 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:
|
||||
@ -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()
|
||||
@ -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')
|
||||
@ -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/<version> <messagetype> <encryptionAlgorithmID>[:<ivValue>][ <keyHashAlgorithmID>:<keyHash>.<salt>]
|
||||
GNTP_INFO_LINE = re.compile(
|
||||
r'GNTP/(?P<version>\d+\.\d+) (?P<messagetype>REGISTER|NOTIFY|SUBSCRIBE|\-OK|\-ERROR)' +
|
||||
r' (?P<encryptionAlgorithmID>[A-Z0-9]+(:(?P<ivValue>[A-F0-9]+))?) ?' +
|
||||
r'((?P<keyHashAlgorithmID>[A-Z0-9]+):(?P<keyHash>[A-F0-9]+).(?P<salt>[A-F0-9]+))?\r\n',
|
||||
re.IGNORECASE
|
||||
)
|
||||
|
||||
GNTP_INFO_LINE_SHORT = re.compile(
|
||||
r'GNTP/(?P<version>\d+\.\d+) (?P<messagetype>REGISTER|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')
|
||||
@ -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"
|
||||
@ -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 <http://code.google.com/p/growl/source/browse/Bindings/python/Growl.py>`_
|
||||
|
||||
"""
|
||||
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')
|
||||
@ -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"
|
||||
@ -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'
|
||||
@ -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',
|
||||
]
|
||||
|
||||
@ -9,3 +9,4 @@ cryptography
|
||||
|
||||
# Plugin Dependencies
|
||||
sleekxmpp
|
||||
gntp
|
||||
|
||||
@ -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}
|
||||
|
||||
@ -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 = _
|
||||
|
||||
@ -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
|
||||
|
||||
Loading…
Reference in New Issue
Block a user