diff --git a/README.md b/README.md
index 3127f114..57766fac 100644
--- a/README.md
+++ b/README.md
@@ -39,7 +39,7 @@ The table below identifies the services this tool supports and some example serv
| [Faast](https://github.com/caronc/apprise/wiki/Notify_faast) | faast:// | (TCP) 443 | faast://authorizationtoken
| [Gnome](https://github.com/caronc/apprise/wiki/Notify_gnome) | gnome:// | n/a | gnome://
| [Growl](https://github.com/caronc/apprise/wiki/Notify_growl) | growl:// | (UDP) 23053 | growl://hostname
growl://hostname:portno
growl://password@hostname
growl://password@hostname:port**Note**: you can also use the get parameter _version_ which can allow the growl request to behave using the older v1.x protocol. An example would look like: growl://hostname?version=1
-| [IFTTT](https://github.com/caronc/apprise/wiki/Notify_ifttt) | ifttt:// | (TCP) 443 | ifttt://webhooksID/EventToTrigger
ifttt://webhooksID/EventToTrigger/Value1/Value2/Value3
ifttt://webhooksID/EventToTrigger/?Value3=NewEntry&Value2=AnotherValue
+| [IFTTT](https://github.com/caronc/apprise/wiki/Notify_ifttt) | ifttt:// | (TCP) 443 | ifttt://webhooksID/EventToTrigger
ifttt://webhooksID/EventToTrigger/Event2/EventN
| [Join](https://github.com/caronc/apprise/wiki/Notify_join) | join:// | (TCP) 443 | join://apikey/device
join://apikey/device1/device2/deviceN/
join://apikey/group
join://apikey/groupA/groupB/groupN
join://apikey/DeviceA/groupA/groupN/DeviceN/
| [KODI](https://github.com/caronc/apprise/wiki/Notify_kodi) | kodi:// or kodis:// | (TCP) 8080 or 443 | kodi://hostname
kodi://user@hostname
kodi://user:password@hostname:port
| [Matrix](https://github.com/caronc/apprise/wiki/Notify_matrix) | matrix:// or matrixs:// | (TCP) 80 or 443 | matrix://token
matrix://user@token
matrixs://token?mode=slack
matrixs://user@token
diff --git a/apprise/Apprise.py b/apprise/Apprise.py
index 576f1d54..9c9b81de 100644
--- a/apprise/Apprise.py
+++ b/apprise/Apprise.py
@@ -171,7 +171,7 @@ class Apprise(object):
# URL information
plugin = SCHEMA_MAP[results['schema']](**results)
- except:
+ except Exception:
# the arguments are invalid or can not be used.
logger.error('Could not load URL: %s' % url)
return None
@@ -432,6 +432,33 @@ class Apprise(object):
return response
+ def urls(self):
+ """
+ Returns all of the loaded URLs defined in this apprise object.
+ """
+ return [x.url() for x in self.servers]
+
+ def pop(self, index):
+ """
+ Removes an indexed Notification Service from the stack and
+ returns it.
+ """
+
+ # Remove our entry
+ return self.servers.pop(index)
+
+ def __getitem__(self, index):
+ """
+ Returns the indexed server entry of a loaded notification server
+ """
+ return self.servers[index]
+
+ def __iter__(self):
+ """
+ Returns an iterator to our server list
+ """
+ return iter(self.servers)
+
def __len__(self):
"""
Returns the number of servers loaded
diff --git a/apprise/plugins/NotifyBase.py b/apprise/plugins/NotifyBase.py
index a8471f83..9e05a908 100644
--- a/apprise/plugins/NotifyBase.py
+++ b/apprise/plugins/NotifyBase.py
@@ -260,6 +260,14 @@ class NotifyBase(object):
color_type=color_type,
)
+ def url(self):
+ """
+ Assembles the URL associated with the notification based on the
+ arguments provied.
+
+ """
+ raise NotImplementedError("This is implimented by the child class.")
+
def __contains__(self, tags):
"""
Returns true if the tag specified is associated with this notification.
@@ -347,16 +355,20 @@ class NotifyBase(object):
"""
common urlencode function
+ The query should always be a dictionary.
+
"""
+ # Tidy query by eliminating any records set to None
+ _query = {k: v for (k, v) in query.items() if v is not None}
try:
# Python v3.x
return _urlencode(
- query, doseq=doseq, safe=safe, encoding=encoding,
+ _query, doseq=doseq, safe=safe, encoding=encoding,
errors=errors)
except TypeError:
# Python v2.7
- return _urlencode(query)
+ return _urlencode(_query)
@staticmethod
def split_path(path, unquote=True):
diff --git a/apprise/plugins/NotifyBoxcar.py b/apprise/plugins/NotifyBoxcar.py
index 58891b76..dd55a3a3 100644
--- a/apprise/plugins/NotifyBoxcar.py
+++ b/apprise/plugins/NotifyBoxcar.py
@@ -29,6 +29,7 @@ import re
from time import time
import hmac
from hashlib import sha1
+from itertools import chain
try:
from urlparse import urlparse
@@ -272,6 +273,26 @@ class NotifyBoxcar(NotifyBase):
return True
+ def url(self):
+ """
+ Returns the URL built dynamically based on specified arguments.
+ """
+
+ # Define any arguments set
+ args = {
+ 'format': self.notify_format
+ }
+
+ return '{schema}://{access}/{secret}/{recipients}/?{args}'.format(
+ schema=self.secure_protocol,
+ access=self.quote(self.access),
+ secret=self.quote(self.secret),
+ recipients='/'.join([
+ self.quote(x) for x in chain(
+ self.tags, self.device_tokens) if x != DEFAULT_TAG]),
+ args=self.urlencode(args),
+ )
+
@staticmethod
def parse_url(url):
"""
diff --git a/apprise/plugins/NotifyDBus.py b/apprise/plugins/NotifyDBus.py
index 0517f1b6..8083a4a6 100644
--- a/apprise/plugins/NotifyDBus.py
+++ b/apprise/plugins/NotifyDBus.py
@@ -287,6 +287,13 @@ class NotifyDBus(NotifyBase):
return True
+ def url(self):
+ """
+ Returns the URL built dynamically based on specified arguments.
+ """
+
+ return '{schema}://'.format(schema=self.schema)
+
@staticmethod
def parse_url(url):
"""
diff --git a/apprise/plugins/NotifyDiscord.py b/apprise/plugins/NotifyDiscord.py
index 7cdf283c..2ce78bd0 100644
--- a/apprise/plugins/NotifyDiscord.py
+++ b/apprise/plugins/NotifyDiscord.py
@@ -180,8 +180,8 @@ class NotifyDiscord(NotifyBase):
else:
# not markdown
- payload['content'] = body if not title \
- else "{}\r\n{}".format(title, body)
+ payload['content'] = \
+ body if not title else "{}\r\n{}".format(title, body)
if self.avatar and image_url:
payload['avatar_url'] = image_url
@@ -241,6 +241,27 @@ class NotifyDiscord(NotifyBase):
return True
+ def url(self):
+ """
+ Returns the URL built dynamically based on specified arguments.
+ """
+
+ # Define any arguments set
+ args = {
+ 'format': self.notify_format,
+ 'tts': 'yes' if self.tts else 'no',
+ 'avatar': 'yes' if self.avatar else 'no',
+ 'footer': 'yes' if self.footer else 'no',
+ 'thumbnail': 'yes' if self.thumbnail else 'no',
+ }
+
+ return '{schema}://{webhook_id}/{webhook_token}/?{args}'.format(
+ schema=self.secure_protocol,
+ webhook_id=self.quote(self.webhook_id),
+ webhook_token=self.quote(self.webhook_token),
+ args=self.urlencode(args),
+ )
+
@staticmethod
def parse_url(url):
"""
diff --git a/apprise/plugins/NotifyEmail.py b/apprise/plugins/NotifyEmail.py
index 69d836c8..bdc29a2d 100644
--- a/apprise/plugins/NotifyEmail.py
+++ b/apprise/plugins/NotifyEmail.py
@@ -421,6 +421,52 @@ class NotifyEmail(NotifyBase):
return True
+ def url(self):
+ """
+ Returns the URL built dynamically based on specified arguments.
+ """
+
+ # Define any arguments set
+ args = {
+ 'format': self.notify_format,
+ 'to': self.to_addr,
+ 'from': self.from_addr,
+ 'name': self.from_name,
+ 'mode': self.secure_mode,
+ 'smtp': self.smtp_host,
+ 'timeout': self.timeout,
+ 'user': self.user,
+ }
+
+ # pull email suffix from username (if present)
+ user = self.user.split('@')[0]
+
+ # Determine Authentication
+ auth = ''
+ if self.user and self.password:
+ auth = '{user}:{password}@'.format(
+ user=self.quote(user, safe=''),
+ password=self.quote(self.password, safe=''),
+ )
+ else:
+ # user url
+ auth = '{user}@'.format(
+ user=self.quote(user, safe=''),
+ )
+
+ # Default Port setup
+ default_port = \
+ self.default_secure_port if self.secure else self.default_port
+
+ return '{schema}://{auth}{hostname}{port}/?{args}'.format(
+ schema=self.secure_protocol if self.secure else self.protocol,
+ auth=auth,
+ hostname=self.host,
+ port='' if self.port is None or self.port == default_port
+ else ':{}'.format(self.port),
+ args=self.urlencode(args),
+ )
+
@staticmethod
def parse_url(url):
"""
diff --git a/apprise/plugins/NotifyEmby.py b/apprise/plugins/NotifyEmby.py
index 02093078..4a514ee5 100644
--- a/apprise/plugins/NotifyEmby.py
+++ b/apprise/plugins/NotifyEmby.py
@@ -535,6 +535,38 @@ class NotifyEmby(NotifyBase):
return not has_error
+ def url(self):
+ """
+ Returns the URL built dynamically based on specified arguments.
+ """
+
+ # Define any arguments set
+ args = {
+ 'format': self.notify_format,
+ 'modal': 'yes' if self.modal else 'no',
+ }
+
+ # Determine Authentication
+ auth = ''
+ if self.user and self.password:
+ auth = '{user}:{password}@'.format(
+ user=self.quote(self.user, safe=''),
+ password=self.quote(self.password, safe=''),
+ )
+ else: # self.user is set
+ auth = '{user}@'.format(
+ user=self.quote(self.user, safe=''),
+ )
+
+ return '{schema}://{auth}{hostname}{port}/?{args}'.format(
+ schema=self.secure_protocol if self.secure else self.protocol,
+ auth=auth,
+ hostname=self.host,
+ port='' if self.port is None or self.port == self.emby_default_port
+ else ':{}'.format(self.port),
+ args=self.urlencode(args),
+ )
+
@property
def is_authenticated(self):
"""
diff --git a/apprise/plugins/NotifyFaast.py b/apprise/plugins/NotifyFaast.py
index cabee82b..5fffd724 100644
--- a/apprise/plugins/NotifyFaast.py
+++ b/apprise/plugins/NotifyFaast.py
@@ -124,6 +124,22 @@ class NotifyFaast(NotifyBase):
return True
+ def url(self):
+ """
+ Returns the URL built dynamically based on specified arguments.
+ """
+
+ # Define any arguments set
+ args = {
+ 'format': self.notify_format,
+ }
+
+ return '{schema}://{authtoken}/?{args}'.format(
+ schema=self.protocol,
+ authtoken=self.quote(self.authtoken, safe=''),
+ args=self.urlencode(args),
+ )
+
@staticmethod
def parse_url(url):
"""
diff --git a/apprise/plugins/NotifyGnome.py b/apprise/plugins/NotifyGnome.py
index 7f9747e5..a0f20fcc 100644
--- a/apprise/plugins/NotifyGnome.py
+++ b/apprise/plugins/NotifyGnome.py
@@ -164,6 +164,13 @@ class NotifyGnome(NotifyBase):
return True
+ def url(self):
+ """
+ Returns the URL built dynamically based on specified arguments.
+ """
+
+ return '{schema}://'.format(schema=self.protocol)
+
@staticmethod
def parse_url(url):
"""
diff --git a/apprise/plugins/NotifyGrowl/NotifyGrowl.py b/apprise/plugins/NotifyGrowl/NotifyGrowl.py
index e478998b..b88cbb4d 100644
--- a/apprise/plugins/NotifyGrowl/NotifyGrowl.py
+++ b/apprise/plugins/NotifyGrowl/NotifyGrowl.py
@@ -207,6 +207,43 @@ class NotifyGrowl(NotifyBase):
return True
+ def url(self):
+ """
+ Returns the URL built dynamically based on specified arguments.
+ """
+
+ _map = {
+ GrowlPriority.LOW: 'low',
+ GrowlPriority.MODERATE: 'moderate',
+ GrowlPriority.NORMAL: 'normal',
+ GrowlPriority.HIGH: 'high',
+ GrowlPriority.EMERGENCY: 'emergency',
+ }
+
+ # Define any arguments set
+ args = {
+ 'format': self.notify_format,
+ 'priority':
+ _map[GrowlPriority.NORMAL] if self.priority not in _map
+ else _map[self.priority],
+ 'version': self.version,
+ }
+
+ auth = ''
+ if self.password:
+ auth = '{password}@'.format(
+ password=self.quote(self.user, safe=''),
+ )
+
+ return '{schema}://{auth}{hostname}{port}/?{args}'.format(
+ schema=self.secure_protocol if self.secure else self.protocol,
+ auth=auth,
+ hostname=self.host,
+ port='' if self.port is None or self.port == self.default_port
+ else ':{}'.format(self.port),
+ args=self.urlencode(args),
+ )
+
@staticmethod
def parse_url(url):
"""
@@ -239,15 +276,10 @@ class NotifyGrowl(NotifyBase):
if 'priority' in results['qsd'] and len(results['qsd']['priority']):
_map = {
'l': GrowlPriority.LOW,
- '-2': GrowlPriority.LOW,
'm': GrowlPriority.MODERATE,
- '-1': GrowlPriority.MODERATE,
'n': GrowlPriority.NORMAL,
- '0': GrowlPriority.NORMAL,
'h': GrowlPriority.HIGH,
- '1': GrowlPriority.HIGH,
'e': GrowlPriority.EMERGENCY,
- '2': GrowlPriority.EMERGENCY,
}
try:
results['priority'] = \
diff --git a/apprise/plugins/NotifyIFTTT.py b/apprise/plugins/NotifyIFTTT.py
index 45a7782b..2f0bad50 100644
--- a/apprise/plugins/NotifyIFTTT.py
+++ b/apprise/plugins/NotifyIFTTT.py
@@ -34,7 +34,7 @@
# URL. For example, it might look like this:
# https://maker.ifttt.com/use/a3nHB7gA9TfBQSqJAHklod
#
-# In the above example a3nHB7gA9TfBQSqJAHklod becomes your {apikey}
+# In the above example a3nHB7gA9TfBQSqJAHklod becomes your {webhook_id}
# You will need this to make this notification work correctly
#
# For each event you create you will assign it a name (this will be known as
@@ -44,6 +44,7 @@ import requests
from json import dumps
from .NotifyBase import NotifyBase
from .NotifyBase import HTTP_ERROR_MAP
+from ..utils import parse_list
class NotifyIFTTT(NotifyBase):
@@ -59,7 +60,7 @@ class NotifyIFTTT(NotifyBase):
service_url = 'https://ifttt.com/'
# The default protocol
- protocol = 'ifttt'
+ secure_protocol = 'ifttt'
# A URL that takes you to the setup/help of the specific protocol
setup_url = 'https://github.com/caronc/apprise/wiki/Notify_ifttt'
@@ -87,35 +88,28 @@ class NotifyIFTTT(NotifyBase):
ifttt_default_type_key = 'value3'
# IFTTT uses the http protocol with JSON requests
- notify_url = 'https://maker.ifttt.com/trigger/{event}/with/key/{apikey}'
+ notify_url = 'https://maker.ifttt.com/' \
+ 'trigger/{event}/with/key/{webhook_id}'
- def __init__(self, apikey, event, event_args=None, **kwargs):
+ def __init__(self, webhook_id, events, **kwargs):
"""
Initialize IFTTT Object
"""
super(NotifyIFTTT, self).__init__(**kwargs)
- if not apikey:
- raise TypeError('You must specify the Webhooks apikey.')
+ if not webhook_id:
+ raise TypeError('You must specify the Webhooks webhook_id.')
- if not event:
- raise TypeError('You must specify the Event you wish to trigger.')
+ # Store our Events we wish to trigger
+ self.events = parse_list(events)
+
+ if not self.events:
+ raise TypeError(
+ 'You must specify at least one event you wish to trigger on.')
# Store our APIKey
- self.apikey = apikey
-
- # Store our Event we wish to trigger
- self.event = event
-
- if isinstance(event_args, dict):
- # Make a copy of the arguments so that they can't change
- # outside of this plugin
- self.event_args = event_args.copy()
-
- else:
- # Force a dictionary
- self.event_args = dict()
+ self.webhook_id = webhook_id
def notify(self, title, body, notify_type, **kwargs):
"""
@@ -134,72 +128,101 @@ class NotifyIFTTT(NotifyBase):
self.ifttt_default_type_key: notify_type,
}
- # Update our payload using any other event_args specified
- payload.update(self.event_args)
-
# Eliminate empty fields; users wishing to cancel the use of the
# self.ifttt_default_ entries can preset these keys to being
# empty so that they get caught here and removed.
payload = {x: y for x, y in payload.items() if y}
- # URL to transmit content via
- url = self.notify_url.format(
- apikey=self.apikey,
- event=self.event,
+ # Track our failures
+ error_count = 0
+
+ # Create a copy of our event lit
+ events = list(self.events)
+
+ while len(events):
+
+ # Retrive an entry off of our event list
+ event = events.pop(0)
+
+ # URL to transmit content via
+ url = self.notify_url.format(
+ webhook_id=self.webhook_id,
+ event=event,
+ )
+
+ self.logger.debug('IFTTT POST URL: %s (cert_verify=%r)' % (
+ url, self.verify_certificate,
+ ))
+ self.logger.debug('IFTTT Payload: %s' % str(payload))
+ try:
+ r = requests.post(
+ url,
+ data=dumps(payload),
+ headers=headers,
+ verify=self.verify_certificate,
+ )
+ self.logger.debug(
+ u"IFTTT HTTP response status: %r" % r.status_code)
+ self.logger.debug(
+ u"IFTTT HTTP response headers: %r" % r.headers)
+ self.logger.debug(
+ u"IFTTT HTTP response body: %r" % r.content)
+
+ if r.status_code != requests.codes.ok:
+ # We had a problem
+ try:
+ self.logger.warning(
+ 'Failed to send IFTTT:%s '
+ 'notification: %s (error=%s).' % (
+ event,
+ HTTP_ERROR_MAP[r.status_code],
+ r.status_code))
+
+ except KeyError:
+ self.logger.warning(
+ 'Failed to send IFTTT:%s '
+ 'notification (error=%s).' % (
+ event, r.status_code))
+
+ # self.logger.debug('Response Details: %s' % r.content)
+ error_count += 1
+
+ else:
+ self.logger.info(
+ 'Sent IFTTT notification to Event %s.' % event)
+
+ except requests.RequestException as e:
+ self.logger.warning(
+ 'A Connection error occured sending IFTTT:%s ' % (
+ event) + 'notification.'
+ )
+ self.logger.debug('Socket Exception: %s' % str(e))
+ error_count += 1
+
+ finally:
+ if len(events):
+ # Prevent thrashing requests
+ self.throttle()
+
+ return (error_count == 0)
+
+ def url(self):
+ """
+ Returns the URL built dynamically based on specified arguments.
+ """
+
+ # Define any arguments set
+ args = {
+ 'format': self.notify_format,
+ }
+
+ return '{schema}://{webhook_id}@{events}/?{args}'.format(
+ schema=self.secure_protocol,
+ webhook_id=self.webhook_id,
+ events='/'.join([self.quote(x, safe='') for x in self.events]),
+ args=self.urlencode(args),
)
- self.logger.debug('IFTTT POST URL: %s (cert_verify=%r)' % (
- url, self.verify_certificate,
- ))
- self.logger.debug('IFTTT Payload: %s' % str(payload))
- try:
- r = requests.post(
- url,
- data=dumps(payload),
- headers=headers,
- verify=self.verify_certificate,
- )
- self.logger.debug(
- u"IFTTT HTTP response status: %r" % r.status_code)
- self.logger.debug(
- u"IFTTT HTTP response headers: %r" % r.headers)
- self.logger.debug(
- u"IFTTT HTTP response body: %r" % r.content)
-
- if r.status_code != requests.codes.ok:
- # We had a problem
- try:
- self.logger.warning(
- 'Failed to send IFTTT:%s '
- 'notification: %s (error=%s).' % (
- self.event,
- HTTP_ERROR_MAP[r.status_code],
- r.status_code))
-
- except KeyError:
- self.logger.warning(
- 'Failed to send IFTTT:%s '
- 'notification (error=%s).' % (
- self.event,
- r.status_code))
-
- # self.logger.debug('Response Details: %s' % r.content)
- return False
-
- else:
- self.logger.info(
- 'Sent IFTTT notification to Event %s.' % self.event)
-
- except requests.RequestException as e:
- self.logger.warning(
- 'A Connection error occured sending IFTTT:%s ' % (
- self.event) + 'notification.'
- )
- self.logger.debug('Socket Exception: %s' % str(e))
- return False
-
- return True
-
@staticmethod
def parse_url(url):
"""
@@ -214,22 +237,14 @@ class NotifyIFTTT(NotifyBase):
return results
# Our Event
- results['event'] = results['host']
+ results['events'] = list()
+ results['events'].append(results['host'])
# Our API Key
- results['apikey'] = results['user']
+ results['webhook_id'] = results['user']
- # Store ValueX entries based on each entry past the host
- results['event_args'] = {
- '{0}{1}'.format(NotifyIFTTT.ifttt_default_key_prefix, n + 1):
- NotifyBase.unquote(x)
- for n, x in enumerate(
- NotifyBase.split_path(results['fullpath'])) if x}
-
- # Allow users to set key=val parameters to specify more types
- # of payload options
- results['event_args'].update(
- {k: NotifyBase.unquote(v)
- for k, v in results['qsd'].items()})
+ # Now fetch the remaining tokens
+ results['events'].extend([x for x in filter(
+ bool, NotifyBase.split_path(results['fullpath']))][0:])
return results
diff --git a/apprise/plugins/NotifyJSON.py b/apprise/plugins/NotifyJSON.py
index ba160e92..c2903c43 100644
--- a/apprise/plugins/NotifyJSON.py
+++ b/apprise/plugins/NotifyJSON.py
@@ -70,6 +70,39 @@ class NotifyJSON(NotifyBase):
return
+ def url(self):
+ """
+ Returns the URL built dynamically based on specified arguments.
+ """
+
+ # Define any arguments set
+ args = {
+ 'format': self.notify_format,
+ }
+
+ # Determine Authentication
+ auth = ''
+ if self.user and self.password:
+ auth = '{user}:{password}@'.format(
+ user=self.quote(self.user, safe=''),
+ password=self.quote(self.password, safe=''),
+ )
+ elif self.user:
+ auth = '{user}@'.format(
+ user=self.quote(self.user, safe=''),
+ )
+
+ default_port = 443 if self.secure else 80
+
+ return '{schema}://{auth}{hostname}{port}/?{args}'.format(
+ schema=self.secure_protocol if self.secure else self.protocol,
+ auth=auth,
+ hostname=self.host,
+ port='' if self.port is None or self.port == default_port
+ else ':{}'.format(self.port),
+ args=self.urlencode(args),
+ )
+
def notify(self, title, body, notify_type, **kwargs):
"""
Perform JSON Notification
diff --git a/apprise/plugins/NotifyJoin.py b/apprise/plugins/NotifyJoin.py
index 67b75ab9..ed74c430 100644
--- a/apprise/plugins/NotifyJoin.py
+++ b/apprise/plugins/NotifyJoin.py
@@ -78,7 +78,7 @@ class NotifyJoin(NotifyBase):
service_url = 'https://joaoapps.com/join/'
# The default protocol
- protocol = 'join'
+ secure_protocol = 'join'
# A URL that takes you to the setup/help of the specific protocol
setup_url = 'https://github.com/caronc/apprise/wiki/Notify_join'
@@ -233,6 +233,22 @@ class NotifyJoin(NotifyBase):
return return_status
+ def url(self):
+ """
+ Returns the URL built dynamically based on specified arguments.
+ """
+
+ # Define any arguments set
+ args = {
+ 'format': self.notify_format,
+ }
+
+ return '{schema}://{apikey}/{devices}/?{args}'.format(
+ schema=self.secure_protocol,
+ apikey=self.quote(self.apikey, safe=''),
+ devices='/'.join([self.quote(x) for x in self.devices]),
+ args=self.urlencode(args))
+
@staticmethod
def parse_url(url):
"""
diff --git a/apprise/plugins/NotifyMatrix.py b/apprise/plugins/NotifyMatrix.py
index 53c38942..27422d1b 100644
--- a/apprise/plugins/NotifyMatrix.py
+++ b/apprise/plugins/NotifyMatrix.py
@@ -112,7 +112,6 @@ class NotifyMatrix(NotifyBase):
if not self.user:
self.logger.warning(
'No user was specified; using %s.' % MATRIX_DEFAULT_USER)
- self.user = MATRIX_DEFAULT_USER
if mode not in MATRIX_NOTIFICATION_MODES:
self.logger.warning('The mode specified (%s) is invalid.' % mode)
@@ -209,7 +208,7 @@ class NotifyMatrix(NotifyBase):
def __slack_mode_payload(self, title, body, notify_type):
# prepare JSON Object
payload = {
- 'username': self.user,
+ 'username': self.user if self.user else MATRIX_DEFAULT_USER,
# Use Markdown language
'mrkdwn': True,
'attachments': [{
@@ -230,13 +229,43 @@ class NotifyMatrix(NotifyBase):
msg = '%s
%s
' % (title, body)
payload = {
- 'displayName': self.user,
+ 'displayName': self.user if self.user else MATRIX_DEFAULT_USER,
'format': 'html',
'text': msg,
}
return payload
+ def url(self):
+ """
+ Returns the URL built dynamically based on specified arguments.
+ """
+
+ # Define any arguments set
+ args = {
+ 'format': self.notify_format,
+ 'mode': self.mode,
+ }
+
+ # Determine Authentication
+ auth = ''
+ if self.user:
+ auth = '{user}@'.format(
+ user=self.quote(self.user, safe=''),
+ )
+
+ default_port = 443 if self.secure else 80
+
+ return '{schema}://{auth}{host}/{token}{port}/?{args}'.format(
+ schema=self.secure_protocol if self.secure else self.protocol,
+ host=self.host,
+ auth=auth,
+ token=self.token,
+ port='' if self.port is None or self.port == default_port
+ else ':{}'.format(self.port),
+ args=self.urlencode(args),
+ )
+
@staticmethod
def parse_url(url):
"""
diff --git a/apprise/plugins/NotifyMatterMost.py b/apprise/plugins/NotifyMatterMost.py
index e92b8987..02ed020c 100644
--- a/apprise/plugins/NotifyMatterMost.py
+++ b/apprise/plugins/NotifyMatterMost.py
@@ -179,6 +179,28 @@ class NotifyMatterMost(NotifyBase):
return True
+ def url(self):
+ """
+ Returns the URL built dynamically based on specified arguments.
+ """
+
+ # Define any arguments set
+ args = {
+ 'format': self.notify_format,
+ }
+
+ default_port = 443 if self.secure else self.default_port
+ default_schema = self.secure_protocol if self.secure else self.protocol
+
+ return '{schema}://{hostname}{port}/{authtoken}/?{args}'.format(
+ schema=default_schema,
+ hostname=self.host,
+ port='' if not self.port or self.port == default_port
+ else ':{}'.format(self.port),
+ authtoken=self.quote(self.authtoken, safe=''),
+ args=self.urlencode(args),
+ )
+
@staticmethod
def parse_url(url):
"""
diff --git a/apprise/plugins/NotifyProwl.py b/apprise/plugins/NotifyProwl.py
index 7ddb6f11..c501d370 100644
--- a/apprise/plugins/NotifyProwl.py
+++ b/apprise/plugins/NotifyProwl.py
@@ -190,6 +190,34 @@ class NotifyProwl(NotifyBase):
return True
+ def url(self):
+ """
+ Returns the URL built dynamically based on specified arguments.
+ """
+
+ _map = {
+ ProwlPriority.LOW: 'low',
+ ProwlPriority.MODERATE: 'moderate',
+ ProwlPriority.NORMAL: 'normal',
+ ProwlPriority.HIGH: 'high',
+ ProwlPriority.EMERGENCY: 'emergency',
+ }
+
+ # Define any arguments set
+ args = {
+ 'format': self.notify_format,
+ 'priority': 'normal' if self.priority not in _map
+ else _map[self.priority]
+ }
+
+ return '{schema}://{apikey}/{providerkey}/?{args}'.format(
+ schema=self.secure_protocol,
+ apikey=self.quote(self.apikey, safe=''),
+ providerkey='' if not self.providerkey
+ else self.quote(self.providerkey, safe=''),
+ args=self.urlencode(args),
+ )
+
@staticmethod
def parse_url(url):
"""
@@ -216,15 +244,10 @@ class NotifyProwl(NotifyBase):
if 'priority' in results['qsd'] and len(results['qsd']['priority']):
_map = {
'l': ProwlPriority.LOW,
- '-2': ProwlPriority.LOW,
'm': ProwlPriority.MODERATE,
- '-1': ProwlPriority.MODERATE,
'n': ProwlPriority.NORMAL,
- '0': ProwlPriority.NORMAL,
'h': ProwlPriority.HIGH,
- '1': ProwlPriority.HIGH,
'e': ProwlPriority.EMERGENCY,
- '2': ProwlPriority.EMERGENCY,
}
try:
results['priority'] = \
diff --git a/apprise/plugins/NotifyPushBullet.py b/apprise/plugins/NotifyPushBullet.py
index 5eac9a46..e772d2eb 100644
--- a/apprise/plugins/NotifyPushBullet.py
+++ b/apprise/plugins/NotifyPushBullet.py
@@ -182,6 +182,28 @@ class NotifyPushBullet(NotifyBase):
return not has_error
+ def url(self):
+ """
+ Returns the URL built dynamically based on specified arguments.
+ """
+
+ # Define any arguments set
+ args = {
+ 'format': self.notify_format,
+ }
+
+ recipients = '/'.join([self.quote(x) for x in self.recipients])
+ if recipients == PUSHBULLET_SEND_TO_ALL:
+ # keyword is reserved for internal usage only; it's safe to remove
+ # it from the recipients list
+ recipients = ''
+
+ return '{schema}://{accesstoken}/{recipients}/?{args}'.format(
+ schema=self.secure_protocol,
+ accesstoken=self.quote(self.accesstoken, safe=''),
+ recipients=recipients,
+ args=self.urlencode(args))
+
@staticmethod
def parse_url(url):
"""
diff --git a/apprise/plugins/NotifyPushed.py b/apprise/plugins/NotifyPushed.py
index b925ba63..df5a72fe 100644
--- a/apprise/plugins/NotifyPushed.py
+++ b/apprise/plugins/NotifyPushed.py
@@ -26,6 +26,7 @@
import re
import requests
from json import dumps
+from itertools import chain
from .NotifyBase import NotifyBase
from .NotifyBase import HTTP_ERROR_MAP
@@ -256,6 +257,29 @@ class NotifyPushed(NotifyBase):
return True
+ def url(self):
+ """
+ Returns the URL built dynamically based on specified arguments.
+ """
+
+ # Define any arguments set
+ args = {
+ 'format': self.notify_format,
+ }
+
+ return '{schema}://{app_key}/{app_secret}/{targets}/?{args}'.format(
+ schema=self.secure_protocol,
+ app_key=self.quote(self.app_key, safe=''),
+ app_secret=self.quote(self.app_secret, safe=''),
+ targets='/'.join(
+ [self.quote(x) for x in chain(
+ # Channels are prefixed with a pound/hashtag symbol
+ ['#{}'.format(x) for x in self.channels],
+ # Users are prefixed with an @ symbol
+ ['@{}'.format(x) for x in self.users],
+ )]),
+ args=self.urlencode(args))
+
@staticmethod
def parse_url(url):
"""
diff --git a/apprise/plugins/NotifyPushjet/NotifyPushjet.py b/apprise/plugins/NotifyPushjet/NotifyPushjet.py
index cb9ddc94..f4bf6062 100644
--- a/apprise/plugins/NotifyPushjet/NotifyPushjet.py
+++ b/apprise/plugins/NotifyPushjet/NotifyPushjet.py
@@ -52,12 +52,15 @@ class NotifyPushjet(NotifyBase):
# A URL that takes you to the setup/help of the specific protocol
setup_url = 'https://github.com/caronc/apprise/wiki/Notify_pushjet'
- def __init__(self, **kwargs):
+ def __init__(self, secret_key, **kwargs):
"""
Initialize Pushjet Object
"""
super(NotifyPushjet, self).__init__(**kwargs)
+ # store our key
+ self.secret_key = secret_key
+
def notify(self, title, body, notify_type):
"""
Perform Pushjet Notification
@@ -72,7 +75,7 @@ class NotifyPushjet(NotifyBase):
server += ":" + str(self.port)
api = pushjet.Api(server)
- service = api.Service(secret_key=self.user)
+ service = api.Service(secret_key=self.secret_key)
service.send(body, title)
self.logger.info('Sent Pushjet notification.')
@@ -84,6 +87,27 @@ class NotifyPushjet(NotifyBase):
return True
+ def url(self):
+ """
+ Returns the URL built dynamically based on specified arguments.
+ """
+
+ # Define any arguments set
+ args = {
+ 'format': self.notify_format,
+ }
+
+ default_port = 443 if self.secure else 80
+
+ return '{schema}://{secret_key}@{hostname}{port}/?{args}'.format(
+ schema=self.secure_protocol if self.secure else self.protocol,
+ secret_key=self.quote(self.secret_key, safe=''),
+ hostname=self.host,
+ port='' if self.port is None or self.port == default_port
+ else ':{}'.format(self.port),
+ args=self.urlencode(args),
+ )
+
@staticmethod
def parse_url(url):
"""
@@ -91,10 +115,10 @@ class NotifyPushjet(NotifyBase):
us to substantiate this object.
Syntax:
- pjet://secret@hostname
- pjet://secret@hostname:port
- pjets://secret@hostname
- pjets://secret@hostname:port
+ pjet://secret_key@hostname
+ pjet://secret_key@hostname:port
+ pjets://secret_key@hostname
+ pjets://secret_key@hostname:port
"""
results = NotifyBase.parse_url(url)
@@ -107,4 +131,7 @@ class NotifyPushjet(NotifyBase):
# a username is required
return None
+ # Store it as it's value
+ results['secret_key'] = results.get('user')
+
return results
diff --git a/apprise/plugins/NotifyPushover.py b/apprise/plugins/NotifyPushover.py
index fee093d7..72232f7b 100644
--- a/apprise/plugins/NotifyPushover.py
+++ b/apprise/plugins/NotifyPushover.py
@@ -237,6 +237,41 @@ class NotifyPushover(NotifyBase):
return not has_error
+ def url(self):
+ """
+ Returns the URL built dynamically based on specified arguments.
+ """
+
+ _map = {
+ PushoverPriority.LOW: 'low',
+ PushoverPriority.MODERATE: 'moderate',
+ PushoverPriority.NORMAL: 'normal',
+ PushoverPriority.HIGH: 'high',
+ PushoverPriority.EMERGENCY: 'emergency',
+ }
+
+ # Define any arguments set
+ args = {
+ 'format': self.notify_format,
+ 'priority':
+ _map[PushoverPriority.NORMAL] if self.priority not in _map
+ else _map[self.priority],
+ }
+
+ devices = '/'.join([self.quote(x) for x in self.devices])
+ if devices == PUSHOVER_SEND_TO_ALL:
+ # keyword is reserved for internal usage only; it's safe to remove
+ # it from the devices list
+ devices = ''
+
+ return '{schema}://{auth}{token}/{devices}/?{args}'.format(
+ schema=self.secure_protocol,
+ auth='' if not self.user
+ else '{user}@'.format(user=self.quote(self.user, safe='')),
+ token=self.quote(self.token, safe=''),
+ devices=devices,
+ args=self.urlencode(args))
+
@staticmethod
def parse_url(url):
"""
diff --git a/apprise/plugins/NotifyRocketChat.py b/apprise/plugins/NotifyRocketChat.py
index cb86b84d..899c8196 100644
--- a/apprise/plugins/NotifyRocketChat.py
+++ b/apprise/plugins/NotifyRocketChat.py
@@ -26,6 +26,7 @@
import re
import requests
from json import loads
+from itertools import chain
from .NotifyBase import NotifyBase
from .NotifyBase import HTTP_ERROR_MAP
@@ -106,6 +107,12 @@ class NotifyRocketChat(NotifyBase):
elif not isinstance(recipients, (set, tuple, list)):
recipients = []
+ if not (self.user and self.password):
+ # Username & Password is required for Rocket Chat to work
+ raise TypeError(
+ 'No Rocket.Chat user/pass combo specified.'
+ )
+
# Validate recipients and drop bad ones:
for recipient in recipients:
result = IS_CHANNEL.match(recipient)
@@ -133,6 +140,40 @@ class NotifyRocketChat(NotifyBase):
# Used to track token headers upon authentication (if successful)
self.headers = {}
+ def url(self):
+ """
+ Returns the URL built dynamically based on specified arguments.
+ """
+
+ # Define any arguments set
+ args = {
+ 'format': self.notify_format,
+ }
+
+ # Determine Authentication
+ auth = '{user}:{password}@'.format(
+ user=self.quote(self.user, safe=''),
+ password=self.quote(self.password, safe=''),
+ )
+
+ default_port = 443 if self.secure else 80
+
+ return '{schema}://{auth}{hostname}{port}/{targets}/?{args}'.format(
+ schema=self.secure_protocol if self.secure else self.protocol,
+ auth=auth,
+ hostname=self.host,
+ port='' if self.port is None or self.port == default_port
+ else ':{}'.format(self.port),
+ targets='/'.join(
+ [self.quote(x) for x in chain(
+ # Channels are prefixed with a pound/hashtag symbol
+ ['#{}'.format(x) for x in self.channels],
+ # Rooms are as is
+ self.rooms,
+ )]),
+ args=self.urlencode(args),
+ )
+
def notify(self, title, body, notify_type, **kwargs):
"""
wrapper to send_notification since we can alert more then one channel
diff --git a/apprise/plugins/NotifyRyver.py b/apprise/plugins/NotifyRyver.py
index c0f6b81c..49b17412 100644
--- a/apprise/plugins/NotifyRyver.py
+++ b/apprise/plugins/NotifyRyver.py
@@ -221,6 +221,32 @@ class NotifyRyver(NotifyBase):
return True
+ def url(self):
+ """
+ Returns the URL built dynamically based on specified arguments.
+ """
+
+ # Define any arguments set
+ args = {
+ 'format': self.notify_format,
+ 'webhook': self.webhook,
+ }
+
+ # Determine if there is a botname present
+ botname = ''
+ if self.user:
+ botname = '{botname}@'.format(
+ botname=self.quote(self.user, safe=''),
+ )
+
+ return '{schema}://{botname}{organization}/{token}/?{args}'.format(
+ schema=self.secure_protocol,
+ botname=botname,
+ organization=self.quote(self.organization, safe=''),
+ token=self.quote(self.token, safe=''),
+ args=self.urlencode(args),
+ )
+
@staticmethod
def parse_url(url):
"""
diff --git a/apprise/plugins/NotifySNS.py b/apprise/plugins/NotifySNS.py
index abc46f42..b29327f0 100644
--- a/apprise/plugins/NotifySNS.py
+++ b/apprise/plugins/NotifySNS.py
@@ -31,6 +31,7 @@ from hashlib import sha256
from datetime import datetime
from collections import OrderedDict
from xml.etree import ElementTree
+from itertools import chain
from .NotifyBase import NotifyBase
from .NotifyBase import HTTP_ERROR_MAP
@@ -58,8 +59,8 @@ LIST_DELIM = re.compile(r'[ \t\r\n,\\/]+')
# region as a delimiter. This is a bit hacky; but it's much easier than having
# users of this product search though this Access Key Secret and escape all
# of the forward slashes!
-IS_REGION = re.compile(r'^\s*(?P[a-z]{2})-'
- r'(?P[a-z]+)-(?P[0-9]+)\s*$', re.I)
+IS_REGION = re.compile(
+ r'^\s*(?P[a-z]{2})-(?P[a-z]+)-(?P[0-9]+)\s*$', re.I)
# Extend HTTP Error Messages
AWS_HTTP_ERROR_MAP = HTTP_ERROR_MAP.copy()
@@ -521,6 +522,32 @@ class NotifySNS(NotifyBase):
return response
+ def url(self):
+ """
+ Returns the URL built dynamically based on specified arguments.
+ """
+
+ # Define any arguments set
+ args = {
+ 'format': self.notify_format,
+ }
+
+ return '{schema}://{key_id}/{key_secret}/{region}/{targets}/'\
+ '?{args}'.format(
+ schema=self.secure_protocol,
+ key_id=self.quote(self.aws_access_key_id, safe=''),
+ key_secret=self.quote(self.aws_secret_access_key, safe=''),
+ region=self.quote(self.aws_region_name, safe=''),
+ targets='/'.join(
+ [self.quote(x) for x in chain(
+ # Phone # are prefixed with a plus symbol
+ ['+{}'.format(x) for x in self.phone],
+ # Topics are prefixed with a pound/hashtag symbol
+ ['#{}'.format(x) for x in self.topics],
+ )]),
+ args=self.urlencode(args),
+ )
+
@staticmethod
def parse_url(url):
"""
diff --git a/apprise/plugins/NotifySlack.py b/apprise/plugins/NotifySlack.py
index ffa8a71a..430a33ff 100644
--- a/apprise/plugins/NotifySlack.py
+++ b/apprise/plugins/NotifySlack.py
@@ -141,7 +141,6 @@ class NotifySlack(NotifyBase):
if not self.user:
self.logger.warning(
'No user was specified; using %s.' % SLACK_DEFAULT_USER)
- self.user = SLACK_DEFAULT_USER
if compat_is_basestring(channels):
self.channels = [x for x in filter(bool, CHANNEL_LIST_DELIM.split(
@@ -231,7 +230,7 @@ class NotifySlack(NotifyBase):
# prepare JSON Object
payload = {
'channel': _channel,
- 'username': self.user,
+ 'username': self.user if self.user else SLACK_DEFAULT_USER,
# Use Markdown language
'mrkdwn': True,
'attachments': [{
@@ -297,6 +296,35 @@ class NotifySlack(NotifyBase):
return notify_okay
+ def url(self):
+ """
+ Returns the URL built dynamically based on specified arguments.
+ """
+
+ # Define any arguments set
+ args = {
+ 'format': self.notify_format,
+ }
+
+ # Determine if there is a botname present
+ botname = ''
+ if self.user:
+ botname = '{botname}@'.format(
+ botname=self.quote(self.user, safe=''),
+ )
+
+ return '{schema}://{botname}{token_a}/{token_b}/{token_c}/{targets}/'\
+ '?{args}'.format(
+ schema=self.secure_protocol,
+ botname=botname,
+ token_a=self.quote(self.token_a, safe=''),
+ token_b=self.quote(self.token_b, safe=''),
+ token_c=self.quote(self.token_c, safe=''),
+ targets='/'.join(
+ [self.quote(x, safe='') for x in self.channels]),
+ args=self.urlencode(args),
+ )
+
@staticmethod
def parse_url(url):
"""
diff --git a/apprise/plugins/NotifyTelegram.py b/apprise/plugins/NotifyTelegram.py
index 4ad54cc8..094665fe 100644
--- a/apprise/plugins/NotifyTelegram.py
+++ b/apprise/plugins/NotifyTelegram.py
@@ -60,8 +60,8 @@ from json import dumps
from .NotifyBase import NotifyBase
from .NotifyBase import HTTP_ERROR_MAP
from ..common import NotifyImageSize
-from ..utils import compat_is_basestring
from ..utils import parse_bool
+from ..utils import parse_list
from ..common import NotifyFormat
TELEGRAM_IMAGE_XY = NotifyImageSize.XY_256
@@ -81,9 +81,6 @@ IS_CHAT_ID_RE = re.compile(
re.IGNORECASE,
)
-# Used to break path apart into list of chat identifiers
-CHAT_ID_LIST_DELIM = re.compile(r'[ \t\r\n,#\\/]+')
-
class NotifyTelegram(NotifyBase):
"""
@@ -134,16 +131,8 @@ class NotifyTelegram(NotifyBase):
# Store our Bot Token
self.bot_token = result.group('key')
- if compat_is_basestring(chat_ids):
- self.chat_ids = [x for x in filter(bool, CHAT_ID_LIST_DELIM.split(
- chat_ids,
- ))]
-
- elif isinstance(chat_ids, (set, tuple, list)):
- self.chat_ids = list(chat_ids)
-
- else:
- self.chat_ids = list()
+ # Parse our list
+ self.chat_ids = parse_list(chat_ids)
if self.user:
# Treat this as a channel too
@@ -153,7 +142,7 @@ class NotifyTelegram(NotifyBase):
_id = self.detect_bot_owner()
if _id:
# Store our id
- self.chat_ids = [str(_id)]
+ self.chat_ids.append(str(_id))
if len(self.chat_ids) == 0:
self.logger.warning('No chat_id(s) were specified.')
@@ -501,6 +490,25 @@ class NotifyTelegram(NotifyBase):
return not has_error
+ def url(self):
+ """
+ Returns the URL built dynamically based on specified arguments.
+ """
+
+ # Define any arguments set
+ args = {
+ 'format': self.notify_format,
+ }
+
+ # No need to check the user token because the user automatically gets
+ # appended into the list of chat ids
+ return '{schema}://{bot_token}/{targets}/?{args}'.format(
+ schema=self.secure_protocol,
+ bot_token=self.quote(self.bot_token, safe=''),
+ targets='/'.join(
+ [self.quote('@{}'.format(x)) for x in self.chat_ids]),
+ args=self.urlencode(args))
+
@staticmethod
def parse_url(url):
"""
@@ -512,17 +520,17 @@ class NotifyTelegram(NotifyBase):
# tgram:// messages since the bot_token has a colon in it.
# It invalidates an normal URL.
- # This hack searches for this bogus URL and corrects it
- # so we can properly load it further down. The other
- # alternative is to ask users to actually change the colon
- # into a slash (which will work too), but it's more likely
- # to cause confusion... So this is the next best thing
+ # This hack searches for this bogus URL and corrects it so we can
+ # properly load it further down. The other alternative is to ask users
+ # to actually change the colon into a slash (which will work too), but
+ # it's more likely to cause confusion... So this is the next best thing
+ # we also check for %3A (incase the URL is encoded) as %3A == :
try:
tgram = re.match(
- r'(?P%s://)(bot)?(?P([a-z0-9_-]+)'
- r'(:[a-z0-9_-]+)?@)?(?P[0-9]+):+'
- r'(?P.*)$' % NotifyTelegram.secure_protocol,
- url, re.I)
+ r'(?P{schema}://)(bot)?(?P([a-z0-9_-]+)'
+ r'(:[a-z0-9_-]+)?@)?(?P[0-9]+)(:|%3A)+'
+ r'(?P.*)$'.format(
+ schema=NotifyTelegram.secure_protocol), url, re.I)
except (TypeError, AttributeError):
# url is bad; force tgram to be None
@@ -534,14 +542,11 @@ class NotifyTelegram(NotifyBase):
if tgram.group('prefix'):
# Try again
- results = NotifyBase.parse_url(
- '%s%s%s/%s' % (
- tgram.group('protocol'),
- tgram.group('prefix'),
- tgram.group('btoken_a'),
- tgram.group('remaining'),
- ),
- )
+ results = NotifyBase.parse_url('%s%s%s/%s' % (
+ tgram.group('protocol'),
+ tgram.group('prefix'),
+ tgram.group('btoken_a'),
+ tgram.group('remaining')))
else:
# Try again
@@ -562,9 +567,8 @@ class NotifyTelegram(NotifyBase):
bot_token = '%s:%s' % (bot_token_a, bot_token_b)
- chat_ids = ','.join(
- [x for x in filter(
- bool, NotifyBase.split_path(results['fullpath']))][1:])
+ chat_ids = [x for x in filter(
+ bool, NotifyBase.split_path(results['fullpath']))][1:]
# Store our bot token
results['bot_token'] = bot_token
diff --git a/apprise/plugins/NotifyTwitter/NotifyTwitter.py b/apprise/plugins/NotifyTwitter/NotifyTwitter.py
index df7dbdd8..394e5473 100644
--- a/apprise/plugins/NotifyTwitter/NotifyTwitter.py
+++ b/apprise/plugins/NotifyTwitter/NotifyTwitter.py
@@ -109,7 +109,9 @@ class NotifyTwitter(NotifyBase):
)
return False
- text = '%s\r\n%s' % (title, body)
+ # Only set title if it was specified
+ text = body if not title else '%s\r\n%s' % (title, body)
+
try:
# Get our API
api = tweepy.API(self.auth)
diff --git a/apprise/plugins/NotifyWindows.py b/apprise/plugins/NotifyWindows.py
index ea0f0e16..86fd6dc8 100644
--- a/apprise/plugins/NotifyWindows.py
+++ b/apprise/plugins/NotifyWindows.py
@@ -175,6 +175,13 @@ class NotifyWindows(NotifyBase):
return True
+ def url(self):
+ """
+ Returns the URL built dynamically based on specified arguments.
+ """
+
+ return '{schema}://'.format(schema=self.protocol)
+
@staticmethod
def parse_url(url):
"""
diff --git a/apprise/plugins/NotifyXBMC.py b/apprise/plugins/NotifyXBMC.py
index 264e8fad..fc46645f 100644
--- a/apprise/plugins/NotifyXBMC.py
+++ b/apprise/plugins/NotifyXBMC.py
@@ -44,11 +44,16 @@ class NotifyXBMC(NotifyBase):
# The services URL
service_url = 'http://kodi.tv/'
+ xbmc_protocol = 'xbmc'
+ xbmc_secure_protocol = 'xbmcs'
+ kodi_protocol = 'kodi'
+ kodi_secure_protocol = 'kodis'
+
# The default protocols
- protocol = ('xbmc', 'kodi')
+ protocol = (xbmc_protocol, kodi_protocol)
# The default secure protocols
- secure_protocol = ('xbmc', 'kodis')
+ secure_protocol = (xbmc_secure_protocol, kodi_secure_protocol)
# A URL that takes you to the setup/help of the specific protocol
setup_url = 'https://github.com/caronc/apprise/wiki/Notify_kodi'
@@ -224,6 +229,44 @@ class NotifyXBMC(NotifyBase):
return True
+ def url(self):
+ """
+ Returns the URL built dynamically based on specified arguments.
+ """
+
+ # Define any arguments set
+ args = {
+ 'format': self.notify_format,
+ }
+
+ # Determine Authentication
+ auth = ''
+ if self.user and self.password:
+ auth = '{user}:{password}@'.format(
+ user=self.quote(self.user, safe=''),
+ password=self.quote(self.password, safe=''),
+ )
+ elif self.user:
+ auth = '{user}@'.format(
+ user=self.quote(self.user, safe=''),
+ )
+
+ default_schema = self.xbmc_protocol if (
+ self.protocol <= self.xbmc_remote_protocol) else self.kodi_protocol
+ default_port = 443 if self.secure else self.xbmc_default_port
+ if self.secure:
+ # Append 's' to schema
+ default_schema + 's'
+
+ return '{schema}://{auth}{hostname}{port}/?{args}'.format(
+ schema=default_schema,
+ auth=auth,
+ hostname=self.host,
+ port='' if not self.port or self.port == default_port
+ else ':{}'.format(self.port),
+ args=self.urlencode(args),
+ )
+
@staticmethod
def parse_url(url):
"""
diff --git a/apprise/plugins/NotifyXML.py b/apprise/plugins/NotifyXML.py
index 9b2f36b2..cad538db 100644
--- a/apprise/plugins/NotifyXML.py
+++ b/apprise/plugins/NotifyXML.py
@@ -85,6 +85,39 @@ class NotifyXML(NotifyBase):
return
+ def url(self):
+ """
+ Returns the URL built dynamically based on specified arguments.
+ """
+
+ # Define any arguments set
+ args = {
+ 'format': self.notify_format,
+ }
+
+ # Determine Authentication
+ auth = ''
+ if self.user and self.password:
+ auth = '{user}:{password}@'.format(
+ user=self.quote(self.user, safe=''),
+ password=self.quote(self.password, safe=''),
+ )
+ elif self.user:
+ auth = '{user}@'.format(
+ user=self.quote(self.user, safe=''),
+ )
+
+ default_port = 443 if self.secure else 80
+
+ return '{schema}://{auth}{hostname}{port}/?{args}'.format(
+ schema=self.secure_protocol if self.secure else self.protocol,
+ auth=auth,
+ hostname=self.host,
+ port='' if self.port is None or self.port == default_port
+ else ':{}'.format(self.port),
+ args=self.urlencode(args),
+ )
+
def notify(self, title, body, notify_type, **kwargs):
"""
Perform XML Notification
diff --git a/apprise/utils.py b/apprise/utils.py
index 091c7e88..420cf2b5 100644
--- a/apprise/utils.py
+++ b/apprise/utils.py
@@ -397,6 +397,10 @@ def parse_list(*args):
elif isinstance(arg, (set, list, tuple)):
result += parse_list(*arg)
+ elif arg is None:
+ # Ignore
+ continue
+
else:
# Convert whatever it is to a string and work with it
result += parse_list(str(arg))
diff --git a/test/test_api.py b/test/test_api.py
index fa6ffada..bf499a65 100644
--- a/test/test_api.py
+++ b/test/test_api.py
@@ -71,15 +71,34 @@ def test_apprise():
a = Apprise(servers=servers)
- # 3 servers loaded
+ # 2 servers loaded
assert(len(a) == 2)
+ # We can retrieve our URLs this way:
+ assert(len(a.urls()) == 2)
+
# We can add another server
assert(
a.add('mmosts://mattermost.server.local/'
'3ccdd113474722377935511fc85d3dd4') is True)
assert(len(a) == 3)
+ # We can pop an object off of our stack by it's indexed value:
+ obj = a.pop(0)
+ assert(isinstance(obj, NotifyBase) is True)
+ assert(len(a) == 2)
+
+ # We can retrieve elements from our list too by reference:
+ assert(compat_is_basestring(a[0].url()) is True)
+
+ # We can iterate over our list too:
+ count = 0
+ for o in a:
+ assert(compat_is_basestring(o.url()) is True)
+ count += 1
+ # verify that we did indeed iterate over each element
+ assert(len(a) == count)
+
# We can empty our set
a.clear()
assert(len(a) == 0)
diff --git a/test/test_email_plugin.py b/test/test_email_plugin.py
index bbd89c8d..80d108f4 100644
--- a/test/test_email_plugin.py
+++ b/test/test_email_plugin.py
@@ -26,6 +26,7 @@
from apprise import plugins
from apprise import NotifyType
from apprise import Apprise
+from apprise.utils import compat_is_basestring
from apprise.plugins import NotifyEmailBase
import smtplib
@@ -49,7 +50,7 @@ TEST_URLS = (
# No Username
('mailtos://:pass@nuxref.com:567', {
# Can't prepare a To address using this expression
- 'exception': TypeError,
+ 'instance': TypeError,
}),
# Pre-Configured Email Services
@@ -115,27 +116,27 @@ TEST_URLS = (
}),
# Invalid From Address
('mailtos://user:pass@nuxref.com?from=@', {
- 'exception': TypeError,
+ 'instance': TypeError,
}),
# Invalid From Address
('mailtos://nuxref.com?user=&pass=.', {
- 'exception': TypeError,
+ 'instance': TypeError,
}),
# Invalid To Address
('mailtos://user:pass@nuxref.com?to=@', {
- 'exception': TypeError,
+ 'instance': TypeError,
}),
# Valid URL, but can't structure a proper email
('mailtos://nuxref.com?user=%20!&pass=.', {
- 'exception': TypeError,
+ 'instance': TypeError,
}),
# Invalid From (and To) Address
('mailtos://nuxref.com?to=test', {
- 'exception': TypeError,
+ 'instance': TypeError,
}),
# Invalid Secure Mode
('mailtos://user:pass@example.com?mode=notamode', {
- 'exception': TypeError,
+ 'instance': TypeError,
}),
# STARTTLS flag checking
('mailtos://user:pass@gmail.com?mode=starttls', {
@@ -172,9 +173,6 @@ def test_email_plugin(mock_smtp, mock_smtpssl):
# Our expected instance
instance = meta.get('instance', None)
- # Our expected exception
- exception = meta.get('exception', None)
-
# Our expected server objects
self = meta.get('self', None)
@@ -217,19 +215,37 @@ def test_email_plugin(mock_smtp, mock_smtpssl):
try:
obj = Apprise.instantiate(url, suppress_exceptions=False)
- assert(exception is None)
-
if obj is None:
- # We're done
+ # We're done (assuming this is what we were expecting)
+ assert instance is None
continue
if instance is None:
# Expected None but didn't get it
- print('%s instantiated %s' % (url, str(obj)))
+ print('%s instantiated %s (but expected None)' % (
+ url, str(obj)))
assert(False)
assert(isinstance(obj, instance))
+ if isinstance(obj, plugins.NotifyBase.NotifyBase):
+ # We loaded okay; now lets make sure we can reverse this url
+ assert(compat_is_basestring(obj.url()) is True)
+
+ # Instantiate the exact same object again using the URL from
+ # the one that was already created properly
+ obj_cmp = 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.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
+ print('TEST FAIL: {} regenerated as {}'.format(
+ url, obj.url()))
+ assert(False)
+
if self:
# Iterate over our expected entries inside of our object
for key, val in self.items():
@@ -258,16 +274,17 @@ def test_email_plugin(mock_smtp, mock_smtpssl):
except Exception as e:
# We can't handle this exception type
- print('%s / %s' % (url, str(e)))
- assert False
+ raise
except AssertionError:
# Don't mess with these entries
+ print('%s AssertionError' % url)
raise
except Exception as e:
# Check that we were expecting this exception to happen
- assert isinstance(e, response)
+ if not isinstance(e, response):
+ raise
except AssertionError:
# Don't mess with these entries
@@ -276,9 +293,11 @@ def test_email_plugin(mock_smtp, mock_smtpssl):
except Exception as e:
# Handle our exception
- print('%s / %s' % (url, str(e)))
- assert(exception is not None)
- assert(isinstance(e, exception))
+ if(instance is None):
+ raise
+
+ if not isinstance(e, instance):
+ raise
@mock.patch('smtplib.SMTP')
diff --git a/test/test_glib_plugin.py b/test/test_glib_plugin.py
index 98d5e6b4..c2650f1a 100644
--- a/test/test_glib_plugin.py
+++ b/test/test_glib_plugin.py
@@ -28,6 +28,7 @@ import mock
import sys
import types
import apprise
+from apprise.utils import compat_is_basestring
try:
# Python v3.4+
@@ -223,6 +224,9 @@ def test_dbus_plugin(mock_mainloop, mock_byte, mock_bytearray,
assert(isinstance(obj, apprise.plugins.NotifyDBus) is True)
obj.duration = 0
+ # Test url() call
+ assert(compat_is_basestring(obj.url()) is True)
+
# Our notification succeeds even though the gi library was not loaded
assert(obj.notify(title='title', body='body',
notify_type=apprise.NotifyType.INFO) is True)
diff --git a/test/test_gnome_plugin.py b/test/test_gnome_plugin.py
index 150293c9..ef66e8d2 100644
--- a/test/test_gnome_plugin.py
+++ b/test/test_gnome_plugin.py
@@ -28,6 +28,7 @@ import sys
import types
import apprise
+from apprise.utils import compat_is_basestring
try:
# Python v3.4+
@@ -113,6 +114,9 @@ def test_gnome_plugin():
# Check that it found our mocked environments
assert(obj._enabled is True)
+ # Test url() call
+ assert(compat_is_basestring(obj.url()) is True)
+
# test notifications
assert(obj.notify(title='title', body='body',
notify_type=apprise.NotifyType.INFO) is True)
diff --git a/test/test_growl_plugin.py b/test/test_growl_plugin.py
index 289f726f..295ec5fb 100644
--- a/test/test_growl_plugin.py
+++ b/test/test_growl_plugin.py
@@ -26,6 +26,8 @@
from apprise import plugins
from apprise import NotifyType
from apprise import Apprise
+from apprise.utils import compat_is_basestring
+
import mock
import re
@@ -195,7 +197,6 @@ def test_growl_plugin(mock_gntp):
except Exception as e:
# We can't handle this exception type
- print('%s / %s' % (url, str(e)))
assert False
# We're done this part of the test
@@ -216,10 +217,27 @@ def test_growl_plugin(mock_gntp):
if instance is None:
# Expected None but didn't get it
- print('%s instantiated %s' % (url, str(obj)))
assert(False)
- assert(isinstance(obj, instance))
+ assert(isinstance(obj, instance) is True)
+
+ if isinstance(obj, plugins.NotifyBase.NotifyBase):
+ # We loaded okay; now lets make sure we can reverse this url
+ assert(compat_is_basestring(obj.url()) is True)
+
+ # Instantiate the exact same object again using the URL from
+ # the one that was already created properly
+ obj_cmp = 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.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
+ print('TEST FAIL: {} regenerated as {}'.format(
+ url, obj.url()))
+ assert(False)
if self:
# Iterate over our expected entries inside of our object
diff --git a/test/test_notify_base.py b/test/test_notify_base.py
index dc3ee937..1afd328c 100644
--- a/test/test_notify_base.py
+++ b/test/test_notify_base.py
@@ -52,6 +52,16 @@ def test_notify_base():
nb = NotifyBase(port=10)
assert nb.port == 10
+ try:
+ nb.url()
+ assert False
+
+ except NotImplementedError:
+ # Each sub-module is that inherits this as a parent is required to
+ # over-ride this function. So direct calls to this throws a not
+ # implemented error intentionally
+ assert True
+
# Throttle overrides..
nb = NotifyBase()
nb.throttle_attempt = 0.0
diff --git a/test/test_pushjet_plugin.py b/test/test_pushjet_plugin.py
index 789d4844..f2d19028 100644
--- a/test/test_pushjet_plugin.py
+++ b/test/test_pushjet_plugin.py
@@ -26,6 +26,8 @@
from apprise import plugins
from apprise import NotifyType
from apprise import Apprise
+from apprise.utils import compat_is_basestring
+
import mock
TEST_URLS = (
@@ -104,13 +106,37 @@ def test_plugin(mock_refresh, mock_send):
try:
obj = Apprise.instantiate(url, suppress_exceptions=False)
- if instance is None:
- # Check that we got what we came for
- assert obj is instance
+ if obj is None:
+ # We're done (assuming this is what we were expecting)
+ assert instance is None
continue
+ if instance is None:
+ # Expected None but didn't get it
+ print('%s instantiated %s (but expected None)' % (
+ url, str(obj)))
+ assert(False)
+
assert(isinstance(obj, instance))
+ if isinstance(obj, plugins.NotifyBase.NotifyBase):
+ # We loaded okay; now lets make sure we can reverse this url
+ assert(compat_is_basestring(obj.url()) is True)
+
+ # Instantiate the exact same object again using the URL from
+ # the one that was already created properly
+ obj_cmp = 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.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
+ print('TEST FAIL: {} regenerated as {}'.format(
+ url, obj.url()))
+ assert(False)
+
if self:
# Iterate over our expected entries inside of our object
for key, val in self.items():
@@ -144,21 +170,27 @@ def test_plugin(mock_refresh, mock_send):
except Exception as e:
# We can't handle this exception type
- assert False
+ raise
except AssertionError:
# Don't mess with these entries
+ print('%s AssertionError' % url)
raise
except Exception as e:
# Check that we were expecting this exception to happen
- assert isinstance(e, response)
+ if not isinstance(e, response):
+ raise
except AssertionError:
# Don't mess with these entries
+ print('%s AssertionError' % url)
raise
except Exception as e:
# Handle our exception
- assert(instance is not None)
- assert(isinstance(e, instance))
+ if(instance is None):
+ raise
+
+ if not isinstance(e, instance):
+ raise
diff --git a/test/test_rest_plugins.py b/test/test_rest_plugins.py
index bee5fbea..f0f6d29a 100644
--- a/test/test_rest_plugins.py
+++ b/test/test_rest_plugins.py
@@ -263,42 +263,13 @@ TEST_URLS = (
('ifttt://EventID/', {
'instance': TypeError,
}),
- # Value1 gets assigned Entry1
- # Title =
- # Body =
- ('ifttt://WebHookID@EventID/Entry1/', {
- 'instance': plugins.NotifyIFTTT,
- }),
- # Value1, Value2, and Value2, the below assigns:
- # Value1 = Entry1
- # Value2 = AnotherEntry
- # Value3 = ThirdValue
- # Title =
- # Body =
- ('ifttt://WebHookID@EventID/Entry1/AnotherEntry/ThirdValue', {
- 'instance': plugins.NotifyIFTTT,
- }),
- # Mix and match content, the below assigns:
- # Value1 = FirstValue
- # AnotherKey = Hello
- # Value5 = test
- # Title =
- # Body =
- ('ifttt://WebHookID@EventID/FirstValue/?AnotherKey=Hello&Value5=test', {
- 'instance': plugins.NotifyIFTTT,
- }),
- # This would assign:
- # Value1 = FirstValue
- # Title = - disable the one passed by the notify call
- # Body = - disable the one passed by the notify call
- # The idea here is maybe you just want to use the apprise IFTTTT hook
- # to trigger something and not nessisarily pass text along to it
- ('ifttt://WebHookID@EventID/FirstValue/?Title=&Body=', {
- 'instance': plugins.NotifyIFTTT,
- }),
('ifttt://:@/', {
'instance': None,
}),
+ # A nicely formed ifttt url with 2 events defined:
+ ('ifttt://WebHookID@EventID/EventID2/', {
+ 'instance': plugins.NotifyIFTTT,
+ }),
# Test website connection failures
('ifttt://WebHookID@EventID', {
'instance': plugins.NotifyIFTTT,
@@ -392,6 +363,9 @@ TEST_URLS = (
('json://user:pass@localhost', {
'instance': plugins.NotifyJSON,
}),
+ ('json://user@localhost', {
+ 'instance': plugins.NotifyJSON,
+ }),
('json://localhost:8080', {
'instance': plugins.NotifyJSON,
}),
@@ -529,8 +503,8 @@ TEST_URLS = (
('matrix://user@localhost/%s' % ('a' * 64), {
'instance': plugins.NotifyMatrix,
}),
- # Name, port and token (secure)
- ('matrixs://user@localhost:9000/%s' % ('a' * 64), {
+ # port and token (secure)
+ ('matrixs://localhost:9000/%s' % ('a' * 64), {
'instance': plugins.NotifyMatrix,
}),
# Name, port, token and slack mode
@@ -963,6 +937,14 @@ TEST_URLS = (
('rocket://user:pass@localhost/#/!/@', {
'instance': TypeError,
}),
+ # No user/pass combo
+ ('rocket://user@localhost/room/', {
+ 'instance': TypeError,
+ }),
+ # No user/pass combo
+ ('rocket://localhost/room/', {
+ 'instance': TypeError,
+ }),
# A room and port identifier
('rocket://user:pass@localhost:8080/room/', {
'instance': plugins.NotifyRocketChat,
@@ -1383,7 +1365,7 @@ TEST_URLS = (
('xbmc://user:pass@localhost:8080', {
'instance': plugins.NotifyXBMC,
}),
- ('xbmc://localhost', {
+ ('xbmc://user@localhost', {
'instance': plugins.NotifyXBMC,
# don't include an image by default
'include_image': False,
@@ -1432,6 +1414,9 @@ TEST_URLS = (
('xml://localhost', {
'instance': plugins.NotifyXML,
}),
+ ('xml://user@localhost', {
+ 'instance': plugins.NotifyXML,
+ }),
('xml://user:pass@localhost', {
'instance': plugins.NotifyXML,
}),
@@ -1487,6 +1472,8 @@ def test_rest_plugins(mock_post, mock_get):
API: REST Based Plugins()
"""
+ # Disable Throttling to speed testing
+ plugins.NotifyBase.NotifyBase.throttle_attempt = 0
# iterate over our dictionary and test it out
for (url, meta) in TEST_URLS:
@@ -1567,10 +1554,31 @@ def test_rest_plugins(mock_post, mock_get):
assert instance is None
continue
+ if instance is None:
+ # Expected None but didn't get it
+ print('%s instantiated %s (but expected None)' % (
+ url, str(obj)))
+ assert(False)
+
assert(isinstance(obj, instance))
- # Disable throttling to speed up unit tests
- obj.throttle_attempt = 0
+ if isinstance(obj, plugins.NotifyBase.NotifyBase):
+ # We loaded okay; now lets make sure we can reverse this url
+ assert(compat_is_basestring(obj.url()) is True)
+
+ # Instantiate the exact same object again using the URL from
+ # the one that was already created properly
+ obj_cmp = 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.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
+ print('TEST FAIL: {} regenerated as {}'.format(
+ url, obj.url()))
+ assert(False)
if self:
# Iterate over our expected entries inside of our object
@@ -1605,8 +1613,7 @@ def test_rest_plugins(mock_post, mock_get):
except Exception as e:
# We can't handle this exception type
- print('%s / %s' % (url, str(e)))
- assert False
+ raise
except AssertionError:
# Don't mess with these entries
@@ -1615,7 +1622,8 @@ def test_rest_plugins(mock_post, mock_get):
except Exception as e:
# Check that we were expecting this exception to happen
- assert isinstance(e, response)
+ if not isinstance(e, response):
+ raise
#
# Stage 2: without title defined
@@ -1643,8 +1651,7 @@ def test_rest_plugins(mock_post, mock_get):
except Exception as e:
# We can't handle this exception type
- print('%s / %s' % (url, str(e)))
- assert False
+ raise
except AssertionError:
# Don't mess with these entries
@@ -1653,7 +1660,8 @@ def test_rest_plugins(mock_post, mock_get):
except Exception as e:
# Check that we were expecting this exception to happen
- assert isinstance(e, response)
+ if not isinstance(e, response):
+ raise
except AssertionError:
# Don't mess with these entries
@@ -1662,9 +1670,11 @@ def test_rest_plugins(mock_post, mock_get):
except Exception as e:
# Handle our exception
- print('%s / %s' % (url, str(e)))
- assert(instance is not None)
- assert(isinstance(e, instance))
+ if(instance is None):
+ raise
+
+ if not isinstance(e, instance):
+ raise
@mock.patch('requests.get')
@@ -1674,6 +1684,9 @@ def test_notify_boxcar_plugin(mock_post, mock_get):
API: NotifyBoxcar() Extra Checks
"""
+ # Disable Throttling to speed testing
+ plugins.NotifyBase.NotifyBase.throttle_attempt = 0
+
# Generate some generic message types
device = 'A' * 64
tag = '@B' * 63
@@ -1724,9 +1737,6 @@ def test_notify_boxcar_plugin(mock_post, mock_get):
# Test notifications without a body or a title
p = plugins.NotifyBoxcar(access=access, secret=secret, recipients=None)
- # Disable throttling to speed up unit tests
- p.throttle_attempt = 0
-
p.notify(body=None, title=None, notify_type=NotifyType.INFO) is True
@@ -1737,6 +1747,8 @@ def test_notify_discord_plugin(mock_post, mock_get):
API: NotifyDiscord() Extra Checks
"""
+ # Disable Throttling to speed testing
+ plugins.NotifyBase.NotifyBase.throttle_attempt = 0
# Initialize some generic (but valid) tokens
webhook_id = 'A' * 24
@@ -1762,9 +1774,6 @@ def test_notify_discord_plugin(mock_post, mock_get):
webhook_token=webhook_token,
footer=True, thumbnail=False)
- # Disable throttling to speed up unit tests
- obj.throttle_attempt = 0
-
# This call includes an image with it's payload:
assert obj.notify(title='title', body='body',
notify_type=NotifyType.INFO) is True
@@ -2190,10 +2199,12 @@ def test_notify_ifttt_plugin(mock_post, mock_get):
API: NotifyIFTTT() Extra Checks
"""
+ # Disable Throttling to speed testing
+ plugins.NotifyBase.NotifyBase.throttle_attempt = 0
# Initialize some generic (but valid) tokens
- apikey = 'webhookid'
- event = 'event'
+ webhook_id = 'webhookid'
+ events = ['event1', 'event2']
# Prepare Mock
mock_get.return_value = requests.Request()
@@ -2204,7 +2215,7 @@ def test_notify_ifttt_plugin(mock_post, mock_get):
mock_post.return_value.content = '{}'
try:
- obj = plugins.NotifyIFTTT(apikey=apikey, event=None, event_args=None)
+ obj = plugins.NotifyIFTTT(webhook_id=webhook_id, events=None)
# No token specified
assert(False)
@@ -2212,12 +2223,8 @@ def test_notify_ifttt_plugin(mock_post, mock_get):
# Exception should be thrown about the fact no token was specified
assert(True)
- obj = plugins.NotifyIFTTT(apikey=apikey, event=event, event_args=None)
+ obj = plugins.NotifyIFTTT(webhook_id=webhook_id, events=events)
assert(isinstance(obj, plugins.NotifyIFTTT))
- assert(len(obj.event_args) == 0)
-
- # Disable throttling to speed up unit tests
- obj.throttle_attempt = 0
assert obj.notify(title='title', body='body',
notify_type=NotifyType.INFO) is True
@@ -2230,6 +2237,9 @@ def test_notify_join_plugin(mock_post, mock_get):
API: NotifyJoin() Extra Checks
"""
+ # Disable Throttling to speed testing
+ plugins.NotifyBase.NotifyBase.throttle_attempt = 0
+
# Generate some generic message types
device = 'A' * 32
group = 'group.chrome'
@@ -2250,9 +2260,6 @@ def test_notify_join_plugin(mock_post, mock_get):
mock_post.return_value.status_code = requests.codes.created
mock_get.return_value.status_code = requests.codes.created
- # Disable throttling to speed up unit tests
- p.throttle_attempt = 0
-
# Test notifications without a body or a title; nothing to send
# so we return False
p.notify(body=None, title=None, notify_type=NotifyType.INFO) is False
@@ -2265,6 +2272,8 @@ def test_notify_slack_plugin(mock_post, mock_get):
API: NotifySlack() Extra Checks
"""
+ # Disable Throttling to speed testing
+ plugins.NotifyBase.NotifyBase.throttle_attempt = 0
# Initialize some generic (but valid) tokens
token_a = 'A' * 9
@@ -2300,9 +2309,6 @@ def test_notify_slack_plugin(mock_post, mock_get):
token_a=token_a, token_b=token_b, token_c=token_c, channels=channels,
include_image=True)
- # Disable throttling to speed up unit tests
- obj.throttle_attempt = 0
-
# This call includes an image with it's payload:
assert obj.notify(title='title', body='body',
notify_type=NotifyType.INFO) is True
@@ -2358,6 +2364,9 @@ def test_notify_pushed_plugin(mock_post, mock_get):
API: NotifyPushed() Extra Checks
"""
+ # Disable Throttling to speed testing
+ plugins.NotifyBase.NotifyBase.throttle_attempt = 0
+
# Chat ID
recipients = '@ABCDEFG, @DEFGHIJ, #channel, #channel2'
@@ -2436,9 +2445,6 @@ def test_notify_pushed_plugin(mock_post, mock_get):
assert(len(obj.channels) == 2)
assert(len(obj.users) == 2)
- # Disable throttling to speed up unit tests
- obj.throttle_attempt = 0
-
# Support the handling of an empty and invalid URL strings
assert plugins.NotifyPushed.parse_url(None) is None
assert plugins.NotifyPushed.parse_url('') is None
@@ -2458,6 +2464,8 @@ def test_notify_pushover_plugin(mock_post, mock_get):
API: NotifyPushover() Extra Checks
"""
+ # Disable Throttling to speed testing
+ plugins.NotifyBase.NotifyBase.throttle_attempt = 0
# Initialize some generic (but valid) tokens
token = 'a' * 30
@@ -2487,9 +2495,6 @@ def test_notify_pushover_plugin(mock_post, mock_get):
assert(isinstance(obj, plugins.NotifyPushover))
assert(len(obj.devices) == 3)
- # Disable throttling to speed up unit tests
- obj.throttle_attempt = 0
-
# This call fails because there is 1 invalid device
assert obj.notify(title='title', body='body',
notify_type=NotifyType.INFO) is False
@@ -2500,9 +2505,6 @@ def test_notify_pushover_plugin(mock_post, mock_get):
# device defined here
assert(len(obj.devices) == 1)
- # Disable throttling to speed up unit tests
- obj.throttle_attempt = 0
-
# This call succeeds because all of the devices are valid
assert obj.notify(title='title', body='body',
notify_type=NotifyType.INFO) is True
@@ -2526,9 +2528,16 @@ def test_notify_rocketchat_plugin(mock_post, mock_get):
API: NotifyRocketChat() Extra Checks
"""
+ # Disable Throttling to speed testing
+ plugins.NotifyBase.NotifyBase.throttle_attempt = 0
+
# Chat ID
recipients = 'l2g, lead2gold, #channel, #channel2'
+ # Authentication
+ user = 'myuser'
+ password = 'mypass'
+
# Prepare Mock
mock_get.return_value = requests.Request()
mock_post.return_value = requests.Request()
@@ -2538,7 +2547,8 @@ def test_notify_rocketchat_plugin(mock_post, mock_get):
mock_get.return_value.text = ''
try:
- obj = plugins.NotifyRocketChat(recipients=None)
+ obj = plugins.NotifyRocketChat(
+ user=user, password=password, recipients=None)
# invalid recipients list (None)
assert(False)
@@ -2548,7 +2558,8 @@ def test_notify_rocketchat_plugin(mock_post, mock_get):
assert(True)
try:
- obj = plugins.NotifyRocketChat(recipients=object())
+ obj = plugins.NotifyRocketChat(
+ user=user, password=password, recipients=object())
# invalid recipients list (object)
assert(False)
@@ -2558,7 +2569,8 @@ def test_notify_rocketchat_plugin(mock_post, mock_get):
assert(True)
try:
- obj = plugins.NotifyRocketChat(recipients=set())
+ obj = plugins.NotifyRocketChat(
+ user=user, password=password, recipients=set())
# invalid recipient list/set (no entries)
assert(False)
@@ -2567,14 +2579,12 @@ def test_notify_rocketchat_plugin(mock_post, mock_get):
# specified
assert(True)
- obj = plugins.NotifyRocketChat(recipients=recipients)
+ obj = plugins.NotifyRocketChat(
+ user=user, password=password, recipients=recipients)
assert(isinstance(obj, plugins.NotifyRocketChat))
assert(len(obj.channels) == 2)
assert(len(obj.rooms) == 2)
- # Disable throttling to speed up unit tests
- obj.throttle_attempt = 0
-
#
# Logout
#
@@ -2653,6 +2663,9 @@ def test_notify_telegram_plugin(mock_post, mock_get):
API: NotifyTelegram() Extra Checks
"""
+ # Disable Throttling to speed testing
+ plugins.NotifyBase.NotifyBase.throttle_attempt = 0
+
# Bot Token
bot_token = '123456789:abcdefg_hijklmnop'
invalid_bot_token = 'abcd:123'
@@ -2710,6 +2723,12 @@ def test_notify_telegram_plugin(mock_post, mock_get):
assert(isinstance(obj, plugins.NotifyTelegram))
assert(len(obj.chat_ids) == 2)
+ # test url call
+ assert(compat_is_basestring(obj.url()))
+ # Test that we can load the string we generate back:
+ obj = plugins.NotifyTelegram(**plugins.NotifyTelegram.parse_url(obj.url()))
+ assert(isinstance(obj, plugins.NotifyTelegram))
+
# Support the handling of an empty and invalid URL strings
assert(plugins.NotifyTelegram.parse_url(None) is None)
assert(plugins.NotifyTelegram.parse_url('') is None)
@@ -2730,10 +2749,6 @@ def test_notify_telegram_plugin(mock_post, mock_get):
nimg_obj = plugins.NotifyTelegram(bot_token=bot_token, chat_ids=chat_ids)
nimg_obj.asset = AppriseAsset(image_path_mask=False, image_url_mask=False)
- # Disable throttling to speed up unit tests
- nimg_obj.throttle_attempt = 0
- obj.throttle_attempt = 0
-
# Test that our default settings over-ride base settings since they are
# not the same as the one specified in the base; this check merely
# ensures our plugin inheritance is working properly
diff --git a/test/test_windows_plugin.py b/test/test_windows_plugin.py
index 1c577ede..bfe52632 100644
--- a/test/test_windows_plugin.py
+++ b/test/test_windows_plugin.py
@@ -29,6 +29,7 @@ import types
# Rebuild our Apprise environment
import apprise
+from apprise.utils import compat_is_basestring
try:
# Python v3.4+
@@ -107,6 +108,9 @@ def test_windows_plugin():
obj = apprise.Apprise.instantiate('windows://', suppress_exceptions=False)
obj.duration = 0
+ # Test URL functionality
+ assert(compat_is_basestring(obj.url()) is True)
+
# Check that it found our mocked environments
assert(obj._enabled is True)