mirror of
https://github.com/caronc/apprise.git
synced 2024-11-22 16:13:12 +01:00
100% test coverage
This commit is contained in:
Chris Caron
parent
cc79763b3f
commit
a03525a859
@ -15,12 +15,17 @@
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU Lesser General Public License for more details.
|
||||
|
||||
import re
|
||||
from .pushjet import errors
|
||||
from .pushjet import pushjet
|
||||
|
||||
from ..NotifyBase import NotifyBase
|
||||
|
||||
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)
|
||||
|
||||
SECRET_KEY_RE = re.compile(r'^[a-z0-9]{32}$', re.I)
|
||||
|
||||
|
||||
class NotifyPushjet(NotifyBase):
|
||||
"""
|
||||
|
||||
@ -74,16 +74,16 @@ class NotifyRocketChat(NotifyBase):
|
||||
# Initialize channels list
|
||||
self.channels = list()
|
||||
|
||||
# Initialize room_id list
|
||||
self.room_ids = list()
|
||||
# Initialize room list
|
||||
self.rooms = list()
|
||||
|
||||
if recipients is None:
|
||||
recipients = []
|
||||
|
||||
elif compat_is_basestring(recipients):
|
||||
recipients = filter(bool, LIST_DELIM.split(
|
||||
recipients = [x for x in filter(bool, LIST_DELIM.split(
|
||||
recipients,
|
||||
))
|
||||
))]
|
||||
|
||||
elif not isinstance(recipients, (set, tuple, list)):
|
||||
recipients = []
|
||||
@ -98,62 +98,88 @@ class NotifyRocketChat(NotifyBase):
|
||||
|
||||
result = IS_ROOM_ID.match(recipient)
|
||||
if result:
|
||||
# store valid room_id
|
||||
self.channels.append(result.group('name'))
|
||||
# store valid room
|
||||
self.rooms.append(result.group('name'))
|
||||
continue
|
||||
|
||||
self.logger.warning(
|
||||
'Dropped invalid channel/room_id ' +
|
||||
'Dropped invalid channel/room ' +
|
||||
'(%s) specified.' % recipient,
|
||||
)
|
||||
|
||||
if len(self.room_ids) == 0 and len(self.channels) == 0:
|
||||
if len(self.rooms) == 0 and len(self.channels) == 0:
|
||||
raise TypeError(
|
||||
'No Rocket.Chat room_id and/or channels specified to notify.'
|
||||
'No Rocket.Chat room and/or channels specified to notify.'
|
||||
)
|
||||
|
||||
# Used to track token headers upon authentication (if successful)
|
||||
self.headers = {}
|
||||
|
||||
# Track whether we authenticated okay
|
||||
self.authenticated = self.login()
|
||||
|
||||
if not self.authenticated:
|
||||
raise TypeError(
|
||||
'Authentication to Rocket.Chat server failed.'
|
||||
)
|
||||
|
||||
def notify(self, title, body, notify_type, **kwargs):
|
||||
"""
|
||||
wrapper to send_notification since we can alert more then one channel
|
||||
"""
|
||||
|
||||
# Track whether we authenticated okay
|
||||
|
||||
if not self.login():
|
||||
return False
|
||||
|
||||
# Prepare our message
|
||||
text = '*%s*\r\n%s' % (title.replace('*', '\*'), body)
|
||||
|
||||
# Send all our defined channels
|
||||
for channel in self.channels:
|
||||
self.send_notification({
|
||||
'text': text,
|
||||
'channel': channel,
|
||||
}, notify_type=notify_type, **kwargs)
|
||||
# Initiaize our error tracking
|
||||
has_error = False
|
||||
|
||||
# Create a copy of our rooms and channels to notify against
|
||||
channels = list(self.channels)
|
||||
rooms = list(self.rooms)
|
||||
|
||||
while len(channels) > 0:
|
||||
# Get Channel
|
||||
channel = channels.pop(0)
|
||||
|
||||
if not self.send_notification(
|
||||
{
|
||||
'text': text,
|
||||
'channel': channel,
|
||||
}, notify_type=notify_type, **kwargs):
|
||||
|
||||
# toggle flag
|
||||
has_error = True
|
||||
|
||||
if len(channels) + len(rooms) > 0:
|
||||
# Prevent thrashing requests
|
||||
self.throttle()
|
||||
|
||||
# Send all our defined room id's
|
||||
for room_id in self.room_ids:
|
||||
self.send_notification({
|
||||
'text': text,
|
||||
'roomId': room_id,
|
||||
}, notify_type=notify_type, **kwargs)
|
||||
while len(rooms):
|
||||
# Get Room
|
||||
room = rooms.pop(0)
|
||||
|
||||
if not self.send_notification(
|
||||
{
|
||||
'text': text,
|
||||
'roomId': room,
|
||||
}, notify_type=notify_type, **kwargs):
|
||||
|
||||
# toggle flag
|
||||
has_error = True
|
||||
|
||||
if len(rooms) > 0:
|
||||
# Prevent thrashing requests
|
||||
self.throttle()
|
||||
|
||||
# logout
|
||||
self.logout()
|
||||
|
||||
return not has_error
|
||||
|
||||
def send_notification(self, payload, notify_type, **kwargs):
|
||||
"""
|
||||
Perform Notify Rocket.Chat Notification
|
||||
"""
|
||||
|
||||
if not self.authenticated:
|
||||
# We couldn't authenticate; we're done
|
||||
return False
|
||||
|
||||
self.logger.debug('Rocket.Chat POST URL: %s (cert_verify=%r)' % (
|
||||
self.api_url + 'chat.postMessage', self.verify_certificate,
|
||||
))
|
||||
@ -173,6 +199,7 @@ class NotifyRocketChat(NotifyBase):
|
||||
'%s (error=%s).' % (
|
||||
RC_HTTP_ERROR_MAP[r.status_code],
|
||||
r.status_code))
|
||||
|
||||
except KeyError:
|
||||
self.logger.warning(
|
||||
'Failed to send Rocket.Chat notification ' +
|
||||
@ -200,6 +227,7 @@ class NotifyRocketChat(NotifyBase):
|
||||
def login(self):
|
||||
"""
|
||||
login to our server
|
||||
|
||||
"""
|
||||
payload = {
|
||||
'username': self.user,
|
||||
@ -220,7 +248,8 @@ class NotifyRocketChat(NotifyBase):
|
||||
'%s (error=%s).' % (
|
||||
RC_HTTP_ERROR_MAP[r.status_code],
|
||||
r.status_code))
|
||||
except IndexError:
|
||||
|
||||
except KeyError:
|
||||
self.logger.warning(
|
||||
'Failed to authenticate with Rocket.Chat server ' +
|
||||
'(error=%s).' % (
|
||||
@ -238,15 +267,12 @@ class NotifyRocketChat(NotifyBase):
|
||||
return False
|
||||
|
||||
# Set our headers for further communication
|
||||
self.headers['X-Auth-Token'] = \
|
||||
response.get('data').get('authToken')
|
||||
self.headers['X-User-Id'] = \
|
||||
response.get('data').get('userId')
|
||||
self.headers['X-Auth-Token'] = response.get(
|
||||
'data', {'authToken': None}).get('authToken')
|
||||
self.headers['X-User-Id'] = response.get(
|
||||
'data', {'userId': None}).get('userId')
|
||||
|
||||
# We're authenticated now
|
||||
self.authenticated = True
|
||||
|
||||
except requests.ConnectionError as e:
|
||||
except requests.RequestException as e:
|
||||
self.logger.warning(
|
||||
'A Connection error occured authenticating to the ' +
|
||||
'Rocket.Chat server.')
|
||||
@ -259,10 +285,6 @@ class NotifyRocketChat(NotifyBase):
|
||||
"""
|
||||
logout of our server
|
||||
"""
|
||||
if not self.authenticated:
|
||||
# Nothing to do
|
||||
return True
|
||||
|
||||
try:
|
||||
r = requests.post(
|
||||
self.api_url + 'logout',
|
||||
@ -278,7 +300,7 @@ class NotifyRocketChat(NotifyBase):
|
||||
RC_HTTP_ERROR_MAP[r.status_code],
|
||||
r.status_code))
|
||||
|
||||
except IndexError:
|
||||
except KeyError:
|
||||
self.logger.warning(
|
||||
'Failed to log off Rocket.Chat server ' +
|
||||
'(error=%s).' % (
|
||||
@ -292,15 +314,13 @@ class NotifyRocketChat(NotifyBase):
|
||||
'Rocket.Chat log off successful; response %s.' % (
|
||||
r.text))
|
||||
|
||||
except requests.ConnectionError as e:
|
||||
except requests.RequestException as e:
|
||||
self.logger.warning(
|
||||
'A Connection error occured logging off the ' +
|
||||
'Rocket.Chat server')
|
||||
self.logger.debug('Socket Exception: %s' % str(e))
|
||||
return False
|
||||
|
||||
# We're no longer authenticated now
|
||||
self.authenticated = False
|
||||
return True
|
||||
|
||||
@staticmethod
|
||||
|
||||
@ -68,18 +68,11 @@ class NotifyTwitter(NotifyBase):
|
||||
'No user was specified.'
|
||||
)
|
||||
|
||||
try:
|
||||
# Attempt to Establish a connection to Twitter
|
||||
self.auth = tweepy.OAuthHandler(ckey, csecret)
|
||||
|
||||
# Apply our Access Tokens
|
||||
self.auth.set_access_token(akey, asecret)
|
||||
|
||||
except Exception:
|
||||
raise TypeError(
|
||||
'Twitter authentication failed; '
|
||||
'please verify your configuration.'
|
||||
)
|
||||
# Store our data
|
||||
self.ckey = ckey
|
||||
self.csecret = csecret
|
||||
self.akey = akey
|
||||
self.asecret = asecret
|
||||
|
||||
return
|
||||
|
||||
@ -88,6 +81,20 @@ class NotifyTwitter(NotifyBase):
|
||||
Perform Twitter Notification
|
||||
"""
|
||||
|
||||
try:
|
||||
# Attempt to Establish a connection to Twitter
|
||||
self.auth = tweepy.OAuthHandler(self.ckey, self.csecret)
|
||||
|
||||
# Apply our Access Tokens
|
||||
self.auth.set_access_token(self.akey, self.asecret)
|
||||
|
||||
except Exception:
|
||||
self.logger.warning(
|
||||
'Twitter authentication failed; '
|
||||
'please verify your configuration.'
|
||||
)
|
||||
return False
|
||||
|
||||
text = '%s\r\n%s' % (title, body)
|
||||
try:
|
||||
# Get our API
|
||||
@ -128,18 +135,19 @@ class NotifyTwitter(NotifyBase):
|
||||
# Now fetch the remaining tokens
|
||||
try:
|
||||
consumer_secret, access_token_key, access_token_secret = \
|
||||
filter(bool, NotifyBase.split_path(results['fullpath']))[0:3]
|
||||
[x for x in filter(bool, NotifyBase.split_path(
|
||||
results['fullpath']))][0:3]
|
||||
|
||||
except (AttributeError, IndexError):
|
||||
except (ValueError, AttributeError, IndexError):
|
||||
# Force some bad values that will get caught
|
||||
# in parsing later
|
||||
consumer_secret = None
|
||||
access_token_key = None
|
||||
access_token_secret = None
|
||||
|
||||
results['ckey'] = consumer_key,
|
||||
results['csecret'] = consumer_secret,
|
||||
results['akey'] = access_token_key,
|
||||
results['asecret'] = access_token_secret,
|
||||
results['ckey'] = consumer_key
|
||||
results['csecret'] = consumer_secret
|
||||
results['akey'] = access_token_key
|
||||
results['asecret'] = access_token_secret
|
||||
|
||||
return results
|
||||
|
||||
@ -26,21 +26,23 @@ from .NotifyFaast import NotifyFaast
|
||||
from .NotifyGrowl.NotifyGrowl import NotifyGrowl
|
||||
from .NotifyGrowl import gntp
|
||||
from .NotifyJSON import NotifyJSON
|
||||
from .NotifyMatterMost import NotifyMatterMost
|
||||
from .NotifyMyAndroid import NotifyMyAndroid
|
||||
from .NotifyProwl import NotifyProwl
|
||||
from .NotifyPushalot import NotifyPushalot
|
||||
from .NotifyPushBullet import NotifyPushBullet
|
||||
from .NotifyPushjet.NotifyPushjet import NotifyPushjet
|
||||
from .NotifyPushjet import pushjet
|
||||
from .NotifyPushover import NotifyPushover
|
||||
from .NotifyRocketChat import NotifyRocketChat
|
||||
from .NotifyTelegram import NotifyTelegram
|
||||
from .NotifyToasty import NotifyToasty
|
||||
from .NotifyTwitter.NotifyTwitter import NotifyTwitter
|
||||
from .NotifyTwitter import tweepy
|
||||
from .NotifyXBMC import NotifyXBMC
|
||||
from .NotifyXML import NotifyXML
|
||||
from .NotifySlack import NotifySlack
|
||||
from .NotifyJoin import NotifyJoin
|
||||
from .NotifyTelegram import NotifyTelegram
|
||||
from .NotifyMatterMost import NotifyMatterMost
|
||||
from .NotifyPushjet.NotifyPushjet import NotifyPushjet
|
||||
|
||||
from ..common import NotifyImageSize
|
||||
from ..common import NOTIFY_IMAGE_SIZES
|
||||
@ -61,6 +63,12 @@ __all__ = [
|
||||
# NotifyEmail Base References (used for Testing)
|
||||
'NotifyEmailBase',
|
||||
|
||||
# gntp (used for Testing)
|
||||
# gntp (used for NotifyGrowl Testing)
|
||||
'gntp',
|
||||
|
||||
# pushjet (used for NotifyPushjet Testing)
|
||||
'pushjet',
|
||||
|
||||
# tweepy (used for NotifyTwitter Testing)
|
||||
'tweepy',
|
||||
]
|
||||
|
||||
157
test/test_pushjet_plugin.py
Normal file
157
test/test_pushjet_plugin.py
Normal file
@ -0,0 +1,157 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
#
|
||||
# NotifyPushjet - Unit Tests
|
||||
#
|
||||
# Copyright (C) 2017 Chris Caron <lead2gold@gmail.com>
|
||||
#
|
||||
# This file is part of apprise.
|
||||
#
|
||||
# This program is free software; you can redistribute it and/or modify it
|
||||
# under the terms of the GNU Lesser General Public License as published by
|
||||
# the Free Software Foundation; either version 3 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU Lesser General Public License for more details.
|
||||
|
||||
from apprise import plugins
|
||||
from apprise import NotifyType
|
||||
from apprise import Apprise
|
||||
import mock
|
||||
|
||||
TEST_URLS = (
|
||||
##################################
|
||||
# NotifyPushjet
|
||||
##################################
|
||||
('pjet://', {
|
||||
'instance': None,
|
||||
}),
|
||||
('pjets://', {
|
||||
'instance': None,
|
||||
}),
|
||||
# Default query (uses pushjet server)
|
||||
('pjet://%s' % ('a' * 32), {
|
||||
'instance': plugins.NotifyPushjet,
|
||||
}),
|
||||
# Specify your own server
|
||||
('pjet://%s@localhost' % ('a' * 32), {
|
||||
'instance': plugins.NotifyPushjet,
|
||||
}),
|
||||
# Specify your own server with port
|
||||
('pjets://%s@localhost:8080' % ('a' * 32), {
|
||||
'instance': plugins.NotifyPushjet,
|
||||
}),
|
||||
('pjet://:@/', {
|
||||
'instance': None,
|
||||
}),
|
||||
('pjet://%s@localhost:8081' % ('a' * 32), {
|
||||
'instance': plugins.NotifyPushjet,
|
||||
# Throws a series of connection and transfer exceptions when this flag
|
||||
# is set and tests that we gracfully handle them
|
||||
'test_notify_exceptions': True,
|
||||
}),
|
||||
)
|
||||
|
||||
|
||||
@mock.patch('apprise.plugins.pushjet.pushjet.Service.send')
|
||||
@mock.patch('apprise.plugins.pushjet.pushjet.Service.refresh')
|
||||
def test_plugin(mock_refresh, mock_send):
|
||||
"""
|
||||
API: NotifyPushjet Plugin() (pt1)
|
||||
|
||||
"""
|
||||
|
||||
# iterate over our dictionary and test it out
|
||||
for (url, meta) in TEST_URLS:
|
||||
|
||||
# Our expected instance
|
||||
instance = meta.get('instance', None)
|
||||
|
||||
# Our expected server objects
|
||||
self = meta.get('self', None)
|
||||
|
||||
# Our expected Query response (True, False, or exception type)
|
||||
response = meta.get('response', True)
|
||||
|
||||
# Allow us to force the server response code to be something other then
|
||||
# the defaults
|
||||
response = meta.get(
|
||||
'response', True if response else False)
|
||||
|
||||
test_notify_exceptions = meta.get(
|
||||
'test_notify_exceptions', False)
|
||||
|
||||
test_exceptions = (
|
||||
plugins.pushjet.errors.AccessError(
|
||||
0, 'pushjet.AccessError() not handled'),
|
||||
plugins.pushjet.errors.NonexistentError(
|
||||
0, 'pushjet.NonexistentError() not handled'),
|
||||
plugins.pushjet.errors.SubscriptionError(
|
||||
0, 'gntp.SubscriptionError() not handled'),
|
||||
plugins.pushjet.errors.RequestError(
|
||||
'pushjet.RequestError() not handled'),
|
||||
)
|
||||
|
||||
try:
|
||||
obj = Apprise.instantiate(url, suppress_exceptions=False)
|
||||
|
||||
if instance is None:
|
||||
# Check that we got what we came for
|
||||
assert obj is instance
|
||||
continue
|
||||
|
||||
assert(isinstance(obj, instance))
|
||||
|
||||
if self:
|
||||
# Iterate over our expected entries inside of our object
|
||||
for key, val in self.items():
|
||||
# Test that our object has the desired key
|
||||
assert(hasattr(key, obj))
|
||||
assert(getattr(key, obj) == val)
|
||||
|
||||
try:
|
||||
if test_notify_exceptions is False:
|
||||
# Store our response
|
||||
mock_send.return_value = response
|
||||
mock_send.side_effect = None
|
||||
|
||||
# check that we're as expected
|
||||
assert obj.notify(
|
||||
title='test', body='body',
|
||||
notify_type=NotifyType.INFO) == response
|
||||
|
||||
else:
|
||||
for exception in test_exceptions:
|
||||
mock_send.side_effect = exception
|
||||
mock_send.return_value = None
|
||||
try:
|
||||
assert obj.notify(
|
||||
title='test', body='body',
|
||||
notify_type=NotifyType.INFO) is False
|
||||
|
||||
except AssertionError:
|
||||
# Don't mess with these entries
|
||||
raise
|
||||
|
||||
except Exception as e:
|
||||
# We can't handle this exception type
|
||||
assert False
|
||||
|
||||
except AssertionError:
|
||||
# Don't mess with these entries
|
||||
raise
|
||||
|
||||
except Exception as e:
|
||||
# Check that we were expecting this exception to happen
|
||||
assert isinstance(e, response)
|
||||
|
||||
except AssertionError:
|
||||
# Don't mess with these entries
|
||||
raise
|
||||
|
||||
except Exception as e:
|
||||
# Handle our exception
|
||||
assert(instance is not None)
|
||||
assert(isinstance(e, instance))
|
||||
@ -20,6 +20,7 @@ from apprise import plugins
|
||||
from apprise import NotifyType
|
||||
from apprise import Apprise
|
||||
from apprise import AppriseAsset
|
||||
from apprise.utils import compat_is_basestring
|
||||
from json import dumps
|
||||
import requests
|
||||
import mock
|
||||
@ -37,9 +38,8 @@ TEST_URLS = (
|
||||
}),
|
||||
# An invalid access and secret key specified
|
||||
('boxcar://access.key/secret.key/', {
|
||||
'instance': plugins.NotifyBoxcar,
|
||||
# Thrown because there were no recipients specified
|
||||
'exception': TypeError,
|
||||
'instance': TypeError,
|
||||
}),
|
||||
# Provide both an access and a secret
|
||||
('boxcar://%s/%s' % ('a' * 64, 'b' * 64), {
|
||||
@ -136,9 +136,8 @@ TEST_URLS = (
|
||||
}),
|
||||
# Invalid APIKey
|
||||
('join://%s' % ('a' * 24), {
|
||||
'instance': None,
|
||||
# Missing a channel
|
||||
'exception': TypeError,
|
||||
'instance': TypeError,
|
||||
}),
|
||||
# APIKey + device
|
||||
('join://%s/%s' % ('a' * 32, 'd' * 32), {
|
||||
@ -336,14 +335,12 @@ TEST_URLS = (
|
||||
'instance': plugins.NotifyMatterMost,
|
||||
}),
|
||||
('mmosts://localhost', {
|
||||
'instance': plugins.NotifyMatterMost,
|
||||
# Thrown because there was no webhook id specified
|
||||
'exception': TypeError,
|
||||
'instance': TypeError,
|
||||
}),
|
||||
('mmost://localhost/bad-web-hook', {
|
||||
'instance': plugins.NotifyMatterMost,
|
||||
# Thrown because the webhook is not in a valid format
|
||||
'exception': TypeError,
|
||||
'instance': TypeError,
|
||||
}),
|
||||
('mmost://:@/', {
|
||||
'instance': None,
|
||||
@ -379,7 +376,7 @@ TEST_URLS = (
|
||||
}),
|
||||
# Invalid APIKey
|
||||
('nma://%s' % ('a' * 24), {
|
||||
'exception': TypeError,
|
||||
'instance': TypeError,
|
||||
}),
|
||||
# APIKey
|
||||
('nma://%s' % ('a' * 48), {
|
||||
@ -401,7 +398,7 @@ TEST_URLS = (
|
||||
}),
|
||||
# APIKey + Invalid DevAPI Key
|
||||
('nma://%s/%s' % ('a' * 48, 'b' * 24), {
|
||||
'exception': TypeError,
|
||||
'instance': TypeError,
|
||||
}),
|
||||
# APIKey + DevAPI Key
|
||||
('nma://%s/%s' % ('a' * 48, 'b' * 48), {
|
||||
@ -462,7 +459,7 @@ TEST_URLS = (
|
||||
}),
|
||||
# Invalid APIKey
|
||||
('prowl://%s' % ('a' * 24), {
|
||||
'exception': TypeError,
|
||||
'instance': TypeError,
|
||||
}),
|
||||
# APIKey
|
||||
('prowl://%s' % ('a' * 40), {
|
||||
@ -484,7 +481,7 @@ TEST_URLS = (
|
||||
}),
|
||||
# APIKey + Invalid Provider Key
|
||||
('prowl://%s/%s' % ('a' * 40, 'b' * 24), {
|
||||
'exception': TypeError,
|
||||
'instance': TypeError,
|
||||
}),
|
||||
# APIKey + No Provider Key (empty)
|
||||
('prowl://%s///' % ('a' * 40), {
|
||||
@ -539,9 +536,8 @@ TEST_URLS = (
|
||||
}),
|
||||
# Invalid AuthToken
|
||||
('palot://%s' % ('a' * 24), {
|
||||
'instance': None,
|
||||
# Missing a channel
|
||||
'exception': TypeError,
|
||||
'instance': TypeError,
|
||||
}),
|
||||
# AuthToken + bad url
|
||||
('palot://:@/', {
|
||||
@ -635,15 +631,15 @@ TEST_URLS = (
|
||||
}),
|
||||
# APIkey; no user
|
||||
('pover://%s' % ('a' * 30), {
|
||||
'exception': TypeError,
|
||||
'instance': TypeError,
|
||||
}),
|
||||
# APIkey; invalid user
|
||||
('pover://%s@%s' % ('u' * 20, 'a' * 30), {
|
||||
'exception': TypeError,
|
||||
'instance': TypeError,
|
||||
}),
|
||||
# Invalid APIKey; valid User
|
||||
('pover://%s@%s' % ('u' * 30, 'a' * 24), {
|
||||
'exception': TypeError,
|
||||
'instance': TypeError,
|
||||
}),
|
||||
# APIKey + Valid User
|
||||
('pover://%s@%s' % ('u' * 30, 'a' * 30), {
|
||||
@ -706,6 +702,122 @@ TEST_URLS = (
|
||||
'test_requests_exceptions': True,
|
||||
}),
|
||||
|
||||
##################################
|
||||
# NotifyRocketChat
|
||||
##################################
|
||||
('rocket://', {
|
||||
'instance': None,
|
||||
}),
|
||||
('rockets://', {
|
||||
'instance': None,
|
||||
}),
|
||||
# No username or pass
|
||||
('rocket://localhost', {
|
||||
'instance': TypeError,
|
||||
}),
|
||||
# No room or channel
|
||||
('rocket://user:pass@localhost', {
|
||||
'instance': TypeError,
|
||||
}),
|
||||
# No valid rooms or channels
|
||||
('rocket://user:pass@localhost/#/!/@', {
|
||||
'instance': TypeError,
|
||||
}),
|
||||
# A room and port identifier
|
||||
('rocket://user:pass@localhost:8080/room/', {
|
||||
'instance': plugins.NotifyRocketChat,
|
||||
# The response text is expected to be the following on a success
|
||||
'requests_response_text': {
|
||||
'status': 'success',
|
||||
'data': {
|
||||
'authToken': 'abcd',
|
||||
'userId': 'user',
|
||||
},
|
||||
},
|
||||
}),
|
||||
# A channel
|
||||
('rockets://user:pass@localhost/#channel', {
|
||||
'instance': plugins.NotifyRocketChat,
|
||||
# The response text is expected to be the following on a success
|
||||
'requests_response_text': {
|
||||
'status': 'success',
|
||||
'data': {
|
||||
'authToken': 'abcd',
|
||||
'userId': 'user',
|
||||
},
|
||||
},
|
||||
}),
|
||||
# Several channels
|
||||
('rocket://user:pass@localhost/#channel1/#channel2/', {
|
||||
'instance': plugins.NotifyRocketChat,
|
||||
# The response text is expected to be the following on a success
|
||||
'requests_response_text': {
|
||||
'status': 'success',
|
||||
'data': {
|
||||
'authToken': 'abcd',
|
||||
'userId': 'user',
|
||||
},
|
||||
},
|
||||
}),
|
||||
# Several Rooms
|
||||
('rocket://user:pass@localhost/room1/room2', {
|
||||
'instance': plugins.NotifyRocketChat,
|
||||
# The response text is expected to be the following on a success
|
||||
'requests_response_text': {
|
||||
'status': 'success',
|
||||
'data': {
|
||||
'authToken': 'abcd',
|
||||
'userId': 'user',
|
||||
},
|
||||
},
|
||||
}),
|
||||
# A room and channel
|
||||
('rocket://user:pass@localhost/room/#channel', {
|
||||
'instance': plugins.NotifyRocketChat,
|
||||
# The response text is expected to be the following on a success
|
||||
'requests_response_text': {
|
||||
'status': 'success',
|
||||
'data': {
|
||||
'authToken': 'abcd',
|
||||
'userId': 'user',
|
||||
},
|
||||
},
|
||||
}),
|
||||
('rocket://:@/', {
|
||||
'instance': None,
|
||||
}),
|
||||
# A room and channel
|
||||
('rockets://user:pass@localhost/rooma/#channela', {
|
||||
# The response text is expected to be the following on a success
|
||||
'requests_response_code': requests.codes.ok,
|
||||
'requests_response_text': {
|
||||
# return something other then a success message type
|
||||
'status': 'failure',
|
||||
},
|
||||
# Exception is thrown in this case
|
||||
'instance': plugins.NotifyRocketChat,
|
||||
# Notifications will fail in this event
|
||||
'response': False,
|
||||
}),
|
||||
('rocket://user:pass@localhost:8081/room1/room2', {
|
||||
'instance': plugins.NotifyRocketChat,
|
||||
# force a failure
|
||||
'response': False,
|
||||
'requests_response_code': requests.codes.internal_server_error,
|
||||
}),
|
||||
('rocket://user:pass@localhost:8082/#channel', {
|
||||
'instance': plugins.NotifyRocketChat,
|
||||
# throw a bizzare code forcing us to fail to look it up
|
||||
'response': False,
|
||||
'requests_response_code': 999,
|
||||
}),
|
||||
('rocket://user:pass@localhost:8083/#chan1/#chan2/room', {
|
||||
'instance': plugins.NotifyRocketChat,
|
||||
# Throws a series of connection and transfer exceptions when this flag
|
||||
# is set and tests that we gracfully handle them
|
||||
'test_requests_exceptions': True,
|
||||
}),
|
||||
|
||||
##################################
|
||||
# NotifySlack
|
||||
##################################
|
||||
@ -741,19 +853,19 @@ TEST_URLS = (
|
||||
}),
|
||||
('slack://username@T1JJ3T3L2/A1BRTD4JD/TIiajkdnlazkcOXrIdevi7FQ', {
|
||||
# Missing a channel
|
||||
'exception': TypeError,
|
||||
'instance': TypeError,
|
||||
}),
|
||||
('slack://username@INVALID/A1BRTD4JD/TIiajkdnlazkcOXrIdevi7FQ/#cool', {
|
||||
# invalid 1st Token
|
||||
'exception': TypeError,
|
||||
'instance': TypeError,
|
||||
}),
|
||||
('slack://username@T1JJ3T3L2/INVALID/TIiajkdnlazkcOXrIdevi7FQ/#great', {
|
||||
# invalid 2rd Token
|
||||
'exception': TypeError,
|
||||
'instance': TypeError,
|
||||
}),
|
||||
('slack://username@T1JJ3T3L2/A1BRTD4JD/INVALID/#channel', {
|
||||
# invalid 3rd Token
|
||||
'exception': TypeError,
|
||||
'instance': TypeError,
|
||||
}),
|
||||
('slack://l2g@T1JJ3T3L2/A1BRTD4JD/TIiajkdnlazkcOXrIdevi7FQ/#usenet', {
|
||||
'instance': plugins.NotifySlack,
|
||||
@ -906,7 +1018,7 @@ TEST_URLS = (
|
||||
}),
|
||||
# No username specified but contains a device
|
||||
('toasty://%s' % ('d' * 32), {
|
||||
'exception': TypeError,
|
||||
'instance': TypeError,
|
||||
}),
|
||||
# User + 1 device
|
||||
('toasty://user@device', {
|
||||
@ -1067,9 +1179,6 @@ def test_rest_plugins(mock_post, mock_get):
|
||||
# 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)
|
||||
|
||||
@ -1083,6 +1192,13 @@ def test_rest_plugins(mock_post, mock_get):
|
||||
requests.codes.ok if response else requests.codes.not_found,
|
||||
)
|
||||
|
||||
# Allow us to force the server response text to be something other then
|
||||
# the defaults
|
||||
requests_response_text = meta.get('requests_response_text')
|
||||
if not compat_is_basestring(requests_response_text):
|
||||
# Convert to string
|
||||
requests_response_text = dumps(requests_response_text)
|
||||
|
||||
# Allow notification type override, otherwise default to INFO
|
||||
notify_type = meta.get('notify_type', NotifyType.INFO)
|
||||
|
||||
@ -1112,6 +1228,12 @@ def test_rest_plugins(mock_post, mock_get):
|
||||
# Handle our default response
|
||||
mock_post.return_value.status_code = requests_response_code
|
||||
mock_get.return_value.status_code = requests_response_code
|
||||
|
||||
# Handle our default text response
|
||||
mock_get.return_value.text = requests_response_text
|
||||
mock_post.return_value.text = requests_response_text
|
||||
|
||||
# Ensure there is no side effect set
|
||||
mock_post.side_effect = None
|
||||
mock_get.side_effect = None
|
||||
|
||||
@ -1135,20 +1257,11 @@ def test_rest_plugins(mock_post, mock_get):
|
||||
obj = Apprise.instantiate(
|
||||
url, asset=asset, suppress_exceptions=False)
|
||||
|
||||
# Make sure we weren't expecting an exception and just didn't get
|
||||
# one.
|
||||
assert exception is None
|
||||
|
||||
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' % (url, str(obj)))
|
||||
assert(False)
|
||||
|
||||
assert(isinstance(obj, instance))
|
||||
|
||||
# Disable throttling to speed up unit tests
|
||||
@ -1172,6 +1285,7 @@ def test_rest_plugins(mock_post, mock_get):
|
||||
for _exception in test_requests_exceptions:
|
||||
mock_post.side_effect = _exception
|
||||
mock_get.side_effect = _exception
|
||||
|
||||
try:
|
||||
assert obj.notify(
|
||||
title='test', body='body',
|
||||
@ -1203,8 +1317,8 @@ def test_rest_plugins(mock_post, mock_get):
|
||||
except Exception as e:
|
||||
# Handle our exception
|
||||
print('%s / %s' % (url, str(e)))
|
||||
assert(exception is not None)
|
||||
assert(isinstance(e, exception))
|
||||
assert(instance is not None)
|
||||
assert(isinstance(e, instance))
|
||||
|
||||
|
||||
@mock.patch('requests.get')
|
||||
@ -1466,6 +1580,133 @@ def test_notify_pushover_plugin(mock_post, mock_get):
|
||||
assert(plugins.NotifyPushover.parse_url(42) is None)
|
||||
|
||||
|
||||
@mock.patch('requests.get')
|
||||
@mock.patch('requests.post')
|
||||
def test_notify_rocketchat_plugin(mock_post, mock_get):
|
||||
"""
|
||||
API: NotifyRocketChat() Extra Checks
|
||||
|
||||
"""
|
||||
# Chat ID
|
||||
recipients = 'l2g, lead2gold, #channel, #channel2'
|
||||
|
||||
# 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.NotifyRocketChat(recipients=None)
|
||||
# invalid recipients list (None)
|
||||
assert(False)
|
||||
|
||||
except TypeError:
|
||||
# Exception should be thrown about the fact no recipients were
|
||||
# specified
|
||||
assert(True)
|
||||
|
||||
try:
|
||||
obj = plugins.NotifyRocketChat(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.NotifyRocketChat(recipients=set())
|
||||
# invalid recipient list/set (no entries)
|
||||
assert(False)
|
||||
|
||||
except TypeError:
|
||||
# Exception should be thrown about the fact no recipients were
|
||||
# specified
|
||||
assert(True)
|
||||
|
||||
obj = plugins.NotifyRocketChat(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
|
||||
#
|
||||
assert obj.logout() is True
|
||||
|
||||
# Support the handling of an empty and invalid URL strings
|
||||
assert plugins.NotifyRocketChat.parse_url(None) is None
|
||||
assert plugins.NotifyRocketChat.parse_url('') is None
|
||||
assert plugins.NotifyRocketChat.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 = ''
|
||||
|
||||
#
|
||||
# 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
|
||||
|
||||
#
|
||||
# Logout
|
||||
#
|
||||
assert obj.logout() is False
|
||||
|
||||
# KeyError handling
|
||||
mock_post.return_value.status_code = 999
|
||||
mock_get.return_value.status_code = 999
|
||||
|
||||
#
|
||||
# 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
|
||||
|
||||
#
|
||||
# Logout
|
||||
#
|
||||
assert obj.logout() is False
|
||||
|
||||
mock_post.return_value.text = ''
|
||||
# Generate exceptions
|
||||
mock_get.side_effect = requests.ConnectionError(
|
||||
0, 'requests.ConnectionError() not handled')
|
||||
mock_post.side_effect = mock_get.side_effect
|
||||
mock_get.return_value.text = ''
|
||||
mock_post.return_value.text = ''
|
||||
|
||||
#
|
||||
# Send Notification
|
||||
#
|
||||
assert obj.send_notification(
|
||||
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
|
||||
#
|
||||
# Logout
|
||||
#
|
||||
assert obj.logout() is False
|
||||
|
||||
|
||||
@mock.patch('requests.get')
|
||||
@mock.patch('requests.post')
|
||||
def test_notify_toasty_plugin(mock_post, mock_get):
|
||||
|
||||
189
test/test_twitter_plugin.py
Normal file
189
test/test_twitter_plugin.py
Normal file
@ -0,0 +1,189 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
#
|
||||
# NotifyTwitter - Unit Tests
|
||||
#
|
||||
# Copyright (C) 2017 Chris Caron <lead2gold@gmail.com>
|
||||
#
|
||||
# This file is part of apprise.
|
||||
#
|
||||
# This program is free software; you can redistribute it and/or modify it
|
||||
# under the terms of the GNU Lesser General Public License as published by
|
||||
# the Free Software Foundation; either version 3 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU Lesser General Public License for more details.
|
||||
|
||||
from apprise import plugins
|
||||
from apprise import NotifyType
|
||||
from apprise import Apprise
|
||||
import mock
|
||||
|
||||
|
||||
TEST_URLS = (
|
||||
##################################
|
||||
# NotifyPushjet
|
||||
##################################
|
||||
('tweet://', {
|
||||
'instance': None,
|
||||
}),
|
||||
('tweet://consumer_key', {
|
||||
# Missing Keys
|
||||
'instance': TypeError,
|
||||
}),
|
||||
('tweet://consumer_key/consumer_key/', {
|
||||
# Missing Keys
|
||||
'instance': TypeError,
|
||||
}),
|
||||
('tweet://consumer_key/consumer_key/access_token/', {
|
||||
# Missing Access Secret
|
||||
'instance': TypeError,
|
||||
}),
|
||||
('tweet://consumer_key/consumer_key/access_token/access_secret', {
|
||||
# Missing User
|
||||
'instance': TypeError,
|
||||
}),
|
||||
('tweet://user@consumer_key/consumer_key/access_token/access_secret', {
|
||||
# We're good!
|
||||
'instance': plugins.NotifyTwitter,
|
||||
}),
|
||||
('tweet://:@/', {
|
||||
'instance': None,
|
||||
}),
|
||||
)
|
||||
|
||||
|
||||
@mock.patch('apprise.plugins.tweepy.API')
|
||||
@mock.patch('apprise.plugins.tweepy.OAuthHandler')
|
||||
def test_plugin(mock_oauth, mock_api):
|
||||
"""
|
||||
API: NotifyTwitter Plugin() (pt1)
|
||||
|
||||
"""
|
||||
|
||||
# iterate over our dictionary and test it out
|
||||
for (url, meta) in TEST_URLS:
|
||||
|
||||
# Our expected instance
|
||||
instance = meta.get('instance', None)
|
||||
|
||||
# Our expected server objects
|
||||
self = meta.get('self', None)
|
||||
|
||||
# Our expected Query response (True, False, or exception type)
|
||||
response = meta.get('response', True)
|
||||
|
||||
# Allow us to force the server response code to be something other then
|
||||
# the defaults
|
||||
response = meta.get(
|
||||
'response', True if response else False)
|
||||
|
||||
try:
|
||||
obj = Apprise.instantiate(url, suppress_exceptions=False)
|
||||
|
||||
if instance is None:
|
||||
# Check that we got what we came for
|
||||
assert obj is instance
|
||||
continue
|
||||
|
||||
assert(isinstance(obj, instance))
|
||||
|
||||
if self:
|
||||
# Iterate over our expected entries inside of our object
|
||||
for key, val in self.items():
|
||||
# Test that our object has the desired key
|
||||
assert(hasattr(key, obj))
|
||||
assert(getattr(key, obj) == val)
|
||||
|
||||
# check that we're as expected
|
||||
assert obj.notify(
|
||||
title='test', body='body',
|
||||
notify_type=NotifyType.INFO) == response
|
||||
|
||||
except AssertionError:
|
||||
# Don't mess with these entries
|
||||
raise
|
||||
|
||||
except Exception as e:
|
||||
# Handle our exception
|
||||
assert(instance is not None)
|
||||
assert(isinstance(e, instance))
|
||||
|
||||
|
||||
@mock.patch('apprise.plugins.tweepy.API.send_direct_message')
|
||||
@mock.patch('apprise.plugins.tweepy.OAuthHandler.set_access_token')
|
||||
def test_twitter_plugin_init(set_access_token, send_direct_message):
|
||||
"""
|
||||
API: NotifyTwitter Plugin() (pt2)
|
||||
|
||||
"""
|
||||
|
||||
try:
|
||||
plugins.NotifyTwitter(
|
||||
ckey=None, csecret=None, akey=None, asecret=None)
|
||||
assert False
|
||||
except TypeError:
|
||||
# All keys set to none
|
||||
assert True
|
||||
|
||||
try:
|
||||
plugins.NotifyTwitter(
|
||||
ckey='value', csecret=None, akey=None, asecret=None)
|
||||
assert False
|
||||
except TypeError:
|
||||
# csecret not set
|
||||
assert True
|
||||
|
||||
try:
|
||||
plugins.NotifyTwitter(
|
||||
ckey='value', csecret='value', akey=None, asecret=None)
|
||||
assert False
|
||||
except TypeError:
|
||||
# akey not set
|
||||
assert True
|
||||
|
||||
try:
|
||||
plugins.NotifyTwitter(
|
||||
ckey='value', csecret='value', akey='value', asecret=None)
|
||||
assert False
|
||||
except TypeError:
|
||||
# asecret not set
|
||||
assert True
|
||||
|
||||
try:
|
||||
plugins.NotifyTwitter(
|
||||
ckey='value', csecret='value', akey='value', asecret='value')
|
||||
assert False
|
||||
except TypeError:
|
||||
# user not set
|
||||
assert True
|
||||
|
||||
try:
|
||||
obj = plugins.NotifyTwitter(
|
||||
ckey='value', csecret='value', akey='value', asecret='value',
|
||||
user='l2g')
|
||||
# We should initialize properly
|
||||
assert True
|
||||
|
||||
except TypeError:
|
||||
# We should not reach here
|
||||
assert False
|
||||
|
||||
set_access_token.side_effect = TypeError('Invalid')
|
||||
|
||||
assert obj.notify(
|
||||
title='test', body='body',
|
||||
notify_type=NotifyType.INFO) is False
|
||||
|
||||
# Make it so we can pass authentication, but fail on message
|
||||
# delivery
|
||||
set_access_token.side_effect = None
|
||||
set_access_token.return_value = True
|
||||
send_direct_message.side_effect = plugins.tweepy.error.TweepError(
|
||||
0, 'pushjet.TweepyError() not handled'),
|
||||
|
||||
assert obj.notify(
|
||||
title='test', body='body',
|
||||
notify_type=NotifyType.INFO) is False
|
||||
Loading…
Reference in New Issue
Block a user