diff --git a/.travis.yml b/.travis.yml index ec96f1ec..95635ed5 100644 --- a/.travis.yml +++ b/.travis.yml @@ -1,10 +1,3 @@ -dist: trusty -sudo: false -cache: - directories: - - $HOME/.cache/pip - - language: python diff --git a/apprise/Apprise.py b/apprise/Apprise.py index eb8c4f48..7a162bf6 100644 --- a/apprise/Apprise.py +++ b/apprise/Apprise.py @@ -28,6 +28,7 @@ from .utils import compat_is_basestring from .AppriseAsset import AppriseAsset +from . import NotifyBase from . import plugins logger = logging.getLogger(__name__) @@ -111,6 +112,70 @@ class Apprise(object): if servers: self.add(servers) + @staticmethod + def instantiate(url, asset=None, suppress_exceptions=True): + """ + Returns the instance of a instantiated plugin based on the provided + Server URL. If the url fails to be parsed, then None is returned. + + """ + # swap hash (#) tag values with their html version + # This is useful for accepting channels (as arguments to pushbullet) + _url = url.replace('/#', '/%23') + + # Attempt to acquire the schema at the very least to allow our plugins + # to determine if they can make a better interpretation of a URL + # geared for them anyway. + schema = GET_SCHEMA_RE.match(_url) + if schema is None: + logger.error('%s is an unparseable server url.' % url) + return None + + # Update the schema + schema = schema.group('schema').lower() + + # Some basic validation + if schema not in SCHEMA_MAP: + logger.error( + '{0} is not a supported server type (url={1}).'.format( + schema, + _url, + ) + ) + return None + + # Parse our url details + # the server object is a dictionary containing all of the information + # parsed from our URL + results = SCHEMA_MAP[schema].parse_url(_url) + + if not results: + # Failed to parse the server URL + logger.error('Could not parse URL: %s' % url) + return None + + if suppress_exceptions: + try: + # Attempt to create an instance of our plugin using the parsed + # URL information + plugin = SCHEMA_MAP[results['schema']](**results) + + except: + # the arguments are invalid or can not be used. + logger.error('Could not load URL: %s' % url) + return None + + else: + # Attempt to create an instance of our plugin using the parsed + # URL information but don't wrap it in a try catch + plugin = SCHEMA_MAP[results['schema']](**results) + + # Save our asset + if asset: + plugin.asset = asset + + return plugin + def add(self, servers, asset=None): """ Adds one or more server URLs into our list. @@ -120,66 +185,27 @@ class Apprise(object): # Initialize our return status return_status = True + if asset is None: + # prepare default asset + asset = self.asset + + if isinstance(servers, NotifyBase): + # Go ahead and just add our plugin into our list + self.servers.append(servers) + return True + servers = parse_list(servers) for _server in servers: - # swap hash (#) tag values with their html version - # This is useful for accepting channels (as arguments to - # pushbullet) - _server = _server.replace('/#', '/%23') - - # Attempt to acquire the schema at the very least to allow - # our plugins to determine if they can make a better - # interpretation of a URL geared for them anyway. - schema = GET_SCHEMA_RE.match(_server) - if schema is None: - logger.error( - '%s is an unparseable server url.' % _server, - ) + # Instantiate ourselves an object, this function throws or + # returns None if it fails + instance = Apprise.instantiate(_server, asset=asset) + if not instance: return_status = False continue - # Update the schema - schema = schema.group('schema').lower() - - # Some basic validation - if schema not in SCHEMA_MAP: - logger.error( - '%s is not a supported server type.' % schema, - ) - return_status = False - continue - - # Parse our url details - # the server object is a dictionary containing all of the - # information parsed from our URL - results = SCHEMA_MAP[schema].parse_url(_server) - - if not results: - # Failed to parse the server URL - logger.error('Could not parse URL: %s' % _server) - return_status = False - continue - - try: - # Attempt to create an instance of our plugin using the parsed - # URL information - plugin = SCHEMA_MAP[results['schema']](**results) - - except: - # the arguments are invalid or can not be used. - return_status = False - continue - - # Save our asset - if asset: - plugin.asset = asset - - else: - plugin.asset = self.asset - # Add our initialized plugin to our server listings - self.servers.append(plugin) + self.servers.append(instance) # Return our status return return_status diff --git a/apprise/__init__.py b/apprise/__init__.py index ba7b3d76..3331a008 100644 --- a/apprise/__init__.py +++ b/apprise/__init__.py @@ -27,6 +27,7 @@ from .common import NOTIFY_TYPES from .common import NOTIFY_IMAGE_SIZES from .common import NotifyImageSize from .plugins.NotifyBase import NotifyFormat +from .plugins.NotifyBase import NotifyBase from .Apprise import Apprise from .AppriseAsset import AppriseAsset @@ -38,7 +39,7 @@ logging.getLogger(__name__).addHandler(NullHandler()) __all__ = [ # Core - 'Apprise', 'AppriseAsset', + 'Apprise', 'AppriseAsset', 'NotifyBase', # Reference 'NotifyType', 'NotifyImageSize', 'NotifyFormat', 'NOTIFY_TYPES', diff --git a/apprise/plugins/NotifyBase.py b/apprise/plugins/NotifyBase.py index 5bfa39fe..53ec9c7c 100644 --- a/apprise/plugins/NotifyBase.py +++ b/apprise/plugins/NotifyBase.py @@ -38,26 +38,15 @@ from ..common import NOTIFY_TYPES from ..AppriseAsset import AppriseAsset -# Define a general HTML Escaping -try: - # use sax first because it's faster - from xml.sax.saxutils import escape as sax_escape +# use sax first because it's faster +from xml.sax.saxutils import escape as sax_escape - def _escape(text): - """ - saxutil escape tool - """ - return sax_escape(text, {"'": "'", "\"": """}) -except ImportError: - # if we can't, then fall back to cgi escape - from cgi import escape as cgi_escape - - def _escape(text): - """ - cgi escape tool - """ - return cgi_escape(text, quote=True) +def _escape(text): + """ + saxutil escape tool + """ + return sax_escape(text, {"'": "'", "\"": """}) HTTP_ERROR_MAP = { @@ -344,17 +333,16 @@ class NotifyBase(object): # Support SSL Certificate 'verify' keyword. Default to being enabled results['verify'] = True - if 'qsd' in results: - if 'verify' in results['qsd']: - results['verify'] = parse_bool( - results['qsd'].get('verify', True)) + if 'verify' in results['qsd']: + results['verify'] = parse_bool( + results['qsd'].get('verify', True)) - # Password overrides - if 'pass' in results['qsd']: - results['password'] = results['qsd']['pass'] + # Password overrides + if 'pass' in results['qsd']: + results['password'] = results['qsd']['pass'] - # User overrides - if 'user' in results['qsd']: - results['user'] = results['qsd']['user'] + # User overrides + if 'user' in results['qsd']: + results['user'] = results['qsd']['user'] return results diff --git a/test/test_api.py b/test/test_api.py index 0425a504..88b4ce00 100644 --- a/test/test_api.py +++ b/test/test_api.py @@ -23,9 +23,10 @@ from os.path import dirname from apprise import Apprise from apprise import AppriseAsset from apprise.Apprise import SCHEMA_MAP -from apprise.plugins.NotifyBase import NotifyBase +from apprise import NotifyBase from apprise import NotifyType from apprise import NotifyImageSize +from apprise.Apprise import __load_matrix def test_apprise(): @@ -33,6 +34,11 @@ def test_apprise(): API: Apprise() object """ + # Caling load matix a second time which is an internal function causes it + # to skip over content already loaded into our matrix and thefore accesses + # other if/else parts of the code that aren't otherwise called + __load_matrix() + a = Apprise() # no items @@ -165,12 +171,49 @@ def test_apprise(): # simply returns False assert(a.notify(title="present", body="present") is False) + # Test instantiating a plugin + class ThrowInstantiateNotification(NotifyBase): + def __init__(self, **kwargs): + # Pretend everything is okay + raise TypeError() + SCHEMA_MAP['throw'] = ThrowInstantiateNotification + + # Reset our object + a.clear() + assert(len(a) == 0) + + # Instantiate a good object + plugin = a.instantiate('good://localhost') + assert(isinstance(plugin, NotifyBase)) + + # We an add already substatiated instances into our Apprise object + a.add(plugin) + assert(len(a) == 1) + + # Reset our object again + a.clear() + try: + a.instantiate('throw://localhost', suppress_exceptions=False) + assert(False) + + except TypeError: + assert(True) + assert(len(a) == 0) + + assert(a.instantiate( + 'throw://localhost', suppress_exceptions=True) is None) + assert(len(a) == 0) + def test_apprise_asset(tmpdir): """ API: AppriseAsset() object """ + a = AppriseAsset(theme=None) + # Default theme + assert(a.theme == 'default') + a = AppriseAsset( theme='dark', image_path_mask='/{THEME}/{TYPE}-{XY}.png',