From 3896b4a3e8500969857aa7513fdf5b66a7e7558f Mon Sep 17 00:00:00 2001
From: Anghille <57528205+Anghille@users.noreply.github.com>
Date: Sun, 9 Jun 2024 22:44:44 +0200
Subject: [PATCH] =?UTF-8?q?Add=20Soci=C3=A9t=C3=A9=20Fran=C3=A7aise=20du?=
=?UTF-8?q?=20Radiot=C3=A9l=C3=A9phone=20(SFR)=20Support=20(#1132)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
---
KEYWORDS | 1 +
README.md | 1 +
apprise/plugins/sfr.py | 431 ++++++++++++++++++++
packaging/redhat/python-apprise.spec | 4 +-
test/test_plugin_sfr.py | 569 +++++++++++++++++++++++++++
5 files changed, 1004 insertions(+), 2 deletions(-)
create mode 100644 apprise/plugins/sfr.py
create mode 100644 test/test_plugin_sfr.py
diff --git a/KEYWORDS b/KEYWORDS
index fa931b2a..6ee6c93c 100644
--- a/KEYWORDS
+++ b/KEYWORDS
@@ -81,6 +81,7 @@ Ryver
SendGrid
ServerChan
SES
+SFR
Signal
SimplePush
Sinch
diff --git a/README.md b/README.md
index e8fd60f9..b088b480 100644
--- a/README.md
+++ b/README.md
@@ -153,6 +153,7 @@ The table below identifies the services this tool supports and some example serv
| [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/
| [MSG91](https://github.com/caronc/apprise/wiki/Notify_msg91) | msg91:// | (TCP) 443 | msg91://TemplateID@AuthKey/ToPhoneNo
msg91://TemplateID@AuthKey/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN/
+| [Société Française du Radiotéléphone (SFR)](https://github.com/caronc/apprise/wiki/Notify_sfr) | sfr:// | (TCP) 443 | sfr://\:\@\/\?param=value
| [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/
| [Sinch](https://github.com/caronc/apprise/wiki/Notify_sinch) | sinch:// | (TCP) 443 | sinch://ServicePlanId:ApiToken@FromPhoneNo
sinch://ServicePlanId:ApiToken@FromPhoneNo/ToPhoneNo
sinch://ServicePlanId:ApiToken@FromPhoneNo/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN/
sinch://ServicePlanId:ApiToken@ShortCode/ToPhoneNo
sinch://ServicePlanId:ApiToken@ShortCode/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN/
| [SMSEagle](https://github.com/caronc/apprise/wiki/Notify_smseagle) | smseagle:// or smseagles:// | (TCP) 80 or 443 | smseagles://hostname:port/ToPhoneNo
smseagles://hostname:port/@ToContact
smseagles://hostname:port/#ToGroup
smseagles://hostname:port/ToPhoneNo1/#ToGroup/@ToContact/
diff --git a/apprise/plugins/sfr.py b/apprise/plugins/sfr.py
new file mode 100644
index 00000000..c41e0752
--- /dev/null
+++ b/apprise/plugins/sfr.py
@@ -0,0 +1,431 @@
+# -*- coding: utf-8 -*-
+# BSD 2-Clause License
+#
+# Apprise - Push Notification Library.
+# Copyright (c) 2024, Chris Caron
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are met:
+#
+# 1. Redistributions of source code must retain the above copyright notice,
+# this list of conditions and the following disclaimer.
+#
+# 2. Redistributions in binary form must reproduce the above copyright notice,
+# this list of conditions and the following disclaimer in the documentation
+# and/or other materials provided with the distribution.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+# POSSIBILITY OF SUCH DAMAGE.
+
+# For this to work correctly you need to have a valid SFR DMC service account
+# to whicthe API password can be generated. A "space" is also necessary
+# (space = a logical separation between clients), which will give you a
+# specific spaceId
+#
+# Expected credentials looks a little like this:
+# serviceId: 84920958892 - Random numbers
+# servicePassword: XxXXxXXx - Random characters
+# spaceId: 984348 - Random numbers
+#
+# 1. Visit https://www.sfr.fr/
+#
+# 2. Url will look like this
+# https://www.dmc.sfr-sh.fr/DmcWS/1.5.8/JsonService//
+
+import requests
+import json
+
+from .base import NotifyBase
+from ..common import NotifyType
+from ..locale import gettext_lazy as _
+from ..utils import is_phone_no
+from ..utils import parse_phone_no
+from ..url import PrivacyMode
+
+
+class NotifySFR(NotifyBase):
+ """
+ A wrapper for SFR French Telecom DMC API
+ """
+
+ # The default descriptive name associated with the Notification
+ service_name = _('Société Française du Radiotéléphone')
+
+ # The services URL
+ service_url = 'https://www.sfr.fr/'
+
+ # The default protocol
+ protocol = 'sfr'
+
+ # A URL that takes you to the setup/help of the specific protocol
+ setup_url = 'https://github.com/caronc/apprise/wiki/Notify_sfr'
+
+ # SFR api
+ notify_url = (
+ 'https://www.dmc.sfr-sh.fr/DmcWS/1.5.8/JsonService/'
+ 'MessagesUnitairesWS/addSingleCall' # this is the actual api call
+ )
+
+ # The maximum length of the body
+ body_maxlen = 160
+
+ # 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}://{user}:{password}@{space_id}/{targets}',
+ )
+
+ # Define our tokens
+ template_tokens = dict(
+ NotifyBase.template_tokens, **{
+ 'user': {
+ 'name': _('Service ID'),
+ 'type': 'string',
+ 'required': True,
+ },
+ 'password': {
+ 'name': _('Service Password'),
+ 'type': 'string',
+ 'private': True,
+ 'required': True,
+ },
+ 'space_id': {
+ 'name': _('Space ID'),
+ 'type': 'string',
+ 'private': True,
+ 'required': True,
+ },
+ 'target': {
+ 'name': _('Recipient Phone Number'),
+ 'type': 'string',
+ 'regex': (r'^\+?[0-9\s)(+-]+$', 'i'),
+ 'map_to': 'targets',
+ },
+ 'targets': {
+ 'name': _('Targets'),
+ 'type': 'list:string',
+ 'required': True,
+ },
+ }
+ )
+
+ # Define our template arguments
+ template_args = dict(
+ NotifyBase.template_args, **{
+ 'lang': {
+ 'name': _('Language'),
+ 'type': 'string',
+ 'default': 'fr_FR',
+ 'required': True,
+ },
+ 'sender': {
+ 'name': _('Sender Name'),
+ 'type': 'string',
+ 'required': True,
+ 'default': '',
+ },
+ 'from': {
+ 'alias_of': 'sender'
+ },
+ 'media': {
+ 'name': _('Media Type'),
+ 'type': 'string',
+ 'required': True,
+ 'default': 'SMSUnicode',
+ 'values': ['SMS', 'SMSLong', 'SMSUnicode', 'SMSUnicodeLong'],
+ },
+ 'timeout': {
+ 'name': _('Timeout'),
+ 'type': 'int',
+ 'default': 2880,
+ 'required': False,
+ },
+ 'voice': {
+ 'name': _('TTS Voice'),
+ 'type': 'string',
+ 'default': 'claire08s',
+ 'values': ['claire08s', 'laura8k'],
+ 'required': False,
+ },
+ 'to': {
+ 'alias_of': 'targets',
+ },
+ }
+ )
+
+ def __init__(self, space_id=None, targets=None, lang=None, sender=None,
+ media=None, timeout=None, voice=None, **kwargs):
+ """
+ Initialize SFR Object
+ """
+ super().__init__(**kwargs)
+
+ if not (self.user and self.password):
+ msg = 'A SFR user (serviceId) and password (servicePassword) ' \
+ 'combination was not provided.'
+ self.logger.warning(msg)
+ raise TypeError(msg)
+
+ self.space_id = space_id
+ if not self.space_id:
+ msg = 'A SFR Space ID is required.'
+ self.logger.warning(msg)
+ raise TypeError(msg)
+
+ self.voice = voice \
+ if voice else self.template_args['voice']['default']
+ self.lang = lang \
+ if lang else self.template_args['lang']['default']
+ self.media = media \
+ if media else self.template_args['media']['default']
+ self.sender = sender \
+ if sender else self.template_args['sender']['default']
+
+ # Set our Time to Live Flag
+ self.timeout = self.template_args['timeout']['default']
+ try:
+ self.timeout = int(timeout)
+
+ except (ValueError, TypeError):
+ # set default timeout
+ self.timeout = 2880
+ pass
+
+ # 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'])
+
+ if not self.targets:
+ msg = ('No receiver phone number has been provided. Please '
+ 'provide as least one valid phone number.')
+ self.logger.warning(msg)
+ raise TypeError(msg)
+
+ return
+
+ def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
+ """
+ Perform the SFR notification
+ """
+
+ # error tracking (used for function return)
+ has_error = False
+
+ # Create a copy of the targets list
+ targets = list(self.targets)
+
+ # Construct the authentication JSON
+ auth_payload = json.dumps({
+ 'serviceId': self.user,
+ 'servicePassword': self.password,
+ 'spaceId': self.space_id,
+ 'lang': self.lang,
+ })
+
+ base_payload = {
+ # Can be 'SMS', 'SMSLong', 'SMSUnicode', or 'SMSUnicodeLong'
+ 'media': self.media,
+ # Content of the message
+ 'textMsg': body,
+ # Receiver's phone number (set below)
+ 'to': None,
+ # Optional, default to ''
+ 'from': self.sender,
+ # Optional, default 2880 minutes
+ 'timeout': self.timeout,
+ # Optional, default to French voice
+ 'ttsVoice': self.voice,
+ }
+
+ while len(targets):
+ # Get our target to notify
+ target = targets.pop(0)
+
+ # Prepare our target phone no
+ base_payload['to'] = target
+
+ # Always call throttle before any remote server i/o is made
+ self.throttle()
+
+ # Finalize our payload
+ payload = {
+ 'authenticate': auth_payload,
+ 'messageUnitaire': json.dumps(base_payload, ensure_ascii=True)
+ }
+
+ # Some Debug Logging
+ self.logger.debug('SFR POST URL: {} (cert_verify={})'.format(
+ self.notify_url, self.verify_certificate))
+ self.logger.debug('SFR Payload: {}' .format(payload))
+
+ try:
+ r = requests.post(
+ self.notify_url,
+ params=payload,
+ verify=self.verify_certificate,
+ timeout=self.request_timeout,
+ )
+
+ try:
+ content = json.loads(r.content)
+
+ except (AttributeError, TypeError, ValueError):
+ # ValueError = r.content is Unparsable
+ # TypeError = r.content is None
+ # AttributeError = r is None
+ content = {}
+
+ # Check if the request was successfull
+ if r.status_code not in (
+ requests.codes.ok,
+ requests.codes.no_content,
+ ):
+ # We had a problem
+ status_str = \
+ NotifySFR.http_response_code_lookup(
+ r.status_code)
+
+ self.logger.warning(
+ 'Failed to send SFR 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
+
+ # SFR returns a code 200 even if the authentication fails
+ # It then indicates in the content['success'] field the
+ # Actual state of the transaction
+ if not content.get('success', False):
+ self.logger.warning(
+ 'SFR Notification to {} was not sent by the server: '
+ 'server_error={}, fatal={}.'.format(
+ target,
+ content.get('errorCode', 'UNKNOWN'),
+ content.get('fatal', 'True'),
+ ))
+
+ # Mark our failure
+ has_error = True
+ continue
+
+ self.logger.info(
+ 'Sent SFR notification to %s.' % target)
+
+ except requests.RequestException as e:
+ self.logger.warning(
+ 'A Connection error occurred sending SFR:%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 = {
+ 'from': self.sender,
+ 'timeout': str(self.timeout),
+ 'voice': self.voice,
+ 'lang': self.lang,
+ 'media': self.media,
+ }
+
+ # Extend our parameters
+ params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
+
+ return '{schema}://{user}:{password}@{sid}/{targets}?{params}'.format(
+ schema=self.secure_protocol if self.secure else self.protocol,
+ user=self.user,
+ password=self.pprint(
+ self.password,
+ privacy,
+ mode=PrivacyMode.Secret,
+ safe='',
+ ),
+ sid=self.pprint(self.space_id, privacy, safe=''),
+ targets='/'.join(
+ [NotifySFR.quote(x, safe='') for x in self.targets]),
+ params=self.urlencode(params),
+ )
+
+ def __len__(self):
+ """
+ Returns the number of targets associated with this notification
+ """
+ return len(self.targets)
+
+ @staticmethod
+ def parse_url(url):
+ """
+ Parse the URL and return arguments required to initialize this plugin
+ """
+ # NotifyBase.parse_url() will make the initial parsing of your string
+ # very easy to use. It will tokenize the entire URL for you. The
+ # tokens are then passed into your __init__() function you defined to
+ # generate you're 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
+
+ # Extract user and password
+ results['space_id'] = results.get('host')
+ results['targets'] = NotifySFR.split_path(results['fullpath'])
+
+ # Extract additional parameters
+ qsd = results.get('qsd', {})
+ results['sender'] = \
+ NotifySFR.unquote(qsd.get('sender', qsd.get('from')))
+ results['timeout'] = NotifySFR.unquote(qsd.get('timeout'))
+ results['voice'] = NotifySFR.unquote(qsd.get('voice'))
+ results['lang'] = NotifySFR.unquote(qsd.get('lang'))
+ results['media'] = NotifySFR.unquote(qsd.get('media'))
+
+ # 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'] += \
+ NotifySFR.parse_phone_no(results['qsd']['to'])
+
+ return results
diff --git a/packaging/redhat/python-apprise.spec b/packaging/redhat/python-apprise.spec
index d4f38b8b..a801dee3 100644
--- a/packaging/redhat/python-apprise.spec
+++ b/packaging/redhat/python-apprise.spec
@@ -48,8 +48,8 @@ Microsoft Teams, Misskey, MQTT, MSG91, MyAndroid, Nexmo, Nextcloud,
NextcloudTalk, Notica, Notifiarr, Notifico, ntfy, Office365, OneSignal,
Opsgenie, PagerDuty, PagerTree, ParsePlatform, PopcornNotify, Prowl, Pushalot,
PushBullet, Pushjet, PushMe, Pushover, PushSafer, Pushy, PushDeer, Revolt,
-Reddit, Rocket.Chat, RSyslog, SendGrid, ServerChan, Signal, SimplePush, Sinch,
-Slack, SMSEagle, SMS Manager, SMTP2Go, SparkPost, Splunk, Super Toasty,
+Reddit, Rocket.Chat, RSyslog, SendGrid, ServerChan, SFR, Signal, SimplePush,
+Sinch, Slack, SMSEagle, SMS Manager, SMTP2Go, SparkPost, Splunk, Super Toasty,
Streamlabs, Stride, Synology Chat, Syslog, Techulus Push, Telegram, Threema
Gateway, Twilio, Twitter, Twist, XBMC, VictorOps, Voipms, Vonage, WeCom Bot,
WhatsApp, Webex Teams}
diff --git a/test/test_plugin_sfr.py b/test/test_plugin_sfr.py
new file mode 100644
index 00000000..7f052ecc
--- /dev/null
+++ b/test/test_plugin_sfr.py
@@ -0,0 +1,569 @@
+# -*- coding: utf-8 -*-
+# BSD 2-Clause License
+#
+# Apprise - Push Notification Library.
+# Copyright (c) 2024, Chris Caron
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are met:
+#
+# 1. Redistributions of source code must retain the above copyright notice,
+# this list of conditions and the following disclaimer.
+#
+# 2. Redistributions in binary form must reproduce the above copyright notice,
+# this list of conditions and the following disclaimer in the documentation
+# and/or other materials provided with the distribution.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+# POSSIBILITY OF SUCH DAMAGE.
+
+from unittest import mock
+import pytest
+import requests
+import json
+from apprise.plugins.sfr import NotifySFR
+from helpers import AppriseURLTester
+
+# Disable logging for a cleaner testing output
+import logging
+logging.disable(logging.CRITICAL)
+
+SFR_GOOD_RESPONSE = json.dumps({
+ "success": True,
+ "reponse": 8888888,
+})
+
+SFR_BAD_RESPONSE = json.dumps({
+ 'success': False,
+ 'errorCode': 'THIS_IS_AN_ERROR',
+ 'errorDetail': 'Appel api en erreur',
+ 'fatal': True,
+ 'invalidParams': True,
+})
+
+# Our Testing URLs
+apprise_url_tests = (
+ ('sfr://', {
+ # No host specified
+ 'instance': TypeError,
+ }),
+ ('sfr://:@/', {
+ # Invalid host
+ 'instance': TypeError,
+ }),
+ ('sfr://:service_password', {
+ # No user specified
+ 'instance': TypeError,
+ }),
+ ('sfr://testing:serv@ice_password', {
+ # Invalid Password
+ 'instance': TypeError,
+ }),
+ ('sfr://testing:service_password@/5555555555', {
+ # No spaceId provided
+ 'instance': TypeError,
+ }),
+ ('sfr://testing:service_password@12345/', {
+ # No target provided
+ 'instance': TypeError,
+ }),
+ ('sfr://:service_password@12345/{}'.format(3 * 13), {
+ # No host but everything else provided
+ 'instance': TypeError,
+ }),
+ ('sfr://:service_password@space_id/targets?media=TEST', {
+ 'instance': TypeError,
+ }),
+ ('sfr://service_id:', {
+ 'instance': TypeError,
+ }),
+ ('sfr://service_id:@', {
+ 'instance': TypeError,
+ }),
+ ('sfr://service_id:@{}'.format(
+ '0' * 3), {
+ 'instance': TypeError,
+ }),
+ ('sfr://service_id:@{}/'.format(
+ '0' * 3), {
+ 'instance': TypeError,
+ }),
+ ('sfr://service_id:@{}/targets'.format(
+ '0' * 3), {
+ 'instance': TypeError,
+ }),
+ ('sfr://service_id:@{}/targets?media=TEST'.format(
+ '0' * 3), {
+ 'instance': TypeError,
+ }),
+ ('sfr://service_id:service_password@{}/{}?from=MyApp&timeout=30'.format(
+ '0' * 3, '0' * 10), {
+ # a valid group
+ 'instance': NotifySFR,
+ # Our expected url(privacy=True) startswith() response:
+ 'privacy_url': (
+ 'sfr://service_id:****@0...0/0000000000?'
+ 'from=MyApp&timeout=30&voice=claire08s&'
+ 'lang=fr_FR&media=SMSUnicode&format=text'
+ '&overflow=upstream&rto=4.0&cto=4.0&verify=yes'),
+ # Our response expected server response
+ 'requests_response_text': SFR_GOOD_RESPONSE,
+ }),
+ ('sfr://service_id:service_password@{}/{}?voice=laura8k&lang=en_US'.format(
+ '0' * 3, '0' * 10), {
+ # a valid group
+ 'instance': NotifySFR,
+ # Our expected url(privacy=True) startswith() response:
+ 'privacy_url': (
+ 'sfr://service_id:****@0...0/0000000000?'
+ 'from=&timeout=2880&voice=laura8k&'
+ 'lang=en_US&media=SMSUnicode&format=text'
+ '&overflow=upstream&rto=4.0&cto=4.0&verify=yes'),
+ # Our response expected server response
+ 'requests_response_text': SFR_GOOD_RESPONSE,
+ }),
+ ('sfr://service_id:service_password@{}/{}?media=SMS'.format(
+ '0' * 3, '0' * 10), {
+ # a valid group
+ 'instance': NotifySFR,
+ # Our expected url(privacy=True) startswith() response:
+ 'privacy_url': (
+ 'sfr://service_id:****@0...0/0000000000?'
+ 'from=&timeout=2880&voice=claire08s&'
+ 'lang=fr_FR&media=SMS&format=text'
+ '&overflow=upstream&rto=4.0&cto=4.0&verify=yes'),
+ # Our response expected server response
+ 'requests_response_text': SFR_GOOD_RESPONSE,
+ }),
+ ('sfr://service_id:service_password@{}/{}'.format(
+ '0' * 3, '0' * 10), {
+ # Test case where we get a bad response
+ 'instance': NotifySFR,
+ # Our expected url(privacy=True) startswith() response:
+ 'privacy_url': (
+ 'sfr://service_id:****@0...0/0000000000?'
+ 'from=&timeout=2880&voice=claire08s&'
+ 'lang=fr_FR&media=SMSUnicode&format=text'
+ '&overflow=upstream&rto=4.0&cto=4.0&verify=yes'),
+ # Our failed notification expected server response
+ 'requests_response_text': SFR_BAD_RESPONSE,
+ 'requests_response_code': requests.codes.ok,
+ # as a result, we expect a failed notification
+ 'response': False,
+ }),
+)
+
+
+def test_plugin_sfr_urls():
+ """
+ NotifySFR() Apprise URLs
+ """
+ # Run our general tests
+ AppriseURLTester(tests=apprise_url_tests).run_all()
+
+
+@mock.patch('requests.post')
+def test_plugin_sfr_notification_ok(mock_post):
+ """
+ NotifySFR() Notifications Ok response
+ """
+ # Prepare Mock
+ # Create a mock response object
+ response = mock.Mock()
+ response.status_code = requests.codes.ok
+ response.content = SFR_GOOD_RESPONSE
+ mock_post.return_value = response
+
+ # Test our URL parsing
+ results = NotifySFR.parse_url(
+ 'sfr://srv:pwd@{}/{}?media=SMSLong'.format('1' * 8, '0' * 10))
+
+ assert isinstance(results, dict)
+ assert results['user'] == 'srv'
+ assert results['password'] == 'pwd'
+ assert results['space_id'] == '11111111'
+ assert results['targets'] == ['0000000000']
+ assert results['media'] == 'SMSLong'
+ assert results['timeout'] == ''
+ assert results['voice'] == ''
+ assert results['lang'] == ''
+ assert results['sender'] == ''
+
+ instance = NotifySFR(**results)
+ assert isinstance(instance, NotifySFR)
+ assert len(instance) == 1
+ assert instance.lang == 'fr_FR'
+ assert instance.lang == 'fr_FR'
+ assert instance.sender == ''
+ assert isinstance(instance.targets, list)
+ assert isinstance(instance.timeout, int)
+ assert isinstance(instance.voice, str)
+ assert isinstance(instance.space_id, str)
+
+ response = instance.send(body="test")
+ assert response is True
+ assert mock_post.call_count == 1
+
+
+@mock.patch('requests.post')
+def test_plugin_sfr_notification_multiple_targets_ok(mock_post):
+ """
+ NotifySFR() Notifications ko response
+ """
+ # Reset our object
+ mock_post.reset_mock()
+ # Prepare Mock
+ # Create a mock response object
+ response = mock.Mock()
+ response.status_code = requests.codes.ok
+ response.content = SFR_GOOD_RESPONSE
+ mock_post.return_value = response
+
+ # Test "real" parameters
+ results = NotifySFR.parse_url(
+ 'sfr://{}:other_fjv&8password@{}/?to={},{}&from=MyCustomUser'.format(
+ '4' * 6, '1' * 8, '6' * 10, '8' * 10))
+
+ assert isinstance(results, dict)
+ assert results['user'] == '444444'
+ assert results['password'] == 'other_fjv&8password'
+ assert results['space_id'] == '11111111'
+ assert results['targets'] == ['6666666666', '8888888888']
+ assert results['media'] == ''
+ assert results['timeout'] == ''
+ assert results['voice'] == ''
+ assert results['lang'] == ''
+ assert results['sender'] == 'MyCustomUser'
+
+ instance = NotifySFR(**results)
+ assert isinstance(instance, NotifySFR)
+ assert len(instance) == 2
+ assert instance.lang == 'fr_FR'
+ assert instance.sender == 'MyCustomUser'
+ assert instance.media == 'SMSUnicode'
+ assert isinstance(instance.targets, list)
+ assert instance.timeout == 2880
+ assert instance.voice == 'claire08s'
+ assert isinstance(instance.space_id, str)
+
+ response = instance.send(body="test")
+ assert response is True
+ assert mock_post.call_count == 2
+
+
+@mock.patch('requests.post')
+def test_plugin_sfr_notification_ko(mock_post):
+ """
+ NotifySFR() Notifications ko response
+ """
+ # Reset our object
+ mock_post.reset_mock()
+ # Prepare Mock
+ # Create a mock response object
+ response = mock.Mock()
+ response.status_code = requests.codes.ok
+ response.content = SFR_BAD_RESPONSE
+ mock_post.return_value = response
+
+ # Test "real" parameters
+ results = NotifySFR.parse_url(
+ 'sfr://{}:other_fjv&8password@{}/{}?timeout=30&media=SMS'.format(
+ '4' * 6, '1' * 8, '2' * 10))
+
+ assert isinstance(results, dict)
+ assert results['user'] == '444444'
+ assert results['password'] == 'other_fjv&8password'
+ assert results['space_id'] == '11111111'
+ assert results['media'] == 'SMS'
+ assert results['targets'] == ['2222222222']
+ assert results['timeout'] == '30'
+ assert results['voice'] == ''
+ assert results['lang'] == ''
+ assert results['sender'] == ''
+
+ instance = NotifySFR(**results)
+ assert isinstance(instance, NotifySFR)
+ assert len(instance) == 1
+ assert instance.lang == 'fr_FR'
+ assert instance.sender == ''
+ assert instance.media == 'SMS'
+ assert isinstance(instance.targets, list)
+ assert instance.timeout == 30
+ assert instance.voice == 'claire08s'
+ assert isinstance(instance.space_id, str)
+
+ response = instance.send(body="test")
+ assert response is False
+ assert mock_post.call_count == 1
+
+
+@mock.patch('requests.post')
+def test_plugin_sfr_notification_multiple_targets_all_ko(mock_post):
+ """
+ NotifySFR() Notifications ko response
+ """
+ # Reset our object
+ mock_post.reset_mock()
+ # Prepare Mock
+ # Create a mock response object
+ response = mock.Mock()
+ response.status_code = requests.codes.ok
+ response.content = SFR_BAD_RESPONSE
+ mock_post.return_value = response
+
+ # Test "real" parameters
+ results = NotifySFR.parse_url(
+ 'sfr://{}:other_fjv&8password@{}/?to={},{}&voice=laura8k'.format(
+ '4' * 6, '1' * 8, '6' * 4, '8' * 4))
+
+ assert isinstance(results, dict)
+ assert results['user'] == '444444'
+ assert results['password'] == 'other_fjv&8password'
+ assert results['space_id'] == '11111111'
+ assert results['targets'] == ['6666', '8888']
+ assert results['voice'] == 'laura8k'
+ assert results['media'] == ''
+ assert results['timeout'] == ''
+ assert results['lang'] == ''
+ assert results['sender'] == ''
+
+ # No valid phone number provided
+ with pytest.raises(TypeError):
+ NotifySFR(**results)
+
+
+@mock.patch('requests.post')
+def test_plugin_sfr_notification_multiple_targets_one_ko(mock_post):
+ """
+ NotifySFR() Notifications ko response
+ """
+ # Reset our object
+ mock_post.reset_mock()
+ # Prepare Mock
+ # Create a mock response object
+ response = mock.Mock()
+ response.status_code = requests.codes.ok
+ response.content = SFR_BAD_RESPONSE
+ mock_post.return_value = response
+
+ # Test "real" parameters
+ results = NotifySFR.parse_url(
+ 'sfr://{}:&pass@{}/?to={},{}&media=SMSUnicodeLong&lang=en_US'.format(
+ '4' * 6, '1' * 8, '6' * 10, '8' * 4))
+
+ assert isinstance(results, dict)
+ assert results['user'] == '444444'
+ assert results['password'] == '&pass'
+ assert results['space_id'] == '11111111'
+ assert results['targets'] == ['6666666666', '8888']
+ assert results['voice'] == ''
+ assert results['media'] == 'SMSUnicodeLong'
+ assert results['timeout'] == ''
+ assert results['lang'] == 'en_US'
+ assert results['sender'] == ''
+
+ instance = NotifySFR(**results)
+ assert isinstance(instance, NotifySFR)
+ assert len(instance) == 1
+ assert instance.lang == 'en_US'
+ assert instance.sender == ''
+ assert instance.media == 'SMSUnicodeLong'
+ assert isinstance(instance.targets, list)
+ assert instance.timeout == 2880
+ assert instance.voice == 'claire08s'
+ assert isinstance(instance.space_id, str)
+
+ # One phone number failed to be parsed, therefore notify fails
+ response = instance.send(body="test")
+ assert response is False
+ assert mock_post.call_count == 1
+
+
+@mock.patch('requests.post')
+def test_plugin_sfr_notification_exceptions(mock_post):
+ """
+ NotifySFR() Notifications exceptions
+ """
+ mock_post.reset_mock()
+ # Prepare Mock
+ # Create a mock response object
+ response = mock.Mock()
+ response.status_code = requests.codes.internal_server_error
+ response.content = SFR_GOOD_RESPONSE
+ mock_post.return_value = response
+
+ # Test "real" parameters
+ results = NotifySFR.parse_url(
+ 'sfr://{}:str0*fn_ppw0rd@{}/{}'.format(
+ "404ghwo89144", '9993384', '0959290404'))
+
+ assert isinstance(results, dict)
+ assert results['user'] == '404ghwo89144'
+ assert results['password'] == 'str0*fn_ppw0rd'
+ assert results['space_id'] == '9993384'
+ assert results['targets'] == ['0959290404']
+ assert results['media'] == ''
+ assert results['timeout'] == ''
+ assert results['lang'] == ''
+ assert results['sender'] == ''
+
+ instance = NotifySFR(**results)
+ assert isinstance(instance, NotifySFR)
+ assert len(instance) == 1
+ assert instance.lang == 'fr_FR'
+ assert instance.sender == ''
+ assert instance.media == 'SMSUnicode'
+ assert isinstance(instance.targets, list)
+ assert instance.timeout == 2880
+ assert instance.voice == 'claire08s'
+ assert isinstance(instance.space_id, str)
+
+ response = instance.send(body="test")
+ # Must return False
+ assert response is False
+ assert mock_post.call_count == 1
+
+ # Test invalid content returned by requests
+ mock_post.reset_mock()
+ response = mock.Mock()
+ response.status_code = requests.codes.ok
+ response.content = b'Invalid JSON Content'
+ mock_post.return_value = response
+
+ # Test "real" parameters
+ results = NotifySFR.parse_url(
+ 'sfr://{}:str0*fn_ppw0rd@{}/{}'.format(
+ "404ghwo89144", '9993384', '0959290404'))
+
+ assert isinstance(results, dict)
+ assert results['user'] == '404ghwo89144'
+ assert results['password'] == 'str0*fn_ppw0rd'
+ assert results['space_id'] == '9993384'
+ assert results['targets'] == ['0959290404']
+ assert results['media'] == ''
+ assert results['timeout'] == ''
+ assert results['lang'] == ''
+ assert results['sender'] == ''
+
+ instance = NotifySFR(**results)
+ assert isinstance(instance, NotifySFR)
+ assert len(instance) == 1
+ assert instance.lang == 'fr_FR'
+ assert instance.sender == ''
+ assert instance.media == 'SMSUnicode'
+ assert isinstance(instance.targets, list)
+ assert instance.timeout == 2880
+ assert instance.voice == 'claire08s'
+ assert isinstance(instance.space_id, str)
+
+ response = instance.send(body="test")
+ # Must return False
+ assert response is False
+ assert mock_post.call_count == 1
+
+
+@mock.patch(
+ 'requests.post',
+ side_effect=requests.RequestException("Connection error"),
+)
+def test_plugin_sfr_notification_exceptions_requests(mock_post):
+ """
+ NotifySFR() Notifications requests exceptions
+ """
+ # Test requests socket error return
+ mock_post.reset_mock()
+ # Prepare Mock
+ # Create a mock response object
+ response = mock.Mock()
+ response.status_code = requests.codes.internal_server_error
+ response.content = b'Invalid content'
+ mock_post.return_value = response
+
+ # Test "real" parameters
+ results = NotifySFR.parse_url(
+ 'sfr://{}:str0*fn_ppw0rd@{}/{}'.format(
+ "404ghwo89144", '9993384', '0959290404'))
+
+ assert isinstance(results, dict)
+ assert results['user'] == '404ghwo89144'
+ assert results['password'] == 'str0*fn_ppw0rd'
+ assert results['space_id'] == '9993384'
+ assert results['targets'] == ['0959290404']
+ assert results['media'] == ''
+ assert results['timeout'] == ''
+ assert results['lang'] == ''
+ assert results['sender'] == ''
+
+ instance = NotifySFR(**results)
+ assert isinstance(instance, NotifySFR)
+ assert len(instance) == 1
+ assert instance.lang == 'fr_FR'
+ assert instance.sender == ''
+ assert instance.media == 'SMSUnicode'
+ assert isinstance(instance.targets, list)
+ assert instance.timeout == 2880
+ assert instance.voice == 'claire08s'
+ assert isinstance(instance.space_id, str)
+
+ response = instance.send(body="test")
+ # Must return False do to requests error
+ assert response is False
+ assert mock_post.call_count == 1
+
+
+@mock.patch('requests.post')
+def test_plugin_sfr_failure(mock_post):
+ """
+ NotifySFR() Failure Cases
+ """
+ mock_post.reset_mock()
+ # Prepare Mock
+ # Create a mock response object
+ response = mock.Mock()
+ response.status_code = requests.codes.no_content
+ mock_post.return_value = response
+
+ # Invalid service_id
+ with pytest.raises(TypeError):
+ NotifySFR(
+ user=None,
+ password="service_password",
+ space_id=int('8' * 10),
+ targets=int('8' * 10),
+ )
+
+ # Invalid service_password
+ with pytest.raises(TypeError):
+ NotifySFR(
+ user="service_id",
+ password=None,
+ space_id=int('8' * 10),
+ targets=int('8' * 10),
+ )
+
+ # Invalid space_id
+ with pytest.raises(TypeError):
+ NotifySFR(
+ user="service_id",
+ password="service_password",
+ space_id=None,
+ targets=int('8' * 10),
+ )
+
+ # Invalid targets
+ with pytest.raises(TypeError):
+ NotifySFR(
+ user="service_id",
+ password="service_password",
+ space_id=int('8' * 10),
+ targets=None,
+ )