diff --git a/apprise/Apprise.py b/apprise/Apprise.py index 9d6b792a..9afaa04f 100644 --- a/apprise/Apprise.py +++ b/apprise/Apprise.py @@ -123,7 +123,7 @@ class Apprise(object): self.add(servers) @staticmethod - def instantiate(url, asset=None, suppress_exceptions=True): + def instantiate(url, asset=None, tag=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. @@ -164,6 +164,9 @@ class Apprise(object): logger.error('Could not parse URL: %s' % url) return None + # Build a list of tags to associate with the newly added notifications + results['tag'] = set(parse_list(tag)) + if suppress_exceptions: try: # Attempt to create an instance of our plugin using the parsed @@ -186,10 +189,16 @@ class Apprise(object): return plugin - def add(self, servers, asset=None): + def add(self, servers, asset=None, tag=None): """ Adds one or more server URLs into our list. + You can override the global asset if you wish by including it with the + server(s) that you add. + + The tag allows you to associate 1 or more tag values to the server(s) + being added. tagging a service allows you to exclusively access them + when calling the notify() function. """ # Initialize our return status @@ -204,12 +213,13 @@ class Apprise(object): self.servers.append(servers) return True + # build our server listings servers = parse_list(servers) for _server in servers: # Instantiate ourselves an object, this function throws or # returns None if it fails - instance = Apprise.instantiate(_server, asset=asset) + instance = Apprise.instantiate(_server, asset=asset, tag=tag) if not instance: return_status = False logging.error( @@ -231,13 +241,18 @@ class Apprise(object): self.servers[:] = [] def notify(self, title, body, notify_type=NotifyType.INFO, - body_format=None): + body_format=None, tag=None): """ Send a notification to all of the plugins previously loaded. If the body_format specified is NotifyFormat.MARKDOWN, it will be converted to HTML if the Notification type expects this. + if the tag is specified (either a string or a set/list/tuple + of strings), then only the notifications flagged with that + tagged value are notified. By default all added services + are notified (tag=None) + """ # Initialize our return result @@ -249,8 +264,63 @@ class Apprise(object): # Tracks conversions conversion_map = dict() + # Build our tag setup + # - top level entries are treated as an 'or' + # - second level (or more) entries are treated as 'and' + # + # examples: + # tag="tagA, tagB" = tagA or tagB + # tag=['tagA', 'tagB'] = tagA or tagB + # tag=[('tagA', 'tagC'), 'tagB'] = (tagA and tagC) or tagB + # tag=[('tagB', 'tagC')] = tagB and tagC + # Iterate over our loaded plugins for server in self.servers: + + if tag is not None: + + if isinstance(tag, (list, tuple, set)): + # using the tags detected; determine if we'll allow the + # notification to be sent or not + matched = False + + # Every entry here will be or'ed with the next + for entry in tag: + if isinstance(entry, (list, tuple, set)): + + # treat these entries as though all elements found + # must exist in the notification service + tags = set(parse_list(entry)) + + if len(tags.intersection( + server.tags)) == len(tags): + # our set contains all of the entries found + # in our notification server object + matched = True + break + + elif entry in server: + # our entr(ies) match what was found in our server + # object. + matched = True + break + + # else: keep looking + + if not matched: + # We did not meet any of our and'ed criteria + continue + + elif tag not in server: + # one or more tags were defined and they didn't match the + # entry in the current service; move along... + continue + + # else: our content was found inside the server, so we're good + + # If our code reaches here, we either did not define a tag (it was + # set to None), or we did define a tag and the logic above + # determined we need to notify the service it's associated with if server.notify_format not in conversion_map: if body_format == NotifyFormat.MARKDOWN and \ server.notify_format == NotifyFormat.HTML: diff --git a/apprise/plugins/NotifyBase.py b/apprise/plugins/NotifyBase.py index 87b794ff..d878ea2a 100644 --- a/apprise/plugins/NotifyBase.py +++ b/apprise/plugins/NotifyBase.py @@ -40,6 +40,7 @@ except ImportError: from ..utils import parse_url from ..utils import parse_bool +from ..utils import parse_list from ..utils import is_hostname from ..common import NOTIFY_TYPES from ..common import NotifyFormat @@ -127,6 +128,9 @@ class NotifyBase(object): # Default Notify Format notify_format = NotifyFormat.TEXT + # Maintain a set of tags to associate with this specific notification + tags = set() + # Logging logger = logging.getLogger(__name__) @@ -172,6 +176,12 @@ class NotifyBase(object): # Provide override self.notify_format = notify_format + if 'tag' in kwargs: + # We want to associate some tags with our notification service. + # the code below gets the 'tag' argument if defined, otherwise + # it just falls back to whatever was already defined globally + self.tags = set(parse_list(kwargs.get('tag', self.tags))) + def throttle(self, throttle_time=None): """ A common throttle control @@ -249,6 +259,19 @@ class NotifyBase(object): color_type=color_type, ) + def __contains__(self, tags): + """ + Returns true if the tag specified is associated with this notification. + + tag can also be a tuple, set, and/or list + + """ + if isinstance(tags, (tuple, set, list)): + return bool(set(tags) & self.tags) + + # return any match + return tags in self.tags + @property def app_id(self): return self.asset.app_id diff --git a/test/test_api.py b/test/test_api.py index c903d76a..fa6ffada 100644 --- a/test/test_api.py +++ b/test/test_api.py @@ -38,6 +38,8 @@ from apprise import NotifyImageSize from apprise import __version__ from apprise.Apprise import __load_matrix import pytest +import requests +import mock def test_apprise(): @@ -110,7 +112,7 @@ def test_apprise(): class BadNotification(NotifyBase): def __init__(self, **kwargs): - super(BadNotification, self).__init__() + super(BadNotification, self).__init__(**kwargs) # We fail whenever we're initialized raise TypeError() @@ -118,7 +120,7 @@ def test_apprise(): class GoodNotification(NotifyBase): def __init__(self, **kwargs): super(GoodNotification, self).__init__( - notify_format=NotifyFormat.HTML) + notify_format=NotifyFormat.HTML, **kwargs) def notify(self, **kwargs): # Pretend everything is okay @@ -203,9 +205,20 @@ def test_apprise(): assert(len(a) == 0) # Instantiate a good object - plugin = a.instantiate('good://localhost') + plugin = a.instantiate('good://localhost', tag="good") assert(isinstance(plugin, NotifyBase)) + # Test simple tagging inside of the object + assert("good" in plugin) + assert("bad" not in plugin) + + # the in (__contains__ override) is based on or'ed content; so although + # 'bad' isn't tagged as being in the plugin, 'good' is, so the return + # value of this is True + assert(["bad", "good"] in plugin) + assert(set(["bad", "good"]) in plugin) + assert(("bad", "good") in plugin) + # We an add already substatiated instances into our Apprise object a.add(plugin) assert(len(a) == 1) @@ -225,6 +238,106 @@ def test_apprise(): assert(len(a) == 0) +@mock.patch('requests.get') +@mock.patch('requests.post') +def test_apprise_tagging(mock_post, mock_get): + """ + API: Apprise() object tagging functionality + + """ + + # A request + robj = mock.Mock() + setattr(robj, 'raw', mock.Mock()) + # Allow raw.read() calls + robj.raw.read.return_value = '' + robj.text = '' + robj.content = '' + mock_get.return_value = robj + mock_post.return_value = robj + + # Simulate a successful notification + mock_get.return_value.status_code = requests.codes.ok + mock_post.return_value.status_code = requests.codes.ok + + # Create our object + a = Apprise() + + # Add entry and assign it to a tag called 'awesome' + assert(a.add('json://localhost/path1/', tag='awesome') is True) + + # Add another notification and assign it to a tag called 'awesome' + # and another tag called 'local' + assert(a.add('json://localhost/path2/', tag=['mmost', 'awesome']) is True) + + # notify the awesome tag; this would notify both services behind the + # scenes + assert(a.notify(title="my title", body="my body", tag='awesome') is True) + + # notify all of the tags + assert(a.notify( + title="my title", body="my body", tag=['awesome', 'mmost']) is True) + + # there is nothing to notify using tags 'missing'. However we intentionally + # don't fail as there is value in identifying a tag that simply have + # nothing to notify from while the object itself contains items + assert(a.notify( + title="my title", body="my body", tag='missing') is True) + + # Now to test the ability to and and/or notifications + a = Apprise() + + # Add a tag by tuple + assert(a.add('json://localhost/tagA/', tag=("TagA", )) is True) + # Add 2 tags by string + assert(a.add('json://localhost/tagAB/', tag="TagA, TagB") is True) + # Add a tag using a set + assert(a.add('json://localhost/tagB/', tag=set(["TagB"])) is True) + # Add a tag by string (again) + assert(a.add('json://localhost/tagC/', tag="TagC") is True) + # Add 2 tags using a list + assert(a.add('json://localhost/tagCD/', tag=["TagC", "TagD"]) is True) + # Add a tag by string (again) + assert(a.add('json://localhost/tagD/', tag="TagD") is True) + # add a tag set by set (again) + assert(a.add('json://localhost/tagCDE/', + tag=set(["TagC", "TagD", "TagE"])) is True) + + # Expression: TagC and TagD + # Matches the following only: + # - json://localhost/tagCD/ + # - json://localhost/tagCDE/ + assert(a.notify( + title="my title", body="my body", tag=[('TagC', 'TagD')]) is True) + + # Expression: (TagY and TagZ) or TagX + # Matches nothing + assert(a.notify( + title="my title", body="my body", + tag=[('TagY', 'TagZ'), 'TagX']) is True) + + # Expression: (TagY and TagZ) or TagA + # Matches the following only: + # - json://localhost/tagAB/ + assert(a.notify( + title="my title", body="my body", + tag=[('TagY', 'TagZ'), 'TagA']) is True) + + # Expression: (TagE and TagD) or TagB + # Matches the following only: + # - json://localhost/tagCDE/ + # - json://localhost/tagAB/ + # - json://localhost/tagB/ + assert(a.notify( + title="my title", body="my body", + tag=[('TagE', 'TagD'), 'TagB']) is True) + + # Garbage Entries + assert(a.notify( + title="my title", body="my body", + tag=[(object, ), ]) is True) + + def test_apprise_notify_formats(tmpdir): """ API: Apprise() TextFormat tests