extern/apprise
1
0
Fork 0
mirror of https://github.com/caronc/apprise.git synced 2025-06-25 20:21:59 +02:00
Code Issues Packages Projects Releases Wiki Activity

NotifyBase() home of notify(); calls send() in children

Browse Source
This commit is contained in:
Chris Caron 2019-02-17 22:21:59 -05:00
parent 7d9715aa5a
commit eb4c83f3f8
33 changed files with 281 additions and 166 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
README.md
View File

@ -110,8 +110,8 @@ apobj.add('pbul://o.gn5kj6nfhv736I7jC3cj3QLRiyhgl98b')
# Then notify these services any time you desire. The below would
# notify all of the services loaded into our Apprise object.
apobj.notify(
title='my notification title',
body='what a great notification service!',
title='my notification title',
)
```

39
apprise/Apprise.py
View File

@ -238,7 +238,7 @@ class Apprise(object):
"""
self.servers[:] = []
def notify(self, title, body, notify_type=NotifyType.INFO,
def notify(self, body, title='', notify_type=NotifyType.INFO,
body_format=None, tag=None):
"""
Send a notification to all of the plugins previously loaded.
@ -363,30 +363,25 @@ class Apprise(object):
# Store entry directly
conversion_map[server.notify_format] = body
# Apply our overflow (if defined)
for chunk in server._apply_overflow(
body=conversion_map[server.notify_format], title=title):
try:
# Send notification
if not server.notify(
title=chunk['title'],
body=chunk['body'],
notify_type=notify_type):
try:
# Send notification
if not server.notify(
body=conversion_map[server.notify_format],
title=title,
notify_type=notify_type):
# Toggle our return status flag
status = False
except TypeError:
# These our our internally thrown notifications
# TODO: Change this to a custom one such as
# AppriseNotifyError
# Toggle our return status flag
status = False
except Exception:
# A catch all so we don't have to abort early
# just because one of our plugins has a bug in it.
logging.exception("Notification Exception")
status = False
except TypeError:
# These our our internally thrown notifications
status = False
except Exception:
# A catch all so we don't have to abort early
# just because one of our plugins has a bug in it.
logging.exception("Notification Exception")
status = False
return status

77
apprise/plugins/NotifyBase.py
View File

@ -44,6 +44,7 @@ from ..utils import parse_url
from ..utils import parse_bool
from ..utils import parse_list
from ..utils import is_hostname
from ..common import NotifyType
from ..common import NOTIFY_TYPES
from ..common import NotifyFormat
from ..common import NOTIFY_FORMATS
@ -194,16 +195,16 @@ class NotifyBase(object):
if 'overflow' in kwargs:
# Store the specified format if specified
overflow_mode = kwargs.get('overflow', '')
if overflow_mode.lower() not in OVERFLOW_MODES:
overflow = kwargs.get('overflow', '')
if overflow.lower() not in OVERFLOW_MODES:
self.logger.error(
'Invalid overflow method %s' % overflow_mode,
'Invalid overflow method %s' % overflow,
)
raise TypeError(
'Invalid overflow method %s' % overflow_mode,
'Invalid overflow method %s' % overflow,
)
# Provide override
self.overflow_mode = overflow_mode
self.overflow_mode = overflow
if 'tag' in kwargs:
# We want to associate some tags with our notification service.
@ -314,7 +315,29 @@ class NotifyBase(object):
color_type=color_type,
)
def _apply_overflow(self, body, title=None):
def notify(self, body, title=None, notify_type=NotifyType.INFO,
overflow=None, **kwargs):
"""
Performs notification
"""
# Handle situations where the title is None
title = '' if not title else title
# Apply our overflow (if defined)
for chunk in self._apply_overflow(body=body, title=title,
overflow=overflow):
# Send notification
if not self.send(body=chunk['body'], title=chunk['title'],
notify_type=notify_type):
# Toggle our return status flag
return False
return True
def _apply_overflow(self, body, title=None, overflow=None):
"""
Takes the message body and title as input. This function then
applies any defined overflow restrictions associated with the
@ -337,6 +360,14 @@ class NotifyBase(object):
response = list()
# safety
title = '' if not title else title.strip()
body = '' if not body else body.rstrip()
if overflow is None:
# default
overflow = self.overflow_mode
if self.title_maxlen <= 0:
# Content is appended to body
body = '{}\r\n{}'.format(title, body)
@ -349,12 +380,9 @@ class NotifyBase(object):
body = re.split('\r*\n', body)
body = '\r\n'.join(body[0:self.body_max_line_count])
if self.overflow_mode == OverflowMode.UPSTREAM:
# Nothing to do
response.append({
'body': body,
'title': title,
})
if overflow == OverflowMode.UPSTREAM:
# Nothing more to do
response.append({'body': body, 'title': title})
return response
elif len(title) > self.title_maxlen:
@ -362,13 +390,10 @@ class NotifyBase(object):
title = title[:self.title_maxlen]
if self.body_maxlen > 0 and len(body) <= self.body_maxlen:
response.append({
'body': body,
'title': title,
})
response.append({'body': body, 'title': title})
return response
if self.overflow_mode == OverflowMode.TRUNCATE:
if overflow == OverflowMode.TRUNCATE:
# Truncate our body and return
response.append({
'body': body[:self.body_maxlen],
@ -386,13 +411,20 @@ class NotifyBase(object):
return response
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
Should preform the actual notification itself.
"""
raise NotImplementedError("send() is implimented by the child class.")
def url(self):
"""
Assembles the URL associated with the notification based on the
arguments provied.
"""
raise NotImplementedError("This is implimented by the child class.")
raise NotImplementedError("url() is implimented by the child class.")
def __contains__(self, tags):
"""
@ -555,6 +587,15 @@ class NotifyBase(object):
results['format']))
del results['format']
# Allow overriding the default overflow
if 'overflow' in results['qsd']:
results['overflow'] = results['qsd'].get('overflow')
if results['overflow'] not in OVERFLOW_MODES:
NotifyBase.logger.warning(
'Unsupported overflow specified {}'.format(
results['overflow']))
del results['overflow']
# Password overrides
if 'pass' in results['qsd']:
results['password'] = results['qsd']['pass']

10
apprise/plugins/NotifyBoxcar.py
View File

@ -23,11 +23,11 @@
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
from json import dumps
import requests
import re
from time import time
import requests
import hmac
from json import dumps
from time import time
from hashlib import sha1
from itertools import chain
try:
@ -38,7 +38,7 @@ except ImportError:
from .NotifyBase import NotifyBase
from .NotifyBase import HTTP_ERROR_MAP
from ..common import NotifyType
from ..common import NotifyImageSize
from ..utils import compat_is_basestring
@ -169,7 +169,7 @@ class NotifyBoxcar(NotifyBase):
'(%s) specified.' % recipient,
)
def notify(self, title, body, notify_type, **kwargs):
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
Perform Boxcar Notification
"""

5
apprise/plugins/NotifyDBus.py
View File

@ -26,10 +26,9 @@
from __future__ import absolute_import
from __future__ import print_function
import re
from .NotifyBase import NotifyBase
from ..common import NotifyImageSize
from ..common import NotifyType
from ..utils import GET_SCHEMA_RE
# Default our global support flag
@ -201,7 +200,7 @@ class NotifyDBus(NotifyBase):
self.x_axis = x_axis
self.y_axis = y_axis
def notify(self, title, body, notify_type, **kwargs):
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
Perform DBus Notification
"""

3
apprise/plugins/NotifyDiscord.py
View File

@ -48,6 +48,7 @@ from .NotifyBase import NotifyBase
from .NotifyBase import HTTP_ERROR_MAP
from ..common import NotifyImageSize
from ..common import NotifyFormat
from ..common import NotifyType
from ..utils import parse_bool
@ -113,7 +114,7 @@ class NotifyDiscord(NotifyBase):
return
def notify(self, title, body, notify_type, **kwargs):
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
Perform Discord Notification
"""

9
apprise/plugins/NotifyEmail.py
View File

@ -24,15 +24,14 @@
# THE SOFTWARE.
import re
from datetime import datetime
import smtplib
from socket import error as SocketError
from email.mime.text import MIMEText
from socket import error as SocketError
from datetime import datetime
from .NotifyBase import NotifyBase
from ..common import NotifyFormat
from ..common import NotifyType
class WebBaseLogin(object):
@ -344,7 +343,7 @@ class NotifyEmail(NotifyBase):
break
def notify(self, title, body, **kwargs):
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
Perform Email Notification
"""

3
apprise/plugins/NotifyEmby.py
View File

@ -37,6 +37,7 @@ from json import loads
from .NotifyBase import NotifyBase
from .NotifyBase import HTTP_ERROR_MAP
from ..utils import parse_bool
from ..common import NotifyType
from .. import __version__ as VERSION
@ -445,7 +446,7 @@ class NotifyEmby(NotifyBase):
self.user_id = None
return True
def notify(self, title, body, notify_type, **kwargs):
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
Perform Emby Notification
"""

3
apprise/plugins/NotifyFaast.py
View File

@ -27,6 +27,7 @@ import requests
from .NotifyBase import NotifyBase
from .NotifyBase import HTTP_ERROR_MAP
from ..common import NotifyImageSize
from ..common import NotifyType
class NotifyFaast(NotifyBase):
@ -60,7 +61,7 @@ class NotifyFaast(NotifyBase):
self.authtoken = authtoken
def notify(self, title, body, notify_type, **kwargs):
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
Perform Faast Notification
"""

3
apprise/plugins/NotifyGnome.py
View File

@ -28,6 +28,7 @@ from __future__ import print_function
from .NotifyBase import NotifyBase
from ..common import NotifyImageSize
from ..common import NotifyType
# Default our global support flag
NOTIFY_GNOME_SUPPORT_ENABLED = False
@ -119,7 +120,7 @@ class NotifyGnome(NotifyBase):
else:
self.urgency = urgency
def notify(self, title, body, notify_type, **kwargs):
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
Perform Gnome Notification
"""

25
apprise/plugins/NotifyGrowl/NotifyGrowl.py
View File

@ -23,12 +23,11 @@
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
import re
from .gntp import notifier
from .gntp import errors
from ..NotifyBase import NotifyBase
from ...common import NotifyImageSize
from ...common import NotifyType
# Priorities
@ -69,16 +68,24 @@ class NotifyGrowl(NotifyBase):
# A URL that takes you to the setup/help of the specific protocol
setup_url = 'https://github.com/caronc/apprise/wiki/Notify_growl'
# Allows the user to specify the NotifyImageSize object
image_size = NotifyImageSize.XY_72
# Disable throttle rate for Growl requests since they are normally
# local anyway
request_rate_per_sec = 0
# A title can not be used for Growl Messages. Setting this to zero will
# cause any title (if defined) to get placed into the message body.
title_maxlen = 0
# Limit results to just the first 10 line otherwise there is just to much
# content to display
body_max_line_count = 2
# Default Growl Port
default_port = 23053
# Allows the user to specify the NotifyImageSize object
image_size = NotifyImageSize.XY_72
def __init__(self, priority=None, version=2, **kwargs):
"""
Initialize Growl Object
@ -147,17 +154,11 @@ class NotifyGrowl(NotifyBase):
return
def notify(self, title, body, notify_type, **kwargs):
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
Perform Growl Notification
"""
# Limit results to just the first 2 line otherwise there is just to
# much content to display
body = re.split('[\r\n]+', body)
body[0] = body[0].strip('#').strip()
body = '\r\n'.join(body[0:2])
icon = None
if self.version >= 2:
# URL Based

5
apprise/plugins/NotifyIFTTT.py
View File

@ -40,10 +40,11 @@
# For each event you create you will assign it a name (this will be known as
# the {event} when building your URL.
import requests
from json import dumps
from .NotifyBase import NotifyBase
from .NotifyBase import HTTP_ERROR_MAP
from ..common import NotifyType
from ..utils import parse_list
@ -136,7 +137,7 @@ class NotifyIFTTT(NotifyBase):
'del_token must be a list; {} was provided'.format(
str(type(del_tokens))))
def notify(self, title, body, notify_type, **kwargs):
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
Perform IFTTT Notification
"""

3
apprise/plugins/NotifyJSON.py
View File

@ -29,6 +29,7 @@ from json import dumps
from .NotifyBase import NotifyBase
from .NotifyBase import HTTP_ERROR_MAP
from ..common import NotifyImageSize
from ..common import NotifyType
from ..utils import compat_is_basestring
@ -120,7 +121,7 @@ class NotifyJSON(NotifyBase):
args=self.urlencode(args),
)
def notify(self, title, body, notify_type, **kwargs):
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
Perform JSON Notification
"""

3
apprise/plugins/NotifyJoin.py
View File

@ -39,6 +39,7 @@ import requests
from .NotifyBase import NotifyBase
from .NotifyBase import HTTP_ERROR_MAP
from ..common import NotifyImageSize
from ..common import NotifyType
from ..utils import compat_is_basestring
# Token required as part of the API request
@ -130,7 +131,7 @@ class NotifyJoin(NotifyBase):
# Default to everyone
self.devices.append('group.all')
def notify(self, title, body, notify_type, **kwargs):
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
Perform Join Notification
"""

3
apprise/plugins/NotifyMatrix.py
View File

@ -30,6 +30,7 @@ from time import time
from .NotifyBase import NotifyBase
from .NotifyBase import HTTP_ERROR_MAP
from ..common import NotifyType
# Token required as part of the API request
VALIDATE_TOKEN = re.compile(r'[A-Za-z0-9]{64}')
@ -134,7 +135,7 @@ class NotifyMatrix(NotifyBase):
re.IGNORECASE,
)
def notify(self, title, body, notify_type, **kwargs):
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
Perform Matrix Notification
"""

3
apprise/plugins/NotifyMatterMost.py
View File

@ -30,6 +30,7 @@ from json import dumps
from .NotifyBase import NotifyBase
from .NotifyBase import HTTP_ERROR_MAP
from ..common import NotifyImageSize
from ..common import NotifyType
# Some Reference Locations:
# - https://docs.mattermost.com/developer/webhooks-incoming.html
@ -111,7 +112,7 @@ class NotifyMatterMost(NotifyBase):
return
def notify(self, title, body, notify_type, **kwargs):
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
Perform MatterMost Notification
"""

3
apprise/plugins/NotifyProwl.py
View File

@ -28,6 +28,7 @@ import requests
from .NotifyBase import NotifyBase
from .NotifyBase import HTTP_ERROR_MAP
from ..common import NotifyType
# Used to validate API Key
VALIDATE_APIKEY = re.compile(r'[A-Za-z0-9]{40}')
@ -128,7 +129,7 @@ class NotifyProwl(NotifyBase):
# Store the Provider Key
self.providerkey = providerkey
def notify(self, title, body, **kwargs):
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
Perform Prowl Notification
"""

4
apprise/plugins/NotifyPushBullet.py
View File

@ -30,7 +30,7 @@ from json import dumps
from .NotifyBase import NotifyBase
from .NotifyBase import HTTP_ERROR_MAP
from .NotifyBase import IS_EMAIL_RE
from ..common import NotifyType
from ..utils import compat_is_basestring
# Flag used as a placeholder to sending to all devices
@ -87,7 +87,7 @@ class NotifyPushBullet(NotifyBase):
if len(self.recipients) == 0:
self.recipients = (PUSHBULLET_SEND_TO_ALL, )
def notify(self, title, body, **kwargs):
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
Perform PushBullet Notification
"""

13
apprise/plugins/NotifyPushed.py
View File

@ -30,6 +30,7 @@ from itertools import chain
from .NotifyBase import NotifyBase
from .NotifyBase import HTTP_ERROR_MAP
from ..common import NotifyType
from ..utils import compat_is_basestring
# Used to detect and parse channels
@ -128,7 +129,7 @@ class NotifyPushed(NotifyBase):
return
def notify(self, title, body, notify_type, **kwargs):
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
Perform Pushed Notification
"""
@ -153,7 +154,7 @@ class NotifyPushed(NotifyBase):
if len(self.channels) + len(self.users) == 0:
# Just notify the app
return self.send_notification(
return self._send(
payload=payload, notify_type=notify_type, **kwargs)
# If our code reaches here, we want to target channels and users (by
@ -171,7 +172,7 @@ class NotifyPushed(NotifyBase):
# Get Channel
_payload['target_alias'] = channels.pop(0)
if not self.send_notification(
if not self._send(
payload=_payload, notify_type=notify_type, **kwargs):
# toggle flag
@ -186,7 +187,7 @@ class NotifyPushed(NotifyBase):
# Get User's Pushed ID
_payload['pushed_id'] = users.pop(0)
if not self.send_notification(
if not self._send(
payload=_payload, notify_type=notify_type, **kwargs):
# toggle flag
@ -194,11 +195,11 @@ class NotifyPushed(NotifyBase):
return not has_error
def send_notification(self, payload, notify_type, **kwargs):
def _send(self, payload, notify_type, **kwargs):
"""
A lower level call that directly pushes a payload to the Pushed
Notification servers. This should never be called directly; it is
referenced automatically through the notify() function.
referenced automatically through the send() function.
"""
headers = {

3
apprise/plugins/NotifyPushjet/NotifyPushjet.py
View File

@ -28,6 +28,7 @@ from .pushjet import errors
from .pushjet import pushjet
from ..NotifyBase import NotifyBase
from ...common import NotifyType
PUBLIC_KEY_RE = re.compile(
r'^[a-z0-9]{4}-[a-z0-9]{6}-[a-z0-9]{12}-[a-z0-9]{5}-[a-z0-9]{9}$', re.I)
@ -65,7 +66,7 @@ class NotifyPushjet(NotifyBase):
# store our key
self.secret_key = secret_key
def notify(self, title, body, notify_type):
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
Perform Pushjet Notification
"""

5
apprise/plugins/NotifyPushover.py
View File

@ -26,9 +26,10 @@
import re
import requests
from ..utils import compat_is_basestring
from .NotifyBase import NotifyBase
from .NotifyBase import HTTP_ERROR_MAP
from ..common import NotifyType
from ..utils import compat_is_basestring
# Flag used as a placeholder to sending to all devices
PUSHOVER_SEND_TO_ALL = 'ALL_DEVICES'
@ -149,7 +150,7 @@ class NotifyPushover(NotifyBase):
'The user/group specified (%s) is invalid.' % self.user,
)
def notify(self, title, body, **kwargs):
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
Perform Pushover Notification
"""

11
apprise/plugins/NotifyRocketChat.py
View File

@ -30,6 +30,7 @@ from itertools import chain
from .NotifyBase import NotifyBase
from .NotifyBase import HTTP_ERROR_MAP
from ..common import NotifyType
from ..utils import compat_is_basestring
IS_CHANNEL = re.compile(r'^#(?P<name>[A-Za-z0-9]+)$')
@ -178,9 +179,9 @@ class NotifyRocketChat(NotifyBase):
args=self.urlencode(args),
)
def notify(self, title, body, notify_type, **kwargs):
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
wrapper to send_notification since we can alert more then one channel
wrapper to _send since we can alert more then one channel
"""
# Track whether we authenticated okay
@ -202,7 +203,7 @@ class NotifyRocketChat(NotifyBase):
# Get Channel
channel = channels.pop(0)
if not self.send_notification(
if not self._send(
{
'text': text,
'channel': channel,
@ -216,7 +217,7 @@ class NotifyRocketChat(NotifyBase):
# Get Room
room = rooms.pop(0)
if not self.send_notification(
if not self._send(
{
'text': text,
'roomId': room,
@ -230,7 +231,7 @@ class NotifyRocketChat(NotifyBase):
return not has_error
def send_notification(self, payload, notify_type, **kwargs):
def _send(self, payload, notify_type, **kwargs):
"""
Perform Notify Rocket.Chat Notification
"""

3
apprise/plugins/NotifyRyver.py
View File

@ -38,6 +38,7 @@ from json import dumps
from .NotifyBase import NotifyBase
from .NotifyBase import HTTP_ERROR_MAP
from ..common import NotifyImageSize
from ..common import NotifyType
# Token required as part of the API request
VALIDATE_TOKEN = re.compile(r'[A-Za-z0-9]{15}')
@ -141,7 +142,7 @@ class NotifyRyver(NotifyBase):
re.IGNORECASE,
)
def notify(self, title, body, notify_type, **kwargs):
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
Perform Ryver Notification
"""

6
apprise/plugins/NotifySNS.py
View File

@ -24,7 +24,6 @@
# THE SOFTWARE.
import re
import hmac
import requests
from hashlib import sha256
@ -35,6 +34,7 @@ from itertools import chain
from .NotifyBase import NotifyBase
from .NotifyBase import HTTP_ERROR_MAP
from ..common import NotifyType
from ..utils import compat_is_basestring
# Some Phone Number Detection
@ -194,7 +194,7 @@ class NotifySNS(NotifyBase):
self.logger.warning(
'There are no valid recipient identified to notify.')
def notify(self, title, body, notify_type, **kwargs):
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
wrapper to send_notification since we can alert more then one channel
"""
@ -266,7 +266,7 @@ class NotifySNS(NotifyBase):
def _post(self, payload, to):
"""
Wrapper to request.post() to manage it's response better and make
the notify() function cleaner and easier to maintain.
the send() function cleaner and easier to maintain.
This function returns True if the _post was successful and False
if it wasn't.

3
apprise/plugins/NotifySlack.py
View File

@ -43,6 +43,7 @@ from time import time
from .NotifyBase import NotifyBase
from .NotifyBase import HTTP_ERROR_MAP
from ..common import NotifyImageSize
from ..common import NotifyType
from ..utils import compat_is_basestring
# Token required as part of the API request
@ -174,7 +175,7 @@ class NotifySlack(NotifyBase):
re.IGNORECASE,
)
def notify(self, title, body, notify_type, **kwargs):
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
Perform Slack Notification
"""

5
apprise/plugins/NotifyTelegram.py
View File

@ -59,10 +59,11 @@ from json import dumps
from .NotifyBase import NotifyBase
from .NotifyBase import HTTP_ERROR_MAP
from ..common import NotifyType
from ..common import NotifyImageSize
from ..common import NotifyFormat
from ..utils import parse_bool
from ..utils import parse_list
from ..common import NotifyFormat
TELEGRAM_IMAGE_XY = NotifyImageSize.XY_256
@ -325,7 +326,7 @@ class NotifyTelegram(NotifyBase):
return 0
def notify(self, title, body, notify_type, **kwargs):
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
Perform Telegram Notification
"""

3
apprise/plugins/NotifyTwitter/NotifyTwitter.py
View File

@ -25,6 +25,7 @@
from . import tweepy
from ..NotifyBase import NotifyBase
from ...common import NotifyType
class NotifyTwitter(NotifyBase):
@ -93,7 +94,7 @@ class NotifyTwitter(NotifyBase):
return
def notify(self, title, body, notify_type, **kwargs):
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
Perform Twitter Notification
"""

3
apprise/plugins/NotifyWindows.py
View File

@ -30,6 +30,7 @@ from time import sleep
from .NotifyBase import NotifyBase
from ..common import NotifyImageSize
from ..common import NotifyType
# Default our global support flag
NOTIFY_WINDOWS_SUPPORT_ENABLED = False
@ -107,7 +108,7 @@ class NotifyWindows(NotifyBase):
return None
def notify(self, title, body, notify_type, **kwargs):
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
Perform Windows Notification
"""

2
apprise/plugins/NotifyXBMC.py
View File

@ -161,7 +161,7 @@ class NotifyXBMC(NotifyBase):
return (self.headers, dumps(payload))
def notify(self, title, body, notify_type, **kwargs):
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
Perform XBMC/KODI Notification
"""

3
apprise/plugins/NotifyXML.py
View File

@ -29,6 +29,7 @@ import requests
from .NotifyBase import NotifyBase
from .NotifyBase import HTTP_ERROR_MAP
from ..common import NotifyImageSize
from ..common import NotifyType
from ..utils import compat_is_basestring
@ -135,7 +136,7 @@ class NotifyXML(NotifyBase):
args=self.urlencode(args),
)
def notify(self, title, body, notify_type, **kwargs):
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
Perform XML Notification
"""

27
test/test_email_plugin.py
View File

@ -352,28 +352,15 @@ def test_smtplib_init_fail(mock_smtplib):
assert(isinstance(obj, plugins.NotifyEmail))
# Support Exception handling of smtplib.SMTP
mock_smtplib.side_effect = TypeError('Test')
mock_smtplib.side_effect = RuntimeError('Test')
try:
obj.notify(
title='test', body='body',
notify_type=NotifyType.INFO)
# We should have thrown an exception
assert False
except TypeError:
# Exception thrown as expected
assert True
except Exception:
# Un-Expected
assert False
assert obj.notify(
body='body', title='test', notify_type=NotifyType.INFO) is False
# A handled and expected exception
mock_smtplib.side_effect = smtplib.SMTPException('Test')
assert obj.notify(title='test', body='body',
notify_type=NotifyType.INFO) is False
assert obj.notify(
body='body', title='test', notify_type=NotifyType.INFO) is False
@mock.patch('smtplib.SMTP')
@ -397,7 +384,7 @@ def test_smtplib_send_okay(mock_smtplib):
mock_smtplib.quit.return_value = True
assert(obj.notify(
title='test', body='body', notify_type=NotifyType.INFO) is True)
body='body', title='test', notify_type=NotifyType.INFO) is True)
# Set Text
obj = Apprise.instantiate(
@ -405,4 +392,4 @@ def test_smtplib_send_okay(mock_smtplib):
assert(isinstance(obj, plugins.NotifyEmail))
assert(obj.notify(
title='test', body='body', notify_type=NotifyType.INFO) is True)
body='body', title='test', notify_type=NotifyType.INFO) is True)

44
test/test_notify_base.py
View File

@ -48,6 +48,15 @@ def test_notify_base():
except TypeError:
assert(True)
# invalid types throw exceptions
try:
nb = NotifyBase(**{'overflow': 'invalid'})
# We should never reach here as an exception should be thrown
assert(False)
except TypeError:
assert(True)
# Bad port information
nb = NotifyBase(port='invalid')
assert nb.port is None
@ -65,6 +74,16 @@ def test_notify_base():
# implemented error intentionally
assert True
try:
nb.send('test message')
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.request_rate_per_sec = 0.0
@ -223,6 +242,31 @@ def test_notify_base_urls():
assert 'password' in results
assert results['password'] == "newpassword"
# Options
results = NotifyBase.parse_url('https://localhost?format=invalid')
assert 'format' not in results
results = NotifyBase.parse_url('https://localhost?format=text')
assert 'format' in results
assert results['format'] == 'text'
results = NotifyBase.parse_url('https://localhost?format=markdown')
assert 'format' in results
assert results['format'] == 'markdown'
results = NotifyBase.parse_url('https://localhost?format=html')
assert 'format' in results
assert results['format'] == 'html'
results = NotifyBase.parse_url('https://localhost?overflow=invalid')
assert 'overflow' not in results
results = NotifyBase.parse_url('https://localhost?overflow=upstream')
assert 'overflow' in results
assert results['overflow'] == 'upstream'
results = NotifyBase.parse_url('https://localhost?overflow=split')
assert 'overflow' in results
assert results['overflow'] == 'split'
results = NotifyBase.parse_url('https://localhost?overflow=truncate')
assert 'overflow' in results
assert results['overflow'] == 'truncate'
# User Handling
# user keyword over-rides default password

113
test/test_rest_plugins.py
View File

@ -1489,6 +1489,20 @@ def test_rest_plugins(mock_post, mock_get):
# Disable Throttling to speed testing
plugins.NotifyBase.NotifyBase.request_rate_per_sec = 0
# Define how many characters exist per line
row = 80
# Some variables we use to control the data we work with
body_len = 1024
title_len = 1024
# Create a large body and title with random data
body = ''.join(choice(str_alpha + str_num + ' ') for _ in range(body_len))
body = '\r\n'.join([body[i: i + row] for i in range(0, len(body), row)])
# Create our title using random data
title = ''.join(choice(str_alpha + str_num) for _ in range(title_len))
# iterate over our dictionary and test it out
for (url, meta) in TEST_URLS:
# Our expected instance
@ -1611,9 +1625,24 @@ def test_rest_plugins(mock_post, mock_get):
# check that we're as expected
assert obj.notify(
title='test', body='body',
body=body, title=title,
notify_type=notify_type) == response
# check that this doesn't change using different overflow
# methods
assert obj.notify(
body=body, title=title,
notify_type=notify_type,
overflow=OverflowMode.UPSTREAM) == response
assert obj.notify(
body=body, title=title,
notify_type=notify_type,
overflow=OverflowMode.TRUNCATE) == response
assert obj.notify(
body=body, title=title,
notify_type=notify_type,
overflow=OverflowMode.SPLIT) == response
else:
# Disable throttling
obj.request_rate_per_sec = 0
@ -1624,7 +1653,7 @@ def test_rest_plugins(mock_post, mock_get):
try:
assert obj.notify(
title='test', body='body',
body=body, title=title,
notify_type=NotifyType.INFO) is False
except AssertionError:
@ -1652,8 +1681,7 @@ def test_rest_plugins(mock_post, mock_get):
if test_requests_exceptions is False:
# check that we're as expected
assert obj.notify(
title='', body='body',
notify_type=notify_type) == response
body='body', notify_type=notify_type) == response
else:
for _exception in REQUEST_EXCEPTIONS:
@ -1662,7 +1690,7 @@ def test_rest_plugins(mock_post, mock_get):
try:
assert obj.notify(
title='', body='body',
body=body,
notify_type=NotifyType.INFO) is False
except AssertionError:
@ -1795,8 +1823,8 @@ def test_notify_discord_plugin(mock_post, mock_get):
footer=True, thumbnail=False)
# This call includes an image with it's payload:
assert obj.notify(title='title', body='body',
notify_type=NotifyType.INFO) is True
assert obj.notify(
body='body', title='title', notify_type=NotifyType.INFO) is True
# Test our header parsing
test_markdown = "## Heading one\nbody body\n\n" + \
@ -1814,8 +1842,8 @@ def test_notify_discord_plugin(mock_post, mock_get):
assert(len(results) == 5)
# Use our test markdown string during a notification
assert obj.notify(title='title', body=test_markdown,
notify_type=NotifyType.INFO) is True
assert obj.notify(
body=test_markdown, title='title', notify_type=NotifyType.INFO) is True
# Create an apprise instance
a = Apprise()
@ -1829,18 +1857,18 @@ def test_notify_discord_plugin(mock_post, mock_get):
webhook_token=webhook_token)) is True
# This call includes an image with it's payload:
assert a.notify(title='title', body=test_markdown,
assert a.notify(body=test_markdown, title='title',
notify_type=NotifyType.INFO,
body_format=NotifyFormat.TEXT) is True
assert a.notify(title='title', body=test_markdown,
assert a.notify(body=test_markdown, title='title',
notify_type=NotifyType.INFO,
body_format=NotifyFormat.MARKDOWN) is True
# Toggle our logo availability
a.asset.image_url_logo = None
assert a.notify(title='title', body='body',
notify_type=NotifyType.INFO) is True
assert a.notify(
body='body', title='title', notify_type=NotifyType.INFO) is True
@mock.patch('requests.get')
@ -2254,8 +2282,8 @@ def test_notify_ifttt_plugin(mock_post, mock_get):
obj = plugins.NotifyIFTTT(webhook_id=webhook_id, events=events)
assert(isinstance(obj, plugins.NotifyIFTTT))
assert obj.notify(title='title', body='body',
notify_type=NotifyType.INFO) is True
assert obj.notify(
body='body', title='title', notify_type=NotifyType.INFO) is True
# Test the addition of tokens
obj = plugins.NotifyIFTTT(
@ -2264,8 +2292,8 @@ def test_notify_ifttt_plugin(mock_post, mock_get):
assert(isinstance(obj, plugins.NotifyIFTTT))
assert obj.notify(title='title', body='body',
notify_type=NotifyType.INFO) is True
assert obj.notify(
body='body', title='title', notify_type=NotifyType.INFO) is True
try:
# Invalid del_tokens entry
@ -2283,8 +2311,8 @@ def test_notify_ifttt_plugin(mock_post, mock_get):
assert(isinstance(obj, plugins.NotifyIFTTT))
assert obj.notify(title='title', body='body',
notify_type=NotifyType.INFO) is True
assert obj.notify(
body='body', title='title', notify_type=NotifyType.INFO) is True
# Test removal of tokens by a list
obj = plugins.NotifyIFTTT(
@ -2299,8 +2327,8 @@ def test_notify_ifttt_plugin(mock_post, mock_get):
assert(isinstance(obj, plugins.NotifyIFTTT))
assert obj.notify(title='title', body='body',
notify_type=NotifyType.INFO) is True
assert obj.notify(
body='body', title='title', notify_type=NotifyType.INFO) is True
@mock.patch('requests.get')
@ -2383,8 +2411,8 @@ def test_notify_slack_plugin(mock_post, mock_get):
include_image=True)
# This call includes an image with it's payload:
assert obj.notify(title='title', body='body',
notify_type=NotifyType.INFO) is True
assert obj.notify(
body='body', title='title', notify_type=NotifyType.INFO) is True
@mock.patch('requests.get')
@ -2571,8 +2599,8 @@ def test_notify_pushover_plugin(mock_post, mock_get):
assert(len(obj.devices) == 3)
# This call fails because there is 1 invalid device
assert obj.notify(title='title', body='body',
notify_type=NotifyType.INFO) is False
assert obj.notify(
body='body', title='title', notify_type=NotifyType.INFO) is False
obj = plugins.NotifyPushover(user=user, token=token)
assert(isinstance(obj, plugins.NotifyPushover))
@ -2581,8 +2609,8 @@ def test_notify_pushover_plugin(mock_post, mock_get):
assert(len(obj.devices) == 1)
# This call succeeds because all of the devices are valid
assert obj.notify(title='title', body='body',
notify_type=NotifyType.INFO) is True
assert obj.notify(
body='body', title='title', notify_type=NotifyType.INFO) is True
obj = plugins.NotifyPushover(user=user, token=token, devices=set())
assert(isinstance(obj, plugins.NotifyPushover))
@ -2680,9 +2708,8 @@ def test_notify_rocketchat_plugin(mock_post, mock_get):
# Send Notification
#
assert obj.notify(
title='title', body='body', notify_type=NotifyType.INFO) is False
assert obj.send_notification(
payload='test', notify_type=NotifyType.INFO) is False
body='body', title='title', notify_type=NotifyType.INFO) is False
assert obj._send(payload='test', notify_type=NotifyType.INFO) is False
#
# Logout
@ -2697,9 +2724,8 @@ def test_notify_rocketchat_plugin(mock_post, mock_get):
# Send Notification
#
assert obj.notify(
title='title', body='body', notify_type=NotifyType.INFO) is False
assert obj.send_notification(
payload='test', notify_type=NotifyType.INFO) is False
body='body', title='title', notify_type=NotifyType.INFO) is False
assert obj._send(payload='test', notify_type=NotifyType.INFO) is False
#
# Logout
@ -2717,14 +2743,13 @@ def test_notify_rocketchat_plugin(mock_post, mock_get):
#
# Send Notification
#
assert obj.send_notification(
payload='test', notify_type=NotifyType.INFO) is False
assert obj._send(payload='test', notify_type=NotifyType.INFO) is False
# Attempt the check again but fake a successful login
obj.login = mock.Mock()
obj.login.return_value = True
assert obj.notify(
title='title', body='body', notify_type=NotifyType.INFO) is False
body='body', title='title', notify_type=NotifyType.INFO) is False
#
# Logout
#
@ -2835,9 +2860,11 @@ def test_notify_telegram_plugin(mock_post, mock_get):
# This tests erroneous messages involving multiple chat ids
assert obj.notify(
title='title', body='body', notify_type=NotifyType.INFO) is False
body='body', title='title', notify_type=NotifyType.INFO) is False
assert obj.notify(
body='body', title='title', notify_type=NotifyType.INFO) is False
assert nimg_obj.notify(
title='title', body='body', notify_type=NotifyType.INFO) is False
body='body', title='title', notify_type=NotifyType.INFO) is False
# This tests erroneous messages involving a single chat id
obj = plugins.NotifyTelegram(bot_token=bot_token, chat_ids='l2g')
@ -2845,9 +2872,9 @@ def test_notify_telegram_plugin(mock_post, mock_get):
nimg_obj.asset = AppriseAsset(image_path_mask=False, image_url_mask=False)
assert obj.notify(
title='title', body='body', notify_type=NotifyType.INFO) is False
body='body', title='title', notify_type=NotifyType.INFO) is False
assert nimg_obj.notify(
title='title', body='body', notify_type=NotifyType.INFO) is False
body='body', title='title', notify_type=NotifyType.INFO) is False
# Bot Token Detection
# Just to make it clear to people reading this code and trying to learn
@ -2945,7 +2972,7 @@ def test_notify_telegram_plugin(mock_post, mock_get):
# notification without a bot detection by providing at least 1 chat id
obj = plugins.NotifyTelegram(bot_token=bot_token, chat_ids=['@abcd'])
assert nimg_obj.notify(
title='title', body='body', notify_type=NotifyType.INFO) is False
body='body', title='title', notify_type=NotifyType.INFO) is False
# iterate over our exceptions and test them
for _exception in REQUEST_EXCEPTIONS:
@ -3023,7 +3050,9 @@ def test_notify_overflow_truncate():
# Verify that we break the title to a max length of our title_max
# and that the body remains untouched
chunks = obj._apply_overflow(body=body, title=title)
chunks = obj._apply_overflow(body=body, title=title, overflow=None)
chunks = obj._apply_overflow(
body=body, title=title, overflow=OverflowMode.SPLIT)
assert len(chunks) == 1
assert body == chunks[0].get('body')
assert title[0:TestNotification.title_maxlen] == chunks[0].get('title')