From f2bee74ff143bcce2ee4b709efb78c3f89d668a0 Mon Sep 17 00:00:00 2001 From: Chris Caron Date: Sat, 10 Dec 2022 12:01:25 -0500 Subject: [PATCH 1/2] initial commit --- apprise/plugins/NotifyExotel.py | 495 ++++++++++++++++++++++++++++++++ 1 file changed, 495 insertions(+) create mode 100644 apprise/plugins/NotifyExotel.py diff --git a/apprise/plugins/NotifyExotel.py b/apprise/plugins/NotifyExotel.py new file mode 100644 index 0000000000..cd66acf161 --- /dev/null +++ b/apprise/plugins/NotifyExotel.py @@ -0,0 +1,495 @@ +# -*- coding: utf-8 -*- +# +# Copyright (C) 2022 Chris Caron +# 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 requests + +from .NotifyBase import NotifyBase +from ..URLBase import PrivacyMode +from ..common import NotifyType +from ..utils import is_phone_no +from ..utils import parse_phone_no +from ..utils import validate_regex +from ..AppriseLocale import gettext_lazy as _ + + +class ExotelPriority: + """ + Priorities + """ + NORMAL = 'normal' + HIGH = 'high' + + +EXOTEL_PRIORITIES = ( + ExotelPriority.NORMAL, + ExotelPriority.HIGH, +) + +EXOTEL_PRIORITY_MAP = { + # short for 'normal' + 'normal': ExotelPriority.NORMAL, + # short for 'high' + '+': ExotelPriority.HIGH, + 'high': ExotelPriority.HIGH, +} + + +class ExotelEncoding(object): + """ + The different encodings supported + """ + TEXT = "plain" + UNICODE = "unicode" + + +class ExotelRegion: + """ + Regions + """ + US = 'us' + + # India + IN = 'in' + + +# Exotel APIs +EXOTEL_API_LOOKUP = { + ExotelRegion.US: 'https://api.exotel.com/v1/Accounts/{sid}/Sms/send', + ExotelRegion.IN: 'https://api.in.exotel.com/v1/Accounts/{sid}/Sms/send', +} + +# A List of our regions we can use for verification +EXOTEL_REGIONS = ( + ExotelRegion.US, + ExotelRegion.IN, +) + + +class NotifyExotel(NotifyBase): + """ + A wrapper for Exotel Notifications + """ + + # The default descriptive name associated with the Notification + service_name = 'Exotel' + + # The services URL + service_url = 'https://exotel.com' + + # The default protocol (nexmo kept for backwards compatibility) + secure_protocol = 'exotel' + + # A URL that takes you to the setup/help of the specific protocol + setup_url = 'https://github.com/caronc/apprise/wiki/Notify_exotel' + + # The maximum length of the body + body_maxlen = 2000 + + # A title can not be used for SMS Messages. Setting this to zero will + # cause any title (if defined) to get placed into the message body. + title_maxlen = 0 + + # Define object templates + templates = ( + '{schema}://{sid}:{token}@{from_phone}', + '{schema}://{sid}:{token}@{from_phone}/{targets}', + ) + + # Define our template tokens + template_tokens = dict(NotifyBase.template_tokens, **{ + 'sid': { + 'name': _('Secure ID'), + 'type': 'string', + 'required': True, + 'private': True, + }, + 'token': { + 'name': _('Token'), + 'type': 'string', + 'private': True, + 'required': True, + }, + 'from_phone': { + 'name': _('From Phone No'), + 'type': 'string', + 'required': True, + 'regex': (r'^\+?[0-9\s)(+-]+$', 'i'), + 'map_to': 'source', + }, + 'target_phone': { + 'name': _('Target Phone No'), + 'type': 'string', + 'prefix': '+', + 'regex': (r'^[0-9\s)(+-]+$', 'i'), + 'map_to': 'targets', + }, + 'targets': { + 'name': _('Targets'), + 'type': 'list:string', + }, + }) + + # Define our template arguments + template_args = dict(NotifyBase.template_args, **{ + 'to': { + 'alias_of': 'targets', + }, + 'from': { + 'alias_of': 'from_phone', + }, + 'sid': { + 'alias_of': 'sid', + }, + 'token': { + 'alias_of': 'token', + }, + 'unicode': { + # Unicode characters + 'name': _('Unicode Characters'), + 'type': 'bool', + 'default': True, + }, + 'region': { + 'name': _('Region Name'), + 'type': 'choice:string', + 'values': EXOTEL_REGIONS, + 'default': ExotelRegion.US, + 'map_to': 'region_name', + }, + 'priority': { + 'name': _('Priority'), + 'type': 'choice:int', + 'values': EXOTEL_PRIORITIES, + 'default': ExotelPriority.NORMAL, + }, + }) + + def __init__(self, sid, token, source, targets=None, unicode=None, + priority=None, region_name=None, **kwargs): + """ + Initialize Exotel Object + """ + super().__init__(**kwargs) + + # API SID (associated with account) + self.sid = validate_regex(sid) + if not self.sid: + msg = 'An invalid Exotel SID ' \ + '({}) was specified.'.format(sid) + self.logger.warning(msg) + raise TypeError(msg) + + # API Key (associated with project) + self.token = validate_regex( + token, *self.template_tokens['token']['regex']) + if not self.token: + msg = 'An invalid Exotel API Token ' \ + '({}) was specified.'.format(token) + self.logger.warning(msg) + raise TypeError(msg) + + # Store our region + try: + self.region_name = self.template_args['region']['default'] \ + if region_name is None else region_name.lower() + + if self.region_name not in EXOTEL_REGIONS: + # allow the outer except to handle this common response + raise + except: + # Invalid region specified + msg = 'The Exotel region specified ({}) is invalid.' \ + .format(region_name) + self.logger.warning(msg) + raise TypeError(msg) + + # Define whether or not we should set the unicode flag + self.unicode = self.template_args['unicode']['default'] \ + if unicode is None else bool(unicode) + + # + # 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 + self.priority = self.template_args['priority']['default'] + + except ValueError: + # Input is a string; attempt to get the lookup from our + # priority mapping + 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 priority else None + + # Now test to see if we got a match + if not result: + msg = 'An invalid Exotel priority ' \ + '({}) was specified.'.format(priority) + self.logger.warning(msg) + raise TypeError(msg) + + # 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) + if not result: + msg = 'The Account (From) Phone # specified ' \ + '({}) is invalid.'.format(source) + self.logger.warning(msg) + raise TypeError(msg) + + # Store our parsed value + self.source = result['full'] + + # Parse our targets + self.targets = list() + + for target in parse_phone_no(targets): + # Validate targets and drop bad ones: + result = is_phone_no(target) + if not result: + self.logger.warning( + 'Dropped invalid phone # ' + '({}) specified.'.format(target), + ) + continue + + # store valid phone number + self.targets.append(result['full']) + + return + + def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs): + """ + Perform Exotel Notification + """ + + # error tracking (used for function return) + has_error = False + + # Prepare our headers + headers = { + 'User-Agent': self.app_id, + 'Content-Type': 'application/x-www-form-urlencoded', + } + + # Our authentication + auth = (self.sid, self.token) + + # Prepare our payload + payload = { + 'From': self.source, + 'Body': body, + 'EncodingType': ExotelEncoding.UNICODE + if self.unicode else ExotelEncoding.TEXT, + 'priority': self.priority, + 'StatusCallback': None, + + # The to gets populated in the loop below + 'To': None, + } + + # Create a copy of the targets list + targets = list(self.targets) + + # 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) + + # Prepare our user + payload['To'] = target + + # Some Debug Logging + self.logger.debug('Exotel POST URL: {} (cert_verify={})'.format( + notify_url, self.verify_certificate)) + self.logger.debug('Exotel Payload: {}' .format(payload)) + + # Always call throttle before any remote server i/o is made + self.throttle() + + try: + r = requests.post( + notify_url, + auth=auth, + data=payload, + headers=headers, + verify=self.verify_certificate, + timeout=self.request_timeout, + ) + + if r.status_code != requests.codes.ok: + # We had a problem + status_str = \ + NotifyExotel.http_response_code_lookup( + r.status_code) + + self.logger.warning( + 'Failed to send Exotel notification to {}: ' + '{}{}error={}.'.format( + target, + status_str, + ', ' if status_str else '', + r.status_code)) + + self.logger.debug( + 'Response Details:\r\n{}'.format(r.content)) + + # Mark our failure + has_error = True + continue + + else: + self.logger.info( + 'Sent Exotel notification to %s.' % target) + + except requests.RequestException as e: + self.logger.warning( + 'A Connection error occurred sending Exotel:%s ' + 'notification.' % target + ) + self.logger.debug('Socket Exception: %s' % str(e)) + + # Mark our failure + has_error = True + continue + + return not has_error + + def url(self, privacy=False, *args, **kwargs): + """ + Returns the URL built dynamically based on specified arguments. + """ + + # Define any URL parameters + params = { + 'unicode': 'yes' if self.unicode else 'no', + 'region': self.region_name, + 'priority': self.priority, + } + + # Extend our parameters + params.update(self.url_parameters(privacy=privacy, *args, **kwargs)) + + return '{schema}://{sid}:{token}@{source}/{targets}/?{params}'.format( + schema=self.secure_protocol[0], + 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]), + params=NotifyExotel.urlencode(params)) + + @staticmethod + def parse_url(url): + """ + Parses the URL and returns enough arguments that can allow + us to re-instantiate this object. + + """ + results = NotifyBase.parse_url(url, verify_host=False) + if not results: + # We're done early as we couldn't load the results + return results + + # Get our entries; split_path() looks after unquoting content for us + # by default + results['targets'] = NotifyExotel.split_path(results['fullpath']) + + # The hostname is our source number + results['source'] = NotifyExotel.unquote(results['host']) + + # Get our account_sid and token from the user/pass config + results['sid'] = NotifyExotel.unquote(results['user']) + results['token'] = NotifyExotel.unquote(results['password']) + + # Get region + if 'region' in results['qsd'] and len(results['qsd']['region']): + results['region_name'] = \ + NotifyExotel.unquote(results['qsd']['region']) + + # API Token + if 'token' in results['qsd'] and len(results['qsd']['token']): + # Extract the Token from an argument + results['token'] = \ + NotifyExotel.unquote(results['qsd']['token']) + + # API SID + if 'sid' in results['qsd'] and len(results['qsd']['sid']): + # Extract the API SID from an argument + results['sid'] = \ + NotifyExotel.unquote(results['qsd']['sid']) + + # Support the 'from' and 'source' variable so that we can support + # targets this way too. + # The 'from' makes it easier to use yaml configuration + if 'from' in results['qsd'] and len(results['qsd']['from']): + results['source'] = \ + NotifyExotel.unquote(results['qsd']['from']) + if 'source' in results['qsd'] and len(results['qsd']['source']): + results['source'] = \ + NotifyExotel.unquote(results['qsd']['source']) + + # Support the 'to' variable so that we can support rooms this way too + # The 'to' makes it easier to use yaml configuration + if 'to' in results['qsd'] and len(results['qsd']['to']): + results['targets'] += \ + NotifyExotel.parse_phone_no(results['qsd']['to']) + + # Get priority + if 'priority' in results['qsd'] and len(results['qsd']['priority']): + results['priority'] = \ + NotifyExotel.unquote(results['qsd']['priority']) + + return results From d1ebef506a6d1f0c99d4eb2721a8f5b53878e5df Mon Sep 17 00:00:00 2001 From: Chris Caron Date: Sun, 18 Dec 2022 16:30:05 -0500 Subject: [PATCH 2/2] test coverage added --- KEYWORDS | 1 + README.md | 1 + apprise/plugins/NotifyExotel.py | 58 +++++----- packaging/redhat/python-apprise.spec | 2 +- test/helpers/rest.py | 22 ++-- test/test_plugin_exotel.py | 155 +++++++++++++++++++++++++++ 6 files changed, 202 insertions(+), 37 deletions(-) create mode 100644 test/test_plugin_exotel.py diff --git a/KEYWORDS b/KEYWORDS index a4753cdc5d..d87596664d 100644 --- a/KEYWORDS +++ b/KEYWORDS @@ -12,6 +12,7 @@ Dingtalk Discord Email Emby +Exotel Faast FCM Flock diff --git a/README.md b/README.md index cdbb9a115b..b0277b48b0 100644 --- a/README.md +++ b/README.md @@ -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
clicksend://user:pass@ToPhoneNo1/ToPhoneNo2/ToPhoneNoN | [DAPNET](https://github.com/caronc/apprise/wiki/Notify_dapnet) | dapnet:// | (TCP) 80 | dapnet://user:pass@callsign
dapnet://user:pass@callsign1/callsign2/callsignN | [D7 Networks](https://github.com/caronc/apprise/wiki/Notify_d7networks) | d7sms:// | (TCP) 443 | d7sms://user:pass@PhoneNo
d7sms://user:pass@ToPhoneNo1/ToPhoneNo2/ToPhoneNoN +| [Exotel](https://github.com/caronc/apprise/wiki/Notify_exotel) | exotel:// | (TCP) 443 | exotel://sid:token@FromPhoneNo
exotel://sid:token@FromPhoneNo/ToPhoneNo
exotel://sid:token@FromPhoneNo/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN | [DingTalk](https://github.com/caronc/apprise/wiki/Notify_dingtalk) | dingtalk:// | (TCP) 443 | dingtalk://token/
dingtalk://token/ToPhoneNo
dingtalk://token/ToPhoneNo1/ToPhoneNo2/ToPhoneNo1/ | [Kavenegar](https://github.com/caronc/apprise/wiki/Notify_kavenegar) | kavenegar:// | (TCP) 443 | kavenegar://ApiKey/ToPhoneNo
kavenegar://FromPhoneNo@ApiKey/ToPhoneNo
kavenegar://ApiKey/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN | [MessageBird](https://github.com/caronc/apprise/wiki/Notify_messagebird) | msgbird:// | (TCP) 443 | msgbird://ApiKey/FromPhoneNo
msgbird://ApiKey/FromPhoneNo/ToPhoneNo
msgbird://ApiKey/FromPhoneNo/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN/ diff --git a/apprise/plugins/NotifyExotel.py b/apprise/plugins/NotifyExotel.py index cd66acf161..4c0e821891 100644 --- a/apprise/plugins/NotifyExotel.py +++ b/apprise/plugins/NotifyExotel.py @@ -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 @@ def __init__(self, sid, token, source, targets=None, unicode=None, """ 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 @@ def __init__(self, sid, token, source, targets=None, unicode=None, 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 @@ def __init__(self, sid, token, source, targets=None, unicode=None, # # 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 @@ def __init__(self, sid, token, source, targets=None, unicode=None, # 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 @@ def __init__(self, sid, token, source, targets=None, unicode=None, 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 @@ def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs): 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 @@ def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs): # 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 @@ def url(self, privacy=False, *args, **kwargs): 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 diff --git a/packaging/redhat/python-apprise.spec b/packaging/redhat/python-apprise.spec index 474d22ded2..e029cf148c 100644 --- a/packaging/redhat/python-apprise.spec +++ b/packaging/redhat/python-apprise.spec @@ -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, diff --git a/test/helpers/rest.py b/test/helpers/rest.py index 218c0772f5..5e060928b8 100644 --- a/test/helpers/rest.py +++ b/test/helpers/rest.py @@ -183,8 +183,8 @@ def run(self, url, meta, mock_post, mock_get): 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 @@ def run(self, url, meta, mock_post, mock_get): # 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 @@ def __notify(self, url, obj, meta, asset, mock_del, mock_put, mock_head, 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 diff --git a/test/test_plugin_exotel.py b/test/test_plugin_exotel.py new file mode 100644 index 0000000000..d3144c692e --- /dev/null +++ b/test/test_plugin_exotel.py @@ -0,0 +1,155 @@ +# -*- coding: utf-8 -*- +# +# Copyright (C) 2022 Chris Caron +# 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={}®ion=us'.format( + 'a' * 32, 'b' * 32, '5' * 11), { + # Test region flag (Us) + 'instance': NotifyExotel, + }), + ('exotel://_?sid={}&token={}&from={}®ion=in'.format( + 'a' * 32, 'b' * 32, '5' * 11), { + # Test region flag (India) + 'instance': NotifyExotel, + }), + ('exotel://_?sid={}&token={}&from={}®ion=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()