extern/apprise
1
0
Fork 0
mirror of https://github.com/caronc/apprise.git synced 2024-11-08 17:24:00 +01:00
Code Issues Packages Projects Releases Wiki Activity

more refactoring and unittest completion

Browse Source
This commit is contained in:
Chris Caron 2017-12-03 15:44:39 -05:00
parent 9c4502a18b
commit a65a8a8373
5 changed files with 141 additions and 90 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
.travis.yml
View File

@ -1,10 +1,3 @@
dist: trusty
sudo: false
cache:
directories:
- $HOME/.cache/pip
language: python

132
apprise/Apprise.py
View File

@ -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

3
apprise/__init__.py
View File

@ -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',

14
apprise/plugins/NotifyBase.py
View File

@ -38,27 +38,16 @@ 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
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)
HTTP_ERROR_MAP = {
400: 'Bad Request - Unsupported Parameters.',
@ -344,7 +333,6 @@ 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))

45
test/test_api.py
View File

@ -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',