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

Merge branch 'master' into 56-decommission-stride

Browse Source
This commit is contained in:
Chris Caron 2019-02-06 22:03:00 -05:00 committed by GitHub
commit bf1b61d23c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 557 additions and 521 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
README.md
View File

@ -43,13 +43,13 @@ The table below identifies the services this tool supports and some example serv
| [Matrix](https://github.com/caronc/apprise/wiki/Notify_matrix) | matrix:// | (TCP) 443 | matrix://token<br />matrix://user@token<br />matrixs://token?mode=slack<br />matrixs://user@token
| [Mattermost](https://github.com/caronc/apprise/wiki/Notify_mattermost) | mmost:// | (TCP) 8065 | mmost://hostname/authkey<br />mmost://hostname:80/authkey<br />mmost://user@hostname:80/authkey<br />mmost://hostname/authkey?channel=channel<br />mmosts://hostname/authkey<br />mmosts://user@hostname/authkey<br />
| [Prowl](https://github.com/caronc/apprise/wiki/Notify_prowl) | prowl:// | (TCP) 443 | prowl://apikey<br />prowl://apikey/providerkey
| [Pushalot](https://github.com/caronc/apprise/wiki/Notify_pushalot) | palot:// | (TCP) 443 | palot://authorizationtoken
| [PushBullet](https://github.com/caronc/apprise/wiki/Notify_pushbullet) | pbul:// | (TCP) 443 | pbul://accesstoken<br />pbul://accesstoken/#channel<br/>pbul://accesstoken/A_DEVICE_ID<br />pbul://accesstoken/email@address.com<br />pbul://accesstoken/#channel/#channel2/email@address.net/DEVICE
| [Pushjet](https://github.com/caronc/apprise/wiki/Notify_pushjet) | pjet:// | (TCP) 80 | pjet://secret<br />pjet://secret@hostname<br />pjet://secret@hostname:port<br />pjets://secret@hostname<br />pjets://secret@hostname:port<br /><i>Note: if no hostname defined https://api.pushjet.io will be used
| [Pushjet](https://github.com/caronc/apprise/wiki/Notify_pushjet) | pjet:// | (TCP) 80 | pjet://secret@hostname<br />pjet://secret@hostname:port<br />pjets://secret@hostname<br />pjets://secret@hostname:port
| [Pushed](https://github.com/caronc/apprise/wiki/Notify_pushed) | pushed:// | (TCP) 443 | pushed://appkey/appsecret/<br/>pushed://appkey/appsecret/#ChannelAlias<br/>pushed://appkey/appsecret/#ChannelAlias1/#ChannelAlias2/#ChannelAliasN<br/>pushed://appkey/appsecret/@UserPushedID<br/>pushed://appkey/appsecret/@UserPushedID1/@UserPushedID2/@UserPushedIDN
| [Pushover](https://github.com/caronc/apprise/wiki/Notify_pushover) | pover:// | (TCP) 443 | pover://user@token<br />pover://user@token/DEVICE<br />pover://user@token/DEVICE1/DEVICE2/DEVICEN<br />_Note: you must specify both your user_id and token_
| [Rocket.Chat](https://github.com/caronc/apprise/wiki/Notify_rocketchat) | rocket:// or rockets:// | (TCP) 80 or 443 | rocket://user:password@hostname/RoomID/Channel<br />rockets://user:password@hostname:443/Channel1/Channel1/RoomID<br />rocket://user:password@hostname/Channel
| [Slack](https://github.com/caronc/apprise/wiki/Notify_slack) | slack:// | (TCP) 443 | slack://TokenA/TokenB/TokenC/Channel<br />slack://botname@TokenA/TokenB/TokenC/Channel<br />slack://user@TokenA/TokenB/TokenC/Channel1/Channel2/ChannelN
| [Super Toasty](https://github.com/caronc/apprise/wiki/Notify_toasty) | toasty:// | (TCP) 80 | toasty://user@DEVICE<br />toasty://user@DEVICE1/DEVICE2/DEVICEN<br />_Note: you must specify both your user_id and at least 1 device!_
| [Stride](https://github.com/caronc/apprise/wiki/Notify_stride) | stride:// | (TCP) 443 | stride://auth_token/cloud_id/convo_id
| [Telegram](https://github.com/caronc/apprise/wiki/Notify_telegram) | tgram:// | (TCP) 443 | tgram://bottoken/ChatID<br />tgram://bottoken/ChatID1/ChatID2/ChatIDN
| [Twitter](https://github.com/caronc/apprise/wiki/Notify_twitter) | tweet:// | (TCP) 443 | tweet://user@CKey/CSecret/AKey/ASecret
| [XBMC](https://github.com/caronc/apprise/wiki/Notify_xbmc) | xbmc:// or xbmcs:// | (TCP) 8080 or 443 | xbmc://hostname<br />xbmc://user@hostname<br />xbmc://user:password@hostname:port

168
apprise/plugins/NotifyPushalot.py
View File

@ -1,168 +0,0 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2019 Chris Caron <lead2gold@gmail.com>
# All rights reserved.
#
# This code is licensed under the MIT License.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files(the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and / or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions :
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
import re
import requests
from json import dumps
from .NotifyBase import NotifyBase
from .NotifyBase import HTTP_ERROR_MAP
from ..common import NotifyImageSize
# Extend HTTP Error Messages
PUSHALOT_HTTP_ERROR_MAP = HTTP_ERROR_MAP.copy()
PUSHALOT_HTTP_ERROR_MAP.update({
406: 'Message throttle limit hit.',
410: 'AuthorizedToken is no longer valid.',
})
# Used to validate Authorization Token
VALIDATE_AUTHTOKEN = re.compile(r'[A-Za-z0-9]{32}')
class NotifyPushalot(NotifyBase):
"""
A wrapper for Pushalot Notifications
"""
# The default descriptive name associated with the Notification
service_name = 'Pushalot'
# The services URL
service_url = 'https://pushalot.com/'
# The default protocol is always secured
secure_protocol = 'palot'
# A URL that takes you to the setup/help of the specific protocol
setup_url = 'https://github.com/caronc/apprise/wiki/Notify_pushalot'
# Pushalot uses the http protocol with JSON requests
notify_url = 'https://pushalot.com/api/sendmessage'
# Allows the user to specify the NotifyImageSize object
image_size = NotifyImageSize.XY_72
def __init__(self, authtoken, is_important=False, **kwargs):
"""
Initialize Pushalot Object
"""
super(NotifyPushalot, self).__init__(**kwargs)
# Is Important Flag
self.is_important = is_important
self.authtoken = authtoken
# Validate authtoken
if not VALIDATE_AUTHTOKEN.match(authtoken):
self.logger.warning(
'Invalid Pushalot Authorization Token Specified.'
)
raise TypeError(
'Invalid Pushalot Authorization Token Specified.'
)
def notify(self, title, body, notify_type, **kwargs):
"""
Perform Pushalot Notification
"""
headers = {
'User-Agent': self.app_id,
'Content-Type': 'application/json'
}
# prepare JSON Object
payload = {
'AuthorizationToken': self.authtoken,
'IsImportant': self.is_important,
'Title': title,
'Body': body,
'Source': self.app_id,
}
image_url = self.image_url(notify_type)
if image_url:
payload['Image'] = image_url
self.logger.debug('Pushalot POST URL: %s (cert_verify=%r)' % (
self.notify_url, self.verify_certificate,
))
self.logger.debug('Pushalot Payload: %s' % str(payload))
try:
r = requests.post(
self.notify_url,
data=dumps(payload),
headers=headers,
verify=self.verify_certificate,
)
if r.status_code != requests.codes.ok:
# We had a problem
try:
self.logger.warning(
'Failed to send Pushalot notification: '
'%s (error=%s).' % (
PUSHALOT_HTTP_ERROR_MAP[r.status_code],
r.status_code))
except KeyError:
self.logger.warning(
'Failed to send Pushalot notification '
'(error=%s).' % r.status_code)
# Return; we're done
return False
else:
self.logger.info('Sent Pushalot notification.')
except requests.RequestException as e:
self.logger.warning(
'A Connection error occured sending Pushalot notification.')
self.logger.debug('Socket Exception: %s' % str(e))
# Return; we're done
return False
return True
@staticmethod
def parse_url(url):
"""
Parses the URL and returns enough arguments that can allow
us to substantiate this object.
"""
results = NotifyBase.parse_url(url)
if not results:
# We're done early as we couldn't load the results
return results
# Apply our settings now
results['authtoken'] = results['host']
return results

301
apprise/plugins/NotifyPushed.py Normal file
View File

@ -0,0 +1,301 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2019 Chris Caron <lead2gold@gmail.com>
# All rights reserved.
#
# This code is licensed under the MIT License.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files(the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and / or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions :
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
import re
import requests
from json import dumps
from .NotifyBase import NotifyBase
from .NotifyBase import HTTP_ERROR_MAP
from ..utils import compat_is_basestring
# Used to detect and parse channels
IS_CHANNEL = re.compile(r'^#(?P<name>[A-Za-z0-9]+)$')
# Used to detect and parse a users push id
IS_USER_PUSHED_ID = re.compile(r'^@(?P<name>[A-Za-z0-9]+)$')
# Used to break apart list of potential tags by their delimiter
# into a usable list.
LIST_DELIM = re.compile(r'[ \t\r\n,\\/]+')
class NotifyPushed(NotifyBase):
"""
A wrapper to Pushed Notifications
"""
# The default descriptive name associated with the Notification
service_name = 'Pushed'
# The services URL
service_url = 'https://pushed.co/'
# The default secure protocol
secure_protocol = 'pushed'
# A URL that takes you to the setup/help of the specific protocol
setup_url = 'https://github.com/caronc/apprise/wiki/Notify_pushed'
# Pushed uses the http protocol with JSON requests
notify_url = 'https://api.pushed.co/1/push'
# The maximum allowable characters allowed in the body per message
body_maxlen = 140
def __init__(self, app_key, app_secret, recipients=None, **kwargs):
"""
Initialize Pushed Object
"""
super(NotifyPushed, self).__init__(**kwargs)
if not app_key:
raise TypeError(
'An invalid Application Key was specified.'
)
if not app_secret:
raise TypeError(
'An invalid Application Secret was specified.'
)
# Initialize channel list
self.channels = list()
# Initialize user list
self.users = list()
if recipients is None:
recipients = []
elif compat_is_basestring(recipients):
recipients = [x for x in filter(bool, LIST_DELIM.split(
recipients,
))]
elif not isinstance(recipients, (set, tuple, list)):
raise TypeError(
'An invalid receipient list was specified.'
)
# Validate recipients and drop bad ones:
for recipient in recipients:
result = IS_CHANNEL.match(recipient)
if result:
# store valid device
self.channels.append(result.group('name'))
continue
result = IS_USER_PUSHED_ID.match(recipient)
if result:
# store valid room
self.users.append(result.group('name'))
continue
self.logger.warning(
'Dropped invalid channel/userid '
'(%s) specified.' % recipient,
)
# Store our data
self.app_key = app_key
self.app_secret = app_secret
return
def notify(self, title, body, notify_type, **kwargs):
"""
Perform Pushed Notification
"""
# Initiaize our error tracking
has_error = False
# prepare JSON Object
payload = {
'app_key': self.app_key,
'app_secret': self.app_secret,
'target_type': 'app',
'content': body,
}
# So the logic is as follows:
# - if no user/channel was specified, then we just simply notify the
# app.
# - if there are user/channels specified, then we only alert them
# while respecting throttle limits (in the event there are a lot of
# entries.
if len(self.channels) + len(self.users) == 0:
# Just notify the app
return self.send_notification(
payload=payload, notify_type=notify_type, **kwargs)
# If our code reaches here, we want to target channels and users (by
# their Pushed_ID instead...
# Generate a copy of our original list
channels = list(self.channels)
users = list(self.users)
# Copy our payload
_payload = dict(payload)
_payload['target_type'] = 'channel'
while len(channels) > 0:
# Get Channel
_payload['target_alias'] = channels.pop(0)
if not self.send_notification(
payload=_payload, notify_type=notify_type, **kwargs):
# toggle flag
has_error = True
if len(channels) + len(users) > 0:
# Prevent thrashing requests
self.throttle()
# Copy our payload
_payload = dict(payload)
_payload['target_type'] = 'pushed_id'
# Send all our defined User Pushed ID's
while len(users):
# Get User's Pushed ID
_payload['pushed_id'] = users.pop(0)
if not self.send_notification(
payload=_payload, notify_type=notify_type, **kwargs):
# toggle flag
has_error = True
if len(users) > 0:
# Prevent thrashing requests
self.throttle()
return not has_error
def send_notification(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.
"""
headers = {
'User-Agent': self.app_id,
'Content-Type': 'application/json'
}
self.logger.debug('Pushed POST URL: %s (cert_verify=%r)' % (
self.notify_url, self.verify_certificate,
))
self.logger.debug('Pushed Payload: %s' % str(payload))
try:
r = requests.post(
self.notify_url,
data=dumps(payload),
headers=headers,
verify=self.verify_certificate,
)
if r.status_code != requests.codes.ok:
# We had a problem
try:
self.logger.warning(
'Failed to send Pushed notification: '
'%s (error=%s).' % (
HTTP_ERROR_MAP[r.status_code],
r.status_code))
except KeyError:
self.logger.warning(
'Failed to send Pushed notification '
'(error=%s).' % r.status_code)
self.logger.debug('Response Details: %s' % r.raw.read())
# Return; we're done
return False
else:
self.logger.info('Sent Pushed notification.')
except requests.RequestException as e:
self.logger.warning(
'A Connection error occured sending Pushed notification.')
self.logger.debug('Socket Exception: %s' % str(e))
# Return; we're done
return False
return True
@staticmethod
def parse_url(url):
"""
Parses the URL and returns enough arguments that can allow
us to substantiate this object.
"""
results = NotifyBase.parse_url(url)
if not results:
# We're done early as we couldn't load the results
return results
# Apply our settings now
# The first token is stored in the hostname
app_key = results['host']
# Initialize our recipients
recipients = None
# Now fetch the remaining tokens
try:
app_secret = \
[x for x in filter(bool, NotifyBase.split_path(
results['fullpath']))][0]
except (ValueError, AttributeError, IndexError):
# Force some bad values that will get caught
# in parsing later
app_secret = None
app_key = None
# Get our recipients
recipients = \
[x for x in filter(bool, NotifyBase.split_path(
results['fullpath']))][1:]
results['app_key'] = app_key
results['app_secret'] = app_secret
results['recipients'] = recipients
return results

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

@ -43,9 +43,6 @@ class NotifyPushjet(NotifyBase):
# The default descriptive name associated with the Notification
service_name = 'Pushjet'
# The services URL
service_url = 'https://pushjet.io/'
# The default protocol
protocol = 'pjet'
@ -66,21 +63,16 @@ class NotifyPushjet(NotifyBase):
Perform Pushjet Notification
"""
try:
if self.user and self.host:
server = "http://"
if self.secure:
server = "https://"
server = "http://"
if self.secure:
server = "https://"
server += self.host
if self.port:
server += ":" + str(self.port)
server += self.host
if self.port:
server += ":" + str(self.port)
api = pushjet.Api(server)
service = api.Service(secret_key=self.user)
else:
api = pushjet.Api(pushjet.DEFAULT_API_URL)
service = api.Service(secret_key=self.host)
api = pushjet.Api(server)
service = api.Service(secret_key=self.user)
service.send(body, title)
self.logger.info('Sent Pushjet notification.')
@ -91,3 +83,28 @@ class NotifyPushjet(NotifyBase):
return False
return True
@staticmethod
def parse_url(url):
"""
Parses the URL and returns enough arguments that can allow
us to substantiate this object.
Syntax:
pjet://secret@hostname
pjet://secret@hostname:port
pjets://secret@hostname
pjets://secret@hostname:port
"""
results = NotifyBase.parse_url(url)
if not results:
# We're done early as we couldn't load the results
return results
if not results.get('user'):
# a username is required
return None
return results

187
apprise/plugins/NotifyToasty.py
View File

@ -1,187 +0,0 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2019 Chris Caron <lead2gold@gmail.com>
# All rights reserved.
#
# This code is licensed under the MIT License.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files(the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and / or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions :
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
import re
import requests
from .NotifyBase import NotifyBase
from .NotifyBase import HTTP_ERROR_MAP
from ..common import NotifyImageSize
from ..utils import compat_is_basestring
# Used to break apart list of potential devices by their delimiter
# into a usable list.
DEVICES_LIST_DELIM = re.compile(r'[ \t\r\n,\\/]+')
class NotifyToasty(NotifyBase):
"""
A wrapper for Toasty Notifications
"""
# The default descriptive name associated with the Notification
service_name = 'Toasty'
# The services URL
service_url = 'http://supertoasty.com/'
# The default protocol
protocol = 'toasty'
# A URL that takes you to the setup/help of the specific protocol
setup_url = 'https://github.com/caronc/apprise/wiki/Notify_toasty'
# Toasty uses the http protocol with JSON requests
notify_url = 'http://api.supertoasty.com/notify/'
# Allows the user to specify the NotifyImageSize object
image_size = NotifyImageSize.XY_128
def __init__(self, devices, **kwargs):
"""
Initialize Toasty Object
"""
super(NotifyToasty, self).__init__(**kwargs)
if compat_is_basestring(devices):
self.devices = [x for x in filter(bool, DEVICES_LIST_DELIM.split(
devices,
))]
elif isinstance(devices, (set, tuple, list)):
self.devices = devices
else:
self.devices = list()
if len(devices) == 0:
raise TypeError('You must specify at least 1 device.')
if not self.user:
raise TypeError('You must specify a username.')
def notify(self, title, body, notify_type, **kwargs):
"""
Perform Toasty Notification
"""
headers = {
'User-Agent': self.app_id,
'Content-Type': 'multipart/form-data',
}
# error tracking (used for function return)
has_error = False
# Create a copy of the devices list
devices = list(self.devices)
while len(devices):
device = devices.pop(0)
# prepare JSON Object
payload = {
'sender': NotifyBase.quote(self.user),
'title': NotifyBase.quote(title),
'text': NotifyBase.quote(body),
}
image_url = self.image_url(notify_type)
if image_url:
payload['image'] = image_url
# URL to transmit content via
url = '%s%s' % (self.notify_url, device)
self.logger.debug('Toasty POST URL: %s (cert_verify=%r)' % (
url, self.verify_certificate,
))
self.logger.debug('Toasty Payload: %s' % str(payload))
try:
r = requests.get(
url,
data=payload,
headers=headers,
verify=self.verify_certificate,
)
if r.status_code != requests.codes.ok:
# We had a problem
try:
self.logger.warning(
'Failed to send Toasty:%s '
'notification: %s (error=%s).' % (
device,
HTTP_ERROR_MAP[r.status_code],
r.status_code))
except KeyError:
self.logger.warning(
'Failed to send Toasty:%s '
'notification (error=%s).' % (
device,
r.status_code))
# self.logger.debug('Response Details: %s' % r.raw.read())
# Return; we're done
has_error = True
else:
self.logger.info(
'Sent Toasty notification to %s.' % device)
except requests.RequestException as e:
self.logger.warning(
'A Connection error occured sending Toasty:%s ' % (
device) + 'notification.'
)
self.logger.debug('Socket Exception: %s' % str(e))
has_error = True
if len(devices):
# Prevent thrashing requests
self.throttle()
return not has_error
@staticmethod
def parse_url(url):
"""
Parses the URL and returns enough arguments that can allow
us to substantiate this object.
"""
results = NotifyBase.parse_url(url)
if not results:
# We're done early as we couldn't load the results
return results
# Apply our settings now
devices = NotifyBase.unquote(results['fullpath'])
# Store our devices
results['devices'] = '%s/%s' % (results['host'], devices)
return results

11
apprise/plugins/__init__.py
View File

@ -40,14 +40,13 @@ from .NotifyJSON import NotifyJSON
from .NotifyMatrix import NotifyMatrix
from .NotifyMatterMost import NotifyMatterMost
from .NotifyProwl import NotifyProwl
from .NotifyPushalot import NotifyPushalot
from .NotifyPushed import NotifyPushed
from .NotifyPushBullet import NotifyPushBullet
from .NotifyPushjet.NotifyPushjet import NotifyPushjet
from .NotifyPushover import NotifyPushover
from .NotifyRocketChat import NotifyRocketChat
from .NotifySlack import NotifySlack
from .NotifyTelegram import NotifyTelegram
from .NotifyToasty import NotifyToasty
from .NotifyTwitter.NotifyTwitter import NotifyTwitter
from .NotifyXBMC import NotifyXBMC
from .NotifyXML import NotifyXML
@ -67,10 +66,10 @@ __all__ = [
'NotifyBoxcar', 'NotifyEmail', 'NotifyEmby', 'NotifyDiscord',
'NotifyFaast', 'NotifyGnome', 'NotifyGrowl', 'NotifyIFTTT', 'NotifyJoin',
'NotifyJSON', 'NotifyMatrix', 'NotifyMatterMost', 'NotifyProwl',
'NotifyPushalot', 'NotifyPushBullet', 'NotifyPushjet', 'NotifyPushover',
'NotifyRocketChat', 'NotifySlack', 'NotifyToasty',
'NotifyTwitter', 'NotifyTelegram', 'NotifyXBMC', 'NotifyXML',
'NotifyWindows',
'NotifyPushed', 'NotifyPushBullet', 'NotifyPushjet',
'NotifyPushover', 'NotifyRocketChat', 'NotifySlack',
'NotifyTwitter', 'NotifyTelegram', 'NotifyXBMC',
'NotifyXML', 'NotifyWindows',
# Reference
'NotifyImageSize', 'NOTIFY_IMAGE_SIZES', 'NotifyType', 'NOTIFY_TYPES',

5
test/test_api.py
View File

@ -65,19 +65,18 @@ def test_apprise():
servers = [
'faast://abcdefghijklmnop-abcdefg',
'kodi://kodi.server.local',
'palot://1f418df7577e32b89ac6511f2eb9aa68',
]
a = Apprise(servers=servers)
# 3 servers loaded
assert(len(a) == 3)
assert(len(a) == 2)
# We can add another server
assert(
a.add('mmosts://mattermost.server.local/'
'3ccdd113474722377935511fc85d3dd4') is True)
assert(len(a) == 4)
assert(len(a) == 3)
# We can empty our set
a.clear()

4
test/test_pushjet_plugin.py
View File

@ -38,9 +38,9 @@ TEST_URLS = (
('pjets://', {
'instance': None,
}),
# Default query (uses pushjet server)
# You must specify a username
('pjet://%s' % ('a' * 32), {
'instance': plugins.NotifyPushjet,
'instance': None,
}),
# Specify your own server
('pjet://%s@localhost' % ('a' * 32), {

347
test/test_rest_plugins.py
View File

@ -686,50 +686,6 @@ TEST_URLS = (
'test_requests_exceptions': True,
}),
##################################
# NotifyPushalot
##################################
('palot://', {
'instance': None,
}),
# AuthToken
('palot://%s' % ('a' * 32), {
'instance': plugins.NotifyPushalot,
}),
# AuthToken, no image
('palot://%s' % ('a' * 32), {
'instance': plugins.NotifyPushalot,
# don't include an image by default
'include_image': False,
}),
# Invalid AuthToken
('palot://%s' % ('a' * 24), {
# Missing a channel
'instance': TypeError,
}),
# AuthToken + bad url
('palot://:@/', {
'instance': None,
}),
('palot://%s' % ('a' * 32), {
'instance': plugins.NotifyPushalot,
# force a failure
'response': False,
'requests_response_code': requests.codes.internal_server_error,
}),
('palot://%s' % ('a' * 32), {
'instance': plugins.NotifyPushalot,
# throw a bizzare code forcing us to fail to look it up
'response': False,
'requests_response_code': 999,
}),
('palot://%s' % ('a' * 32), {
'instance': plugins.NotifyPushalot,
# Throws a series of connection and transfer exceptions when this flag
# is set and tests that we gracfully handle them
'test_requests_exceptions': True,
}),
##################################
# NotifyPushBullet
##################################
@ -768,7 +724,25 @@ TEST_URLS = (
('pbul://%s/device/#channel/user@example.com/' % ('a' * 32), {
'instance': plugins.NotifyPushBullet,
}),
# APIKey + bad url
# ,
('pbul://%s' % ('a' * 32), {
'instance': plugins.NotifyPushBullet,
# force a failure
'response': False,
'requests_response_code': requests.codes.internal_server_error,
}),
('pbul://%s' % ('a' * 32), {
'instance': plugins.NotifyPushBullet,
# throw a bizzare code forcing us to fail to look it up
'response': False,
'requests_response_code': 999,
}),
('pbul://%s' % ('a' * 32), {
'instance': plugins.NotifyPushBullet,
# Throws a series of connection and transfer exceptions when this flag
# is set and tests that we gracfully handle them
'test_requests_exceptions': True,
}),
('pbul://:@/', {
'instance': None,
}),
@ -791,6 +765,98 @@ TEST_URLS = (
'test_requests_exceptions': True,
}),
##################################
# NotifyPushed
##################################
('pushed://', {
'instance': None,
}),
# Application Key Only
('pushed://%s' % ('a' * 32), {
'instance': TypeError,
}),
# Application Key+Secret
('pushed://%s/%s' % ('a' * 32, 'a' * 64), {
'instance': plugins.NotifyPushed,
}),
# Application Key+Secret + channel
('pushed://%s/%s/#channel/' % ('a' * 32, 'a' * 64), {
'instance': plugins.NotifyPushed,
}),
# Application Key+Secret + dropped entry
('pushed://%s/%s/dropped/' % ('a' * 32, 'a' * 64), {
'instance': plugins.NotifyPushed,
}),
# Application Key+Secret + 2 channels
('pushed://%s/%s/#channel1/#channel2' % ('a' * 32, 'a' * 64), {
'instance': plugins.NotifyPushed,
}),
# Application Key+Secret + User Pushed ID
('pushed://%s/%s/@ABCD/' % ('a' * 32, 'a' * 64), {
'instance': plugins.NotifyPushed,
}),
# Application Key+Secret + 2 devices
('pushed://%s/%s/@ABCD/@DEFG/' % ('a' * 32, 'a' * 64), {
'instance': plugins.NotifyPushed,
}),
# Application Key+Secret + Combo
('pushed://%s/%s/@ABCD/#channel' % ('a' * 32, 'a' * 64), {
'instance': plugins.NotifyPushed,
}),
# ,
('pushed://%s/%s' % ('a' * 32, 'a' * 64), {
'instance': plugins.NotifyPushed,
# force a failure
'response': False,
'requests_response_code': requests.codes.internal_server_error,
}),
('pushed://%s/%s' % ('a' * 32, 'a' * 64), {
'instance': plugins.NotifyPushed,
# throw a bizzare code forcing us to fail to look it up
'response': False,
'requests_response_code': 999,
}),
('pushed://%s/%s' % ('a' * 32, 'a' * 64), {
'instance': plugins.NotifyPushed,
# Throws a series of connection and transfer exceptions when this flag
# is set and tests that we gracfully handle them
'test_requests_exceptions': True,
}),
('pushed://:@/', {
'instance': None,
}),
('pushed://%s/%s' % ('a' * 32, 'a' * 64), {
'instance': plugins.NotifyPushed,
# force a failure
'response': False,
'requests_response_code': requests.codes.internal_server_error,
}),
('pushed://%s/%s' % ('a' * 32, 'a' * 64), {
'instance': plugins.NotifyPushed,
# throw a bizzare code forcing us to fail to look it up
'response': False,
'requests_response_code': 999,
}),
('pushed://%s/%s/#channel' % ('a' * 32, 'a' * 64), {
'instance': plugins.NotifyPushed,
# throw a bizzare code forcing us to fail to look it up
'response': False,
'requests_response_code': 999,
}),
('pushed://%s/%s/@user' % ('a' * 32, 'a' * 64), {
'instance': plugins.NotifyPushed,
# throw a bizzare code forcing us to fail to look it up
'response': False,
'requests_response_code': 999,
}),
('pushed://%s/%s' % ('a' * 32, 'a' * 64), {
'instance': plugins.NotifyPushed,
# Throws a series of connection and transfer exceptions when this flag
# is set and tests that we gracfully handle them
'test_requests_exceptions': True,
}),
##################################
# NotifyPushover
##################################
@ -1193,49 +1259,6 @@ TEST_URLS = (
'test_requests_exceptions': True,
}),
##################################
# NotifyToasty (SuperToasty)
##################################
('toasty://', {
'instance': None,
}),
# No username specified but contains a device
('toasty://%s' % ('d' * 32), {
'instance': TypeError,
}),
# User + 1 device
('toasty://user@device', {
'instance': plugins.NotifyToasty,
}),
# User + 3 devices
('toasty://user@device0/device1/device2/', {
'instance': plugins.NotifyToasty,
# don't include an image by default
'include_image': False,
}),
# bad url
('toasty://:@/', {
'instance': None,
}),
('toasty://user@device', {
'instance': plugins.NotifyToasty,
# force a failure
'response': False,
'requests_response_code': requests.codes.internal_server_error,
}),
('toasty://user@device', {
'instance': plugins.NotifyToasty,
# throw a bizzare code forcing us to fail to look it up
'response': False,
'requests_response_code': 999,
}),
('toasty://user@device', {
'instance': plugins.NotifyToasty,
# Throws a series of connection and transfer exceptions when this flag
# is set and tests that we gracfully handle them
'test_requests_exceptions': True,
}),
##################################
# NotifyKODI
##################################
@ -2213,6 +2236,106 @@ def test_notify_pushbullet_plugin(mock_post, mock_get):
assert(plugins.NotifyPushBullet.parse_url(42) is None)
@mock.patch('requests.get')
@mock.patch('requests.post')
def test_notify_pushed_plugin(mock_post, mock_get):
"""
API: NotifyPushed() Extra Checks
"""
# Chat ID
recipients = '@ABCDEFG, @DEFGHIJ, #channel, #channel2'
# Some required input
app_key = 'ABCDEFG'
app_secret = 'ABCDEFG'
# Prepare Mock
mock_get.return_value = requests.Request()
mock_post.return_value = requests.Request()
mock_post.return_value.status_code = requests.codes.ok
mock_get.return_value.status_code = requests.codes.ok
mock_post.return_value.text = ''
mock_get.return_value.text = ''
try:
obj = plugins.NotifyPushed(
app_key=app_key,
app_secret=None,
recipients=None,
)
assert(False)
except TypeError:
# No application Secret was specified; it's a good thing if
# this exception was thrown
assert(True)
try:
obj = plugins.NotifyPushed(
app_key=app_key,
app_secret=app_secret,
recipients=None,
)
# recipients list set to (None) is perfectly fine; in this
# case it will notify the App
assert(True)
except TypeError:
# Exception should never be thrown!
assert(False)
try:
obj = plugins.NotifyPushed(
app_key=app_key,
app_secret=app_secret,
recipients=object(),
)
# invalid recipients list (object)
assert(False)
except TypeError:
# Exception should be thrown about the fact no recipients were
# specified
assert(True)
try:
obj = plugins.NotifyPushed(
app_key=app_key,
app_secret=app_secret,
recipients=set(),
)
# Any empty set is acceptable
assert(True)
except TypeError:
# Exception should never be thrown
assert(False)
obj = plugins.NotifyPushed(
app_key=app_key,
app_secret=app_secret,
recipients=recipients,
)
assert(isinstance(obj, plugins.NotifyPushed))
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
assert plugins.NotifyPushed.parse_url(42) is None
# Prepare Mock to fail
mock_post.return_value.status_code = requests.codes.internal_server_error
mock_get.return_value.status_code = requests.codes.internal_server_error
mock_post.return_value.text = ''
mock_get.return_value.text = ''
@mock.patch('requests.get')
@mock.patch('requests.post')
def test_notify_pushover_plugin(mock_post, mock_get):
@ -2408,54 +2531,6 @@ def test_notify_rocketchat_plugin(mock_post, mock_get):
assert obj.logout() is False
@mock.patch('requests.get')
@mock.patch('requests.post')
def test_notify_toasty_plugin(mock_post, mock_get):
"""
API: NotifyToasty() Extra Checks
"""
# Support strings
devices = 'device1,device2,,,,'
# User
user = 'l2g'
# Prepare Mock
mock_get.return_value = requests.Request()
mock_post.return_value = requests.Request()
mock_post.return_value.status_code = requests.codes.ok
mock_get.return_value.status_code = requests.codes.ok
try:
obj = plugins.NotifyToasty(user=user, devices=None)
# No devices specified
assert(False)
except TypeError:
# Exception should be thrown about the fact no token was specified
assert(True)
try:
obj = plugins.NotifyToasty(user=user, devices=set())
# No devices specified
assert(False)
except TypeError:
# Exception should be thrown about the fact no token was specified
assert(True)
obj = plugins.NotifyToasty(user=user, devices=devices)
assert(isinstance(obj, plugins.NotifyToasty))
assert(len(obj.devices) == 2)
# Support the handling of an empty and invalid URL strings
assert(plugins.NotifyToasty.parse_url(None) is None)
assert(plugins.NotifyToasty.parse_url('') is None)
assert(plugins.NotifyToasty.parse_url(42) is None)
@mock.patch('requests.get')
@mock.patch('requests.post')
def test_notify_telegram_plugin(mock_post, mock_get):