github
/
apprise
mirror of https://github.com/caronc/apprise
1
0
Fork 0
Code Issues Releases Wiki Activity

test coverage added

pull/782/head
Chris Caron 2022-12-18 16:30:05 -05:00
parent f2bee74ff1
commit d1ebef506a
6 changed files with 202 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
KEYWORDS
View File

@ -12,6 +12,7 @@ Dingtalk
Discord
Email
Emby
Exotel
Faast
FCM
Flock

1
README.md
View File

@ -131,6 +131,7 @@ The table below identifies the services this tool supports and some example serv
| [ClickSend](https://github.com/caronc/apprise/wiki/Notify_clicksend) | clicksend:// | (TCP) 443 | clicksend://user:pass@PhoneNo<br/>clicksend://user:pass@ToPhoneNo1/ToPhoneNo2/ToPhoneNoN
| [DAPNET](https://github.com/caronc/apprise/wiki/Notify_dapnet) | dapnet:// | (TCP) 80 | dapnet://user:pass@callsign<br/>dapnet://user:pass@callsign1/callsign2/callsignN
| [D7 Networks](https://github.com/caronc/apprise/wiki/Notify_d7networks) | d7sms:// | (TCP) 443 | d7sms://user:pass@PhoneNo<br/>d7sms://user:pass@ToPhoneNo1/ToPhoneNo2/ToPhoneNoN
| [Exotel](https://github.com/caronc/apprise/wiki/Notify_exotel) | exotel:// | (TCP) 443 | exotel://sid:token@FromPhoneNo<br/>exotel://sid:token@FromPhoneNo/ToPhoneNo<br/>exotel://sid:token@FromPhoneNo/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN
| [DingTalk](https://github.com/caronc/apprise/wiki/Notify_dingtalk) | dingtalk:// | (TCP) 443 | dingtalk://token/<br />dingtalk://token/ToPhoneNo<br />dingtalk://token/ToPhoneNo1/ToPhoneNo2/ToPhoneNo1/
| [Kavenegar](https://github.com/caronc/apprise/wiki/Notify_kavenegar) | kavenegar:// | (TCP) 443 | kavenegar://ApiKey/ToPhoneNo<br/>kavenegar://FromPhoneNo@ApiKey/ToPhoneNo<br/>kavenegar://ApiKey/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN
| [MessageBird](https://github.com/caronc/apprise/wiki/Notify_messagebird) | msgbird:// | (TCP) 443 | msgbird://ApiKey/FromPhoneNo<br/>msgbird://ApiKey/FromPhoneNo/ToPhoneNo<br/>msgbird://ApiKey/FromPhoneNo/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN/

58
apprise/plugins/NotifyExotel.py
View File

@ -25,6 +25,8 @@
import requests
from itertools import chain
from .NotifyBase import NotifyBase
from ..URLBase import PrivacyMode
from ..common import NotifyType
@ -193,7 +195,7 @@ class NotifyExotel(NotifyBase):
"""
super().__init__(**kwargs)
# API SID (associated with account)
# Account SID
self.sid = validate_regex(sid)
if not self.sid:
msg = 'An invalid Exotel SID ' \
@ -201,15 +203,17 @@ class NotifyExotel(NotifyBase):
self.logger.warning(msg)
raise TypeError(msg)
# API Key (associated with project)
self.token = validate_regex(
token, *self.template_tokens['token']['regex'])
# API Token (associated with account)
self.token = validate_regex(token)
if not self.token:
msg = 'An invalid Exotel API Token ' \
'({}) was specified.'.format(token)
self.logger.warning(msg)
raise TypeError(msg)
# Used for URL generation afterwards only
self.invalid_targets = list()
# Store our region
try:
self.region_name = self.template_args['region']['default'] \
@ -232,27 +236,21 @@ class NotifyExotel(NotifyBase):
#
# Priority
#
try:
# Acquire our priority if we can:
# - We accept both the integer form as well as a string
# representation
self.priority = int(priority)
except TypeError:
# NoneType means use Default; this is an okay exception
if priority is None:
# Default
self.priority = self.template_args['priority']['default']
except ValueError:
else:
# Input is a string; attempt to get the lookup from our
# priority mapping
priority = priority.lower().strip()
self.priority = priority.lower().strip()
# This little bit of black magic allows us to match against
# low, lo, l (for low);
# normal, norma, norm, nor, no, n (for normal)
# ... etc
result = next((key for key in EXOTEL_PRIORITY_MAP.keys()
if key.startswith(priority)), None) \
if key.startswith(self.priority)), None) \
if priority else None
# Now test to see if we got a match
@ -265,17 +263,10 @@ class NotifyExotel(NotifyBase):
# store our successfully looked up priority
self.priority = EXOTEL_PRIORITY_MAP[result]
if self.priority is not None and \
self.priority not in EXOTEL_PRIORITY_MAP.values():
msg = 'An invalid Exotel priority ' \
'({}) was specified.'.format(priority)
self.logger.warning(msg)
raise TypeError(msg)
# The Source Phone #
self.source = source
result = is_phone_no(source)
result = is_phone_no(source, min_len=9)
if not result:
msg = 'The Account (From) Phone # specified ' \
'({}) is invalid.'.format(source)
@ -290,17 +281,22 @@ class NotifyExotel(NotifyBase):
for target in parse_phone_no(targets):
# Validate targets and drop bad ones:
result = is_phone_no(target)
result = is_phone_no(target, min_len=9)
if not result:
self.logger.warning(
'Dropped invalid phone # '
'({}) specified.'.format(target),
)
self.invalid_targets.append(target)
continue
# store valid phone number
self.targets.append(result['full'])
if len(self.targets) == 0 and not self.invalid_targets:
# No sources specified, use our own phone no
self.targets.append(self.source)
return
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
@ -308,6 +304,11 @@ class NotifyExotel(NotifyBase):
Perform Exotel Notification
"""
if not self.targets:
# There were no endpoints to notify
self.logger.warning('There were no Exotel targets to notify.')
return False
# error tracking (used for function return)
has_error = False
@ -339,10 +340,6 @@ class NotifyExotel(NotifyBase):
# Prepare our notify_url
notify_url = EXOTEL_API_LOOKUP[self.region_name].format(sid=self.sid)
if len(targets) == 0:
# No sources specified, use our own phone no
targets.append(self.source)
while len(targets):
# Get our target to notify
target = targets.pop(0)
@ -422,13 +419,14 @@ class NotifyExotel(NotifyBase):
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
return '{schema}://{sid}:{token}@{source}/{targets}/?{params}'.format(
schema=self.secure_protocol[0],
schema=self.secure_protocol,
sid=self.pprint(
self.sid, privacy, mode=PrivacyMode.Secret, safe=''),
token=self.pprint(self.token, privacy, safe=''),
source=NotifyExotel.quote(self.source, safe=''),
targets='/'.join(
[NotifyExotel.quote(x, safe='') for x in self.targets]),
[NotifyExotel.quote(x, safe='') for x in chain(
self.targets, self.invalid_targets)]),
params=NotifyExotel.urlencode(params))
@staticmethod

2
packaging/redhat/python-apprise.spec
View File

@ -36,7 +36,7 @@ notification services that are out there. Apprise opens the door and makes
it easy to access:
Apprise API, AWS SES, AWS SNS, Bark, BulkSMS, Boxcar, ClickSend, DAPNET,
DingTalk, Discord, E-Mail, Emby, Faast, FCM, Flock, Gitter, Google Chat,
DingTalk, Discord, E-Mail, Emby, Exotel, Faast, FCM, Flock, Gitter, Google Chat,
Gotify, Growl, Guilded, Home Assistant, IFTTT, Join, Kavenegar, KODI, Kumulos,
LaMetric, Line, MacOSX, Mailgun, Mattermost, Matrix, Microsoft Windows,
Mastodon, Microsoft Teams, MessageBird, MQTT, MSG91, MyAndroid, Nexmo,

22
test/helpers/rest.py
View File

@ -183,8 +183,8 @@ class AppriseURLTester:
assert False
if not isinstance(obj, instance):
print('%s instantiated %s (but expected %s)' % (
url, type(instance), str(obj)))
print('%s instantiated as %s (but expected %s)' % (
url, type(obj), str(instance)))
assert False
if isinstance(obj, NotifyBase):
@ -228,8 +228,8 @@ class AppriseURLTester:
# way these tests work. Just printing before
# throwing our assertion failure makes things
# easier to debug later on
print('TEST FAIL: {} regenerated as {}'.format(
url, obj.url()))
print('TEST FAIL: {} became {} and then regenerated as {}'
.format(url, obj.url(), type(obj_cmp)))
assert False
# Tidy our object
@ -358,9 +358,19 @@ class AppriseURLTester:
if test_requests_exceptions is False:
# check that we're as expected
assert obj.notify(
response = obj.notify(
body=self.body, title=self.title,
notify_type=notify_type) == notify_response
notify_type=notify_type)
if response != notify_response:
# We did not get the notify() response we thought
print(
'TEST FAIL: {} notify_response from {}.send() was '
'{} expected {}'
.format(
url, obj.__class__.__name__, response,
notify_response))
assert False
# check that this doesn't change using different overflow
# methods

155
test/test_plugin_exotel.py Normal file
View File

@ -0,0 +1,155 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2022 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.
from apprise.plugins.NotifyExotel import NotifyExotel
from helpers import AppriseURLTester
# Disable logging for a cleaner testing output
import logging
logging.disable(logging.CRITICAL)
# Our Testing URLs
apprise_url_tests = (
('exotel://', {
# No Account SID specified
'instance': TypeError,
}),
('exotel://:@/', {
# invalid Auth token
'instance': TypeError,
}),
('exotel://{}@12345678'.format('a' * 32), {
# Just sid provided
'instance': TypeError,
}),
('exotel://{}:{}@_'.format('a' * 32, 'b' * 32), {
# sid and token provided but invalid from
'instance': TypeError,
}),
('exotel://{}:{}@{}'.format('a' * 32, 'b' * 32, '3' * 8), {
# sid and token provided and from but invalid from no
'instance': TypeError,
}),
('exotel://{}:{}@{}'.format('a' * 32, 'b' * 32, '3' * 9), {
# sid and token provided and from
'instance': NotifyExotel,
}),
('exotel://{}:{}@{}/123/{}/abcd/'.format(
'a' * 32, 'b' * 32, '3' * 11, '9' * 15), {
# valid everything but target numbers
'instance': NotifyExotel,
# Since the targets are invalid, we'll fail to send()
'notify_response': False,
}),
('exotel://{}:{}@12345/{}'.format('a' * 32, 'b' * 32, '4' * 11), {
# using short-code (5 characters) is not supported
'instance': TypeError,
# Our expected url(privacy=True) startswith() response:
'privacy_url': 'exotel://...aaaa:b...b@12345',
}),
('exotel://{}:{}@{}'.format('a' * 32, 'b' * 32, '5' * 11), {
# using phone no with no target - we text ourselves in
# this case
'instance': NotifyExotel,
}),
('exotel://_?sid={}&token={}&from={}'.format(
'a' * 32, 'b' * 32, '5' * 11), {
# use get args to acomplish the same thing
'instance': NotifyExotel,
}),
('exotel://_?sid={}&token={}&from={}&unicode=Yes'.format(
'a' * 32, 'b' * 32, '5' * 11), {
# Test unicode flag
'instance': NotifyExotel,
}),
('exotel://_?sid={}&token={}&from={}&unicode=no'.format(
'a' * 32, 'b' * 32, '5' * 11), {
# Test unicode flag
'instance': NotifyExotel,
}),
('exotel://_?sid={}&token={}&from={}&region=us'.format(
'a' * 32, 'b' * 32, '5' * 11), {
# Test region flag (Us)
'instance': NotifyExotel,
}),
('exotel://_?sid={}&token={}&from={}&region=in'.format(
'a' * 32, 'b' * 32, '5' * 11), {
# Test region flag (India)
'instance': NotifyExotel,
}),
('exotel://_?sid={}&token={}&from={}&region=invalid'.format(
'a' * 32, 'b' * 32, '5' * 11), {
# Test region flag Invalid
'instance': TypeError,
}),
('exotel://_?sid={}&token={}&from={}&priority=normal'.format(
'a' * 32, 'b' * 32, '5' * 11), {
# Test region flag (Us)
'instance': NotifyExotel,
}),
('exotel://_?sid={}&token={}&from={}&priority=high'.format(
'a' * 32, 'b' * 32, '5' * 11), {
# Test region flag (India)
'instance': NotifyExotel,
}),
('exotel://_?sid={}&token={}&from={}&priority=invalid'.format(
'a' * 32, 'b' * 32, '5' * 11), {
# Test region flag Invalid
'instance': TypeError,
}),
('exotel://_?sid={}&token={}&source={}'.format(
'a' * 32, 'b' * 32, '5' * 11), {
# use get args to acomplish the same thing (use source instead of from)
'instance': NotifyExotel,
}),
('exotel://_?sid={}&token={}&from={}&to={}'.format(
'a' * 32, 'b' * 32, '5' * 11, '7' * 13), {
# use to=
'instance': NotifyExotel,
}),
('exotel://{}:{}@{}'.format('a' * 32, 'b' * 32, '6' * 11), {
'instance': NotifyExotel,
# throw a bizzare code forcing us to fail to look it up
'response': False,
'requests_response_code': 999,
}),
('exotel://{}:{}@{}'.format('a' * 32, 'b' * 32, '6' * 11), {
'instance': NotifyExotel,
# Throws a series of connection and transfer exceptions when this flag
# is set and tests that we gracfully handle them
'test_requests_exceptions': True,
}),
)
def test_plugin_exotel_urls():
"""
NotifyExotel() Apprise URLs
"""
# Run our general tests
AppriseURLTester(tests=apprise_url_tests).run_all()