From 41fe86224119f4d7d081ed2efd9036d0295c805a Mon Sep 17 00:00:00 2001 From: Chris Caron Date: Sat, 16 Apr 2022 15:34:30 -0400 Subject: [PATCH] Signal API Support Added (#568) --- README.md | 1 + apprise/AppriseAsset.py | 17 ++ apprise/plugins/NotifySignalAPI.py | 400 +++++++++++++++++++++++++++ packaging/redhat/python-apprise.spec | 19 +- setup.py | 2 +- test/test_plugin_signal.py | 217 +++++++++++++++ 6 files changed, 646 insertions(+), 10 deletions(-) create mode 100644 apprise/plugins/NotifySignalAPI.py create mode 100644 test/test_plugin_signal.py diff --git a/README.md b/README.md index b7f8bbd0..e5c4008d 100644 --- a/README.md +++ b/README.md @@ -80,6 +80,7 @@ The table below identifies the services this tool supports and some example serv | [Ryver](https://github.com/caronc/apprise/wiki/Notify_ryver) | ryver:// | (TCP) 443 | ryver://Organization/Token
ryver://botname@Organization/Token | [SendGrid](https://github.com/caronc/apprise/wiki/Notify_sendgrid) | sendgrid:// | (TCP) 443 | sendgrid://APIToken:FromEmail/
sendgrid://APIToken:FromEmail/ToEmail
sendgrid://APIToken:FromEmail/ToEmail1/ToEmail2/ToEmailN/ | [ServerChan](https://github.com/caronc/apprise/wiki/Notify_serverchan) | schan:// | (TCP) 443 | schan://sendkey/ +| [Signal API](https://github.com/caronc/apprise/wiki/Notify_signal) | signal:// or signals:// | (TCP) 80 or 443 | signal://hostname:port/FromPhoneNo
signal://hostname:port/FromPhoneNo/ToPhoneNo
signal://hostname:port/FromPhoneNo/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN/ | [SimplePush](https://github.com/caronc/apprise/wiki/Notify_simplepush) | spush:// | (TCP) 443 | spush://apikey
spush://salt:password@apikey
spush://apikey?event=Apprise | [Slack](https://github.com/caronc/apprise/wiki/Notify_slack) | slack:// | (TCP) 443 | slack://TokenA/TokenB/TokenC/
slack://TokenA/TokenB/TokenC/Channel
slack://botname@TokenA/TokenB/TokenC/Channel
slack://user@TokenA/TokenB/TokenC/Channel1/Channel2/ChannelN | [SMTP2Go](https://github.com/caronc/apprise/wiki/Notify_smtp2go) | smtp2go:// | (TCP) 443 | smtp2go://user@hostname/apikey
smtp2go://user@hostname/apikey/email
smtp2go://user@hostname/apikey/email1/email2/emailN
smtp2go://user@hostname/apikey/?name="From%20User" diff --git a/apprise/AppriseAsset.py b/apprise/AppriseAsset.py index ce7ace5e..b2275de3 100644 --- a/apprise/AppriseAsset.py +++ b/apprise/AppriseAsset.py @@ -59,6 +59,14 @@ class AppriseAsset(object): NotifyType.WARNING: '#CACF29', } + # Ascii Notification + ascii_notify_map = { + NotifyType.INFO: '[i]', + NotifyType.SUCCESS: '[+]', + NotifyType.FAILURE: '[!]', + NotifyType.WARNING: '[~]', + } + # The default color to return if a mapping isn't found in our table above default_html_color = '#888888' @@ -185,6 +193,15 @@ class AppriseAsset(object): raise ValueError( 'AppriseAsset html_color(): An invalid color_type was specified.') + def ascii(self, notify_type): + """ + Returns an ascii representation based on passed in notify type + + """ + + # look our response up + return self.ascii_notify_map.get(notify_type, self.default_html_color) + def image_url(self, notify_type, image_size, logo=False, extension=None): """ Apply our mask to our image URL diff --git a/apprise/plugins/NotifySignalAPI.py b/apprise/plugins/NotifySignalAPI.py new file mode 100644 index 00000000..a753215b --- /dev/null +++ b/apprise/plugins/NotifySignalAPI.py @@ -0,0 +1,400 @@ +# -*- coding: utf-8 -*- +# +# Copyright (C) 2020 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 json import dumps + +from .NotifyBase import NotifyBase +from ..common import NotifyType +from ..utils import is_phone_no +from ..utils import parse_phone_no +from ..utils import parse_bool +from ..URLBase import PrivacyMode +from ..AppriseLocale import gettext_lazy as _ + + +class NotifySignalAPI(NotifyBase): + """ + A wrapper for SignalAPI Notifications + """ + + # The default descriptive name associated with the Notification + service_name = 'Signal API' + + # The services URL + service_url = 'https://bbernhard.github.io/signal-cli-rest-api/' + + # The default protocol + protocol = 'signal' + + # The default protocol + secure_protocol = 'signals' + + # A URL that takes you to the setup/help of the specific protocol + setup_url = 'https://github.com/caronc/apprise/wiki/Notify_signal' + + # The maximum targets to include when doing batch transfers + default_batch_size = 10 + + # We don't support titles for Signal notifications + title_maxlen = 0 + + # Define object templates + templates = ( + '{schema}://{host}/{from_phone}', + '{schema}://{host}:{port}/{from_phone}', + '{schema}://{user}@{host}/{from_phone}', + '{schema}://{user}@{host}:{port}/{from_phone}', + '{schema}://{user}:{password}@{host}/{from_phone}', + '{schema}://{user}:{password}@{host}:{port}/{from_phone}', + '{schema}://{host}/{from_phone}/{targets}', + '{schema}://{host}:{port}/{from_phone}/{targets}', + '{schema}://{user}@{host}/{from_phone}/{targets}', + '{schema}://{user}@{host}:{port}/{from_phone}/{targets}', + '{schema}://{user}:{password}@{host}/{from_phone}/{targets}', + '{schema}://{user}:{password}@{host}:{port}/{from_phone}/{targets}', + ) + + # Define our template tokens + template_tokens = dict(NotifyBase.template_tokens, **{ + 'host': { + 'name': _('Hostname'), + 'type': 'string', + 'required': True, + }, + 'port': { + 'name': _('Port'), + 'type': 'int', + 'min': 1, + 'max': 65535, + }, + 'user': { + 'name': _('Username'), + 'type': 'string', + }, + 'password': { + 'name': _('Password'), + 'type': 'string', + 'private': 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', + }, + 'batch': { + 'name': _('Batch Mode'), + 'type': 'bool', + 'default': False, + }, + 'status': { + 'name': _('Show Status'), + 'type': 'bool', + 'default': False, + }, + }) + + def __init__(self, source=None, targets=None, batch=False, status=False, + **kwargs): + """ + Initialize SignalAPI Object + """ + super(NotifySignalAPI, self).__init__(**kwargs) + + # Prepare Batch Mode Flag + self.batch = batch + + # Set Status type + self.status = status + + # Parse our targets + self.targets = list() + + # Used for URL generation afterwards only + self.invalid_targets = list() + + # Manage our Source Phone + result = is_phone_no(source) + if not result: + msg = 'An invalid Signal API Source Phone No ' \ + '({}) was provided.'.format(source) + self.logger.warning(msg) + raise TypeError(msg) + + self.source = '+{}'.format(result['full']) + + if targets: + # Validate our targerts + 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), + ) + self.invalid_targets.append(target) + continue + + # store valid phone number + self.targets.append('+{}'.format(result['full'])) + else: + # Send a message to ourselves + self.targets.append(self.source) + + return + + def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs): + """ + Perform Signal API Notification + """ + + if len(self.targets) == 0: + # There were no services to notify + self.logger.warning( + 'There were no Signal API targets to notify.') + return False + + # error tracking (used for function return) + has_error = False + + # Prepare our headers + headers = { + 'User-Agent': self.app_id, + 'Content-Type': 'application/json', + } + + # Prepare our payload + payload = { + 'message': "{}{}".format( + '' if not self.status else '{} '.format( + self.asset.ascii(notify_type)), body), + "number": self.source, + "recipients": [] + } + + # Determine Authentication + auth = None + if self.user: + auth = (self.user, self.password) + + # Set our schema + schema = 'https' if self.secure else 'http' + + # Construct our URL + notify_url = '%s://%s' % (schema, self.host) + if isinstance(self.port, int): + notify_url += ':%d' % self.port + notify_url += '/v2/send' + + # Send in batches if identified to do so + batch_size = 1 if not self.batch else self.default_batch_size + + for index in range(0, len(self.targets), batch_size): + # Prepare our recipients + payload['recipients'] = self.targets[index:index + batch_size] + + self.logger.debug('Signal API POST URL: %s (cert_verify=%r)' % ( + notify_url, self.verify_certificate, + )) + self.logger.debug('Signal API Payload: %s' % str(payload)) + + # Always call throttle before any remote server i/o is made + self.throttle() + try: + r = requests.post( + notify_url, + auth=auth, + data=dumps(payload), + headers=headers, + verify=self.verify_certificate, + timeout=self.request_timeout, + ) + if r.status_code != requests.codes.ok: + # We had a problem + status_str = \ + NotifySignalAPI.http_response_code_lookup( + r.status_code) + + self.logger.warning( + 'Failed to send {} Signal API notification{}: ' + '{}{}error={}.'.format( + len(self.targets[index:index + batch_size]), + ' to {}'.format(self.targets[index]) + if batch_size == 1 else '(s)', + 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 {} Signal API notification{}.' + .format( + len(self.targets[index:index + batch_size]), + ' to {}'.format(self.targets[index]) + if batch_size == 1 else '(s)', + )) + + except requests.RequestException as e: + self.logger.warning( + 'A Connection error occured sending {} Signal API ' + 'notification(s).'.format( + len(self.targets[index:index + batch_size]))) + 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 = { + 'batch': 'yes' if self.batch else 'no', + 'status': 'yes' if self.status else 'no', + } + + # Extend our parameters + params.update(self.url_parameters(privacy=privacy, *args, **kwargs)) + + # Determine Authentication + auth = '' + if self.user and self.password: + auth = '{user}:{password}@'.format( + user=NotifySignalAPI.quote(self.user, safe=''), + password=self.pprint( + self.password, privacy, mode=PrivacyMode.Secret, safe=''), + ) + elif self.user: + auth = '{user}@'.format( + user=NotifySignalAPI.quote(self.user, safe=''), + ) + + default_port = 443 if self.secure else 80 + + # So we can strip out our own phone (if present); create a copy of our + # targets + if len(self.targets) == 1 and self.source in self.targets: + targets = [] + + elif len(self.targets) == 0: + # invalid phone-no were specified + targets = self.invalid_targets + + else: + targets = list(self.targets) + + return '{schema}://{auth}{hostname}{port}/{src}/{dst}?{params}'.format( + schema=self.secure_protocol if self.secure else self.protocol, + auth=auth, + # never encode hostname since we're expecting it to be a valid one + hostname=self.host, + port='' if self.port is None or self.port == default_port + else ':{}'.format(self.port), + src=self.source, + dst='/'.join( + [NotifySignalAPI.quote(x, safe='') for x in targets]), + params=NotifySignalAPI.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'] = \ + NotifySignalAPI.split_path(results['fullpath']) + + # The hostname is our authentication key + results['apikey'] = NotifySignalAPI.unquote(results['host']) + + if 'from' in results['qsd'] and len(results['qsd']['from']): + results['source'] = \ + NotifySignalAPI.unquote(results['qsd']['from']) + + elif results['targets']: + # The from phone no is the first entry in the list otherwise + results['source'] = results['targets'].pop(0) + + # Support the 'to' variable so that we can support targets this way too + # The 'to' makes it easier to use yaml configuration + if 'to' in results['qsd'] and len(results['qsd']['to']): + results['targets'] += \ + NotifySignalAPI.parse_phone_no(results['qsd']['to']) + + # Get Batch Mode Flag + results['batch'] = \ + parse_bool(results['qsd'].get('batch', False)) + + # Get status switch + results['status'] = \ + parse_bool(results['qsd'].get('status', False)) + + return results diff --git a/packaging/redhat/python-apprise.spec b/packaging/redhat/python-apprise.spec index 57cb0b44..102f2a03 100644 --- a/packaging/redhat/python-apprise.spec +++ b/packaging/redhat/python-apprise.spec @@ -47,15 +47,16 @@ Apprise is a Python package for simplifying access to all of the different notification services that are out there. Apprise opens the door and makes it easy to access: -Apprise API, AWS SES, AWS SNS, Boxcar, ClickSend, DAPNET, DingTalk, Discord, E-Mail, -Emby, Faast, FCM, Flock, Gitter, Google Chat, Gotify, Growl, Home Assistant, -IFTTT, Join, Kavenegar, KODI, Kumulos, LaMetric, MacOSX, Mailgun, Mattermost, -Matrix, Microsoft Windows, Microsoft Teams, MessageBird, MQTT, MSG91, MyAndroid, -Nexmo, Nextcloud, NextcloudTalk, Notica, Notifico, ntfy, Office365, OneSignal, -Opsgenie, ParsePlatform, PopcornNotify, Prowl, Pushalot, PushBullet, Pushjet, -Pushover, PushSafer, Reddit, Rocket.Chat, SendGrid, ServerChan, SimplePush, Sinch, -Slack, SMTP2Go, Spontit, SparkPost, Super Toasty, Streamlabs, Stride, Syslog, -Techulus Push, Telegram, Twilio, Twitter, Twist, XBMC, XMPP, Webex Teams} +Apprise API, AWS SES, AWS SNS, Boxcar, ClickSend, DAPNET, DingTalk, Discord, +E-Mail, Emby, Faast, FCM, Flock, Gitter, Google Chat, Gotify, Growl, Home +Assistant, IFTTT, Join, Kavenegar, KODI, Kumulos, LaMetric, MacOSX, Mailgun, +Mattermost, Matrix, Microsoft Windows, Microsoft Teams, MessageBird, MQTT, MSG91, +MyAndroid, Nexmo, Nextcloud, NextcloudTalk, Notica, Notifico, ntfy, Office365, +OneSignal, Opsgenie, ParsePlatform, PopcornNotify, Prowl, Pushalot, +PushBullet, Pushjet, Pushover, PushSafer, Reddit, Rocket.Chat, SendGrid, +ServerChan, Signal, SimplePush, Sinch, Slack, SMTP2Go, Spontit, SparkPost, +Super Toasty, Streamlabs, Stride, Syslog, Techulus Push, Telegram, Twilio, +Twitter, Twist, XBMC, XMPP, Webex Teams} Name: python-%{pypi_name} Version: 0.9.7 diff --git a/setup.py b/setup.py index 9b30e921..9fb655cd 100755 --- a/setup.py +++ b/setup.py @@ -76,7 +76,7 @@ setup( 'MSG91 Nexmo Nextcloud NextcloudTalk Notica Notifico Ntfy Office365 ' 'OneSignal Opsgenie ParsePlatform PopcornNotify Prowl PushBullet ' 'Pushjet Pushed Pushover PushSafer Reddit Rocket.Chat Ryver SendGrid ' - 'ServerChan SimplePush Sinch Slack SMTP2Go SparkPost Spontit ' + 'ServerChan Signal SimplePush Sinch Slack SMTP2Go SparkPost Spontit ' 'Streamlabs Stride Syslog Techulus Telegram Twilio Twist Twitter XBMC ' 'MSTeams Microsoft Windows Webex CLI API', author='Chris Caron', diff --git a/test/test_plugin_signal.py b/test/test_plugin_signal.py new file mode 100644 index 00000000..444783b4 --- /dev/null +++ b/test/test_plugin_signal.py @@ -0,0 +1,217 @@ +# -*- coding: utf-8 -*- +# +# Copyright (C) 2021 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 json import loads +import mock +import pytest +import requests +from apprise import plugins +from apprise import Apprise +from helpers import AppriseURLTester + +# Disable logging for a cleaner testing output +import logging +logging.disable(logging.CRITICAL) + +# Our Testing URLs +apprise_url_tests = ( + ('signal://', { + # No host specified + 'instance': TypeError, + }), + ('signal://:@/', { + # invalid host + 'instance': TypeError, + }), + ('signal://localhost', { + # Just a host provided + 'instance': TypeError, + }), + ('signal://localhost', { + # key and secret provided and from but invalid from no + 'instance': TypeError, + }), + ('signal://localhost/123', { + # invalid from phone + 'instance': TypeError, + + }), + ('signal://localhost/{}/123/'.format('1' * 11), { + # invalid 'to' phone number + 'instance': plugins.NotifySignalAPI, + # Notify will fail because it couldn't send to anyone + 'response': False, + # Our expected url(privacy=True) startswith() response: + 'privacy_url': 'signal://localhost/+{}/123'.format('1' * 11), + }), + ('signal://localhost:8080/{}/'.format('1' * 11), { + # one phone number will notify ourselves + 'instance': plugins.NotifySignalAPI, + }), + + ('signal://localhost:8080/?from={}&to={},{}'.format( + '1' * 11, '2' * 11, '3' * 11), { + # use get args to acomplish the same thing + 'instance': plugins.NotifySignalAPI, + }), + ('signal://localhost:8080/?from={}&to={},{},{}'.format( + '1' * 11, '2' * 11, '3' * 11, '5' * 3), { + # 2 good targets and one invalid one + 'instance': plugins.NotifySignalAPI, + }), + ('signal://localhost:8080/{}/{}/?from={}'.format( + '1' * 11, '2' * 11, '3' * 11), { + # If we have from= specified, then all elements take on the to= value + 'instance': plugins.NotifySignalAPI, + }), + ('signals://user@localhost/{}/{}'.format('1' * 11, '3' * 11), { + # use get args to acomplish the same thing (use source instead of from) + 'instance': plugins.NotifySignalAPI, + }), + ('signals://user:password@localhost/{}/{}'.format('1' * 11, '3' * 11), { + # use get args to acomplish the same thing (use source instead of from) + 'instance': plugins.NotifySignalAPI, + }), + ('signals://localhost/{}/{}/{}?batch=True'.format( + '1' * 11, '3' * 11, '4' * 11), { + # test batch mode + 'instance': plugins.NotifySignalAPI, + }), + ('signals://localhost/{}/{}/{}?status=True'.format( + '1' * 11, '3' * 11, '4' * 11), { + # test status switch + 'instance': plugins.NotifySignalAPI, + }), + ('signal://localhost/{}/{}'.format('1' * 11, '4' * 11), { + 'instance': plugins.NotifySignalAPI, + # throw a bizzare code forcing us to fail to look it up + 'response': False, + 'requests_response_code': 999, + }), + ('signal://localhost/{}/{}'.format('1' * 11, '4' * 11), { + 'instance': plugins.NotifySignalAPI, + # 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_signal_urls(): + """ + NotifySignalAPI() Apprise URLs + + """ + + # Run our general tests + AppriseURLTester(tests=apprise_url_tests).run_all() + + +@mock.patch('requests.post') +def test_plugin_signal_edge_cases(mock_post): + """ + NotifySignalAPI() Edge Cases + + """ + # Disable Throttling to speed testing + plugins.NotifyBase.request_rate_per_sec = 0 + + # Prepare our response + response = requests.Request() + response.status_code = requests.codes.ok + + # Prepare Mock + mock_post.return_value = response + + # Initialize some generic (but valid) tokens + source = '+1 (555) 123-3456' + target = '+1 (555) 987-5432' + body = "test body" + title = "My Title" + + # No apikey specified + with pytest.raises(TypeError): + plugins.NotifySignalAPI(source=None) + + aobj = Apprise() + assert aobj.add("signals://localhost:231/{}/{}".format(source, target)) + assert aobj.notify(title=title, body=body) + + assert mock_post.call_count == 1 + + details = mock_post.call_args_list[0] + assert details[0][0] == 'https://localhost:231/v2/send' + payload = loads(details[1]['data']) + assert payload['message'] == 'My Title\r\ntest body' + + # Reset our mock object + mock_post.reset_mock() + + aobj = Apprise() + assert aobj.add( + "signals://user@localhost:231/{}/{}?status=True".format( + source, target)) + assert aobj.notify(title=title, body=body) + + assert mock_post.call_count == 1 + + details = mock_post.call_args_list[0] + assert details[0][0] == 'https://localhost:231/v2/send' + payload = loads(details[1]['data']) + # Status flag is set + assert payload['message'] == '[i] My Title\r\ntest body' + + +@mock.patch('requests.post') +def test_plugin_signal_test_based_on_feedback(mock_post): + """ + NotifySignalAPI() User Feedback Test + + """ + # Disable Throttling to speed testing + plugins.NotifyBase.request_rate_per_sec = 0 + + # Prepare our response + response = requests.Request() + response.status_code = requests.codes.ok + + # Prepare Mock + mock_post.return_value = response + + body = "test body" + title = "My Title" + + aobj = Apprise() + aobj.add('signal://10.0.0.112:8080/+12512222222/+12513333333') + + assert aobj.notify(title=title, body=body) + + assert mock_post.call_count == 1 + + details = mock_post.call_args_list[0] + assert details[0][0] == 'http://10.0.0.112:8080/v2/send' + payload = loads(details[1]['data']) + assert payload['message'] == 'My Title\r\ntest body' + assert payload['number'] == "+12512222222" + assert payload['recipients'] == ["+12513333333"]