Add Société Française du Radiotéléphone (SFR) Support (#1132)

This commit is contained in:
Anghille 2024-06-09 22:44:44 +02:00 committed by GitHub
parent 81caf9299b
commit 3896b4a3e8
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 1004 additions and 2 deletions

View File

@ -81,6 +81,7 @@ Ryver
SendGrid SendGrid
ServerChan ServerChan
SES SES
SFR
Signal Signal
SimplePush SimplePush
Sinch Sinch

View File

@ -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<br/>kavenegar://FromPhoneNo@ApiKey/ToPhoneNo<br/>kavenegar://ApiKey/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN | [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/ | [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/
| [MSG91](https://github.com/caronc/apprise/wiki/Notify_msg91) | msg91:// | (TCP) 443 | msg91://TemplateID@AuthKey/ToPhoneNo<br/>msg91://TemplateID@AuthKey/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN/ | [MSG91](https://github.com/caronc/apprise/wiki/Notify_msg91) | msg91:// | (TCP) 443 | msg91://TemplateID@AuthKey/ToPhoneNo<br/>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://\<serviceId\>:\<servicePassword\>@\<spaceId\>/\<phoneNumber\>?param=value
| [Signal API](https://github.com/caronc/apprise/wiki/Notify_signal) | signal:// or signals:// | (TCP) 80 or 443 | signal://hostname:port/FromPhoneNo<br/>signal://hostname:port/FromPhoneNo/ToPhoneNo<br/>signal://hostname:port/FromPhoneNo/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN/ | [Signal API](https://github.com/caronc/apprise/wiki/Notify_signal) | signal:// or signals:// | (TCP) 80 or 443 | signal://hostname:port/FromPhoneNo<br/>signal://hostname:port/FromPhoneNo/ToPhoneNo<br/>signal://hostname:port/FromPhoneNo/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN/
| [Sinch](https://github.com/caronc/apprise/wiki/Notify_sinch) | sinch:// | (TCP) 443 | sinch://ServicePlanId:ApiToken@FromPhoneNo<br/>sinch://ServicePlanId:ApiToken@FromPhoneNo/ToPhoneNo<br/>sinch://ServicePlanId:ApiToken@FromPhoneNo/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN/<br/>sinch://ServicePlanId:ApiToken@ShortCode/ToPhoneNo<br/>sinch://ServicePlanId:ApiToken@ShortCode/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN/ | [Sinch](https://github.com/caronc/apprise/wiki/Notify_sinch) | sinch:// | (TCP) 443 | sinch://ServicePlanId:ApiToken@FromPhoneNo<br/>sinch://ServicePlanId:ApiToken@FromPhoneNo/ToPhoneNo<br/>sinch://ServicePlanId:ApiToken@FromPhoneNo/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN/<br/>sinch://ServicePlanId:ApiToken@ShortCode/ToPhoneNo<br/>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<br/>smseagles://hostname:port/@ToContact<br/>smseagles://hostname:port/#ToGroup<br/>smseagles://hostname:port/ToPhoneNo1/#ToGroup/@ToContact/ | [SMSEagle](https://github.com/caronc/apprise/wiki/Notify_smseagle) | smseagle:// or smseagles:// | (TCP) 80 or 443 | smseagles://hostname:port/ToPhoneNo<br/>smseagles://hostname:port/@ToContact<br/>smseagles://hostname:port/#ToGroup<br/>smseagles://hostname:port/ToPhoneNo1/#ToGroup/@ToContact/

431
apprise/plugins/sfr.py Normal file
View File

@ -0,0 +1,431 @@
# -*- coding: utf-8 -*-
# BSD 2-Clause License
#
# Apprise - Push Notification Library.
# Copyright (c) 2024, Chris Caron <lead2gold@gmail.com>
#
# 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/<apiGroup>/<apicall>
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

View File

@ -48,8 +48,8 @@ Microsoft Teams, Misskey, MQTT, MSG91, MyAndroid, Nexmo, Nextcloud,
NextcloudTalk, Notica, Notifiarr, Notifico, ntfy, Office365, OneSignal, NextcloudTalk, Notica, Notifiarr, Notifico, ntfy, Office365, OneSignal,
Opsgenie, PagerDuty, PagerTree, ParsePlatform, PopcornNotify, Prowl, Pushalot, Opsgenie, PagerDuty, PagerTree, ParsePlatform, PopcornNotify, Prowl, Pushalot,
PushBullet, Pushjet, PushMe, Pushover, PushSafer, Pushy, PushDeer, Revolt, PushBullet, Pushjet, PushMe, Pushover, PushSafer, Pushy, PushDeer, Revolt,
Reddit, Rocket.Chat, RSyslog, SendGrid, ServerChan, Signal, SimplePush, Sinch, Reddit, Rocket.Chat, RSyslog, SendGrid, ServerChan, SFR, Signal, SimplePush,
Slack, SMSEagle, SMS Manager, SMTP2Go, SparkPost, Splunk, Super Toasty, Sinch, Slack, SMSEagle, SMS Manager, SMTP2Go, SparkPost, Splunk, Super Toasty,
Streamlabs, Stride, Synology Chat, Syslog, Techulus Push, Telegram, Threema Streamlabs, Stride, Synology Chat, Syslog, Techulus Push, Telegram, Threema
Gateway, Twilio, Twitter, Twist, XBMC, VictorOps, Voipms, Vonage, WeCom Bot, Gateway, Twilio, Twitter, Twist, XBMC, VictorOps, Voipms, Vonage, WeCom Bot,
WhatsApp, Webex Teams} WhatsApp, Webex Teams}

569
test/test_plugin_sfr.py Normal file
View File

@ -0,0 +1,569 @@
# -*- coding: utf-8 -*-
# BSD 2-Clause License
#
# Apprise - Push Notification Library.
# Copyright (c) 2024, Chris Caron <lead2gold@gmail.com>
#
# 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,
)