Browse Source

Added (Simple) Threema Gateway Support (#993)

pull/1002/head
Chris Caron 1 year ago committed by GitHub
parent
commit
f1d697fce0
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
  1. 1
      KEYWORDS
  2. 1
      README.md
  3. 7
      apprise/URLBase.py
  4. 370
      apprise/plugins/NotifyThreema.py
  5. 15
      apprise/utils.py
  6. 4
      packaging/redhat/python-apprise.spec
  7. 178
      test/test_plugin_threema.py

1
KEYWORDS

@ -88,6 +88,7 @@ Stride
Syslog Syslog
Techulus Techulus
Telegram Telegram
Threema Gateway
Twilio Twilio
Twist Twist
Twitter Twitter

1
README.md

@ -146,6 +146,7 @@ The table below identifies the services this tool supports and some example serv
| [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/
| [Threema Gateway](https://github.com/caronc/apprise/wiki/Notify_threema) | threema:// | (TCP) 443 | threema://GatewayID@secret/ToPhoneNo<br/>threema://GatewayID@secret/ToEmail<br/>threema://GatewayID@secret/ToThreemaID/<br/>threema://GatewayID@secret/ToEmail/ToThreemaID/ToPhoneNo/...
| [Twilio](https://github.com/caronc/apprise/wiki/Notify_twilio) | twilio:// | (TCP) 443 | twilio://AccountSid:AuthToken@FromPhoneNo<br/>twilio://AccountSid:AuthToken@FromPhoneNo/ToPhoneNo<br/>twilio://AccountSid:AuthToken@FromPhoneNo/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN/<br/>twilio://AccountSid:AuthToken@FromPhoneNo/ToPhoneNo?apikey=Key<br/>twilio://AccountSid:AuthToken@ShortCode/ToPhoneNo<br/>twilio://AccountSid:AuthToken@ShortCode/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN/ | [Twilio](https://github.com/caronc/apprise/wiki/Notify_twilio) | twilio:// | (TCP) 443 | twilio://AccountSid:AuthToken@FromPhoneNo<br/>twilio://AccountSid:AuthToken@FromPhoneNo/ToPhoneNo<br/>twilio://AccountSid:AuthToken@FromPhoneNo/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN/<br/>twilio://AccountSid:AuthToken@FromPhoneNo/ToPhoneNo?apikey=Key<br/>twilio://AccountSid:AuthToken@ShortCode/ToPhoneNo<br/>twilio://AccountSid:AuthToken@ShortCode/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN/
| [Voipms](https://github.com/caronc/apprise/wiki/Notify_voipms) | voipms:// | (TCP) 443 | voipms://password:email/FromPhoneNo<br/>voipms://password:email/FromPhoneNo/ToPhoneNo<br/>voipms://password:email/FromPhoneNo/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN/ | [Voipms](https://github.com/caronc/apprise/wiki/Notify_voipms) | voipms:// | (TCP) 443 | voipms://password:email/FromPhoneNo<br/>voipms://password:email/FromPhoneNo/ToPhoneNo<br/>voipms://password:email/FromPhoneNo/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN/
| [Vonage](https://github.com/caronc/apprise/wiki/Notify_nexmo) (formerly Nexmo) | nexmo:// | (TCP) 443 | nexmo://ApiKey:ApiSecret@FromPhoneNo<br/>nexmo://ApiKey:ApiSecret@FromPhoneNo/ToPhoneNo<br/>nexmo://ApiKey:ApiSecret@FromPhoneNo/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN/ | [Vonage](https://github.com/caronc/apprise/wiki/Notify_nexmo) (formerly Nexmo) | nexmo:// | (TCP) 443 | nexmo://ApiKey:ApiSecret@FromPhoneNo<br/>nexmo://ApiKey:ApiSecret@FromPhoneNo/ToPhoneNo<br/>nexmo://ApiKey:ApiSecret@FromPhoneNo/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN/

7
apprise/URLBase.py

@ -550,7 +550,7 @@ class URLBase:
return paths return paths
@staticmethod @staticmethod
def parse_list(content, unquote=True): def parse_list(content, allow_whitespace=True, unquote=True):
"""A wrapper to utils.parse_list() with unquoting support """A wrapper to utils.parse_list() with unquoting support
Parses a specified set of data and breaks it into a list. Parses a specified set of data and breaks it into a list.
@ -559,6 +559,9 @@ class URLBase:
content (str): The path to split up into a list. If a list is content (str): The path to split up into a list. If a list is
provided, then it's individual entries are processed. provided, then it's individual entries are processed.
allow_whitespace (:obj:`bool`, optional): whitespace is to be
treated as a delimiter
unquote (:obj:`bool`, optional): call unquote on each element unquote (:obj:`bool`, optional): call unquote on each element
added to the returned list. added to the returned list.
@ -566,7 +569,7 @@ class URLBase:
list: A unique list containing all of the elements in the path list: A unique list containing all of the elements in the path
""" """
content = parse_list(content) content = parse_list(content, allow_whitespace=allow_whitespace)
if unquote: if unquote:
content = \ content = \
[URLBase.unquote(x) for x in filter(bool, content)] [URLBase.unquote(x) for x in filter(bool, content)]

370
apprise/plugins/NotifyThreema.py

@ -0,0 +1,370 @@
# -*- coding: utf-8 -*-
# BSD 2-Clause License
#
# Apprise - Push Notification Library.
# Copyright (c) 2023, 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.
# Create an account https://gateway.threema.ch/en/ if you don't already have
# one
#
# Read more about Threema Gateway API here:
# - https://gateway.threema.ch/en/developer/api
import requests
from itertools import chain
from .NotifyBase import NotifyBase
from ..common import NotifyType
from ..utils import is_phone_no
from ..utils import validate_regex
from ..utils import is_email
from ..URLBase import PrivacyMode
from ..utils import parse_list
from ..AppriseLocale import gettext_lazy as _
class ThreemaRecipientTypes:
"""
The supported recipient specifiers
"""
THREEMA_ID = 'to'
PHONE = 'phone'
EMAIL = 'email'
class NotifyThreema(NotifyBase):
"""
A wrapper for Threema Gateway Notifications
"""
# The default descriptive name associated with the Notification
service_name = 'Threema Gateway'
# The services URL
service_url = 'https://gateway.threema.ch/'
# The default protocol
secure_protocol = 'threema'
# A URL that takes you to the setup/help of the specific protocol
setup_url = 'https://github.com/caronc/apprise/wiki/Notify_threema'
# Threema Gateway uses the http protocol with JSON requests
notify_url = 'https://msgapi.threema.ch/send_simple'
# The maximum length of the body
body_maxlen = 3500
# No title support
title_maxlen = 0
# Define object templates
templates = (
'{schema}://{gateway_id}@{secret}/{targets}',
)
# Define our template tokens
template_tokens = dict(NotifyBase.template_tokens, **{
'gateway_id': {
'name': _('Gateway ID'),
'type': 'string',
'private': True,
'required': True,
'map_to': 'user',
},
'secret': {
'name': _('API Secret'),
'type': 'string',
'private': True,
'required': True,
},
'target_phone': {
'name': _('Target Phone No'),
'type': 'string',
'prefix': '+',
'regex': (r'^[0-9\s)(+-]+$', 'i'),
'map_to': 'targets',
},
'target_email': {
'name': _('Target Email'),
'type': 'string',
'map_to': 'targets',
},
'target_threema_id': {
'name': _('Target Threema ID'),
'type': 'string',
'map_to': 'targets',
},
'targets': {
'name': _('Targets'),
'type': 'list:string',
'required': True,
},
})
# Define our template arguments
template_args = dict(NotifyBase.template_args, **{
'to': {
'alias_of': 'targets',
},
'from': {
'alias_of': 'gateway_id',
},
'gwid': {
'alias_of': 'gateway_id',
},
'secret': {
'alias_of': 'secret',
},
})
def __init__(self, secret=None, targets=None, **kwargs):
"""
Initialize Threema Gateway Object
"""
super().__init__(**kwargs)
# Validate our params here.
if not self.user:
msg = 'Threema Gateway ID must be specified'
self.logger.warning(msg)
raise TypeError(msg)
# Verify our Gateway ID
if len(self.user) != 8:
msg = 'Threema Gateway ID must be 8 characters in length'
self.logger.warning(msg)
raise TypeError(msg)
# Verify our secret
self.secret = validate_regex(secret)
if not self.secret:
msg = \
'An invalid Threema API Secret ({}) was specified'.format(
secret)
self.logger.warning(msg)
raise TypeError(msg)
# Parse our targets
self.targets = list()
# Used for URL generation afterwards only
self.invalid_targets = list()
for target in parse_list(targets, allow_whitespace=False):
if len(target) == 8:
# Store our user
self.targets.append(
(ThreemaRecipientTypes.THREEMA_ID, target))
continue
# Check if an email was defined
result = is_email(target)
if result:
# Store our user
self.targets.append(
(ThreemaRecipientTypes.EMAIL, result['full_email']))
continue
# Validate targets and drop bad ones:
result = is_phone_no(target)
if result:
# store valid phone number
self.targets.append((
ThreemaRecipientTypes.PHONE, result['full']))
continue
self.logger.warning(
'Dropped invalid user/email/phone '
'({}) specified'.format(target),
)
self.invalid_targets.append(target)
return
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
Perform Threema Gateway Notification
"""
if len(self.targets) == 0:
# There were no services to notify
self.logger.warning(
'There were no Threema Gateway 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/x-www-form-urlencoded; charset=utf-8',
'Accept': '*/*',
}
# Prepare our payload
_payload = {
'secret': self.secret,
'from': self.user,
'text': body.encode('utf-8'),
}
# Create a copy of the targets list
targets = list(self.targets)
while len(targets):
# Get our target to notify
key, target = targets.pop(0)
# Prepare a payload object
payload = _payload.copy()
# Set Target
payload[key] = target
# Some Debug Logging
self.logger.debug(
'Threema Gateway GET URL: {} (cert_verify={})'.format(
self.notify_url, self.verify_certificate))
self.logger.debug('Threema Gateway Payload: {}' .format(payload))
# Always call throttle before any remote server i/o is made
self.throttle()
try:
r = requests.post(
self.notify_url,
params=payload,
headers=headers,
verify=self.verify_certificate,
timeout=self.request_timeout,
)
if r.status_code != requests.codes.ok:
# We had a problem
status_str = \
NotifyThreema.http_response_code_lookup(
r.status_code)
self.logger.warning(
'Failed to send Threema Gateway 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
# We wee successful
self.logger.info(
'Sent Threema Gateway notification to %s' % target)
except requests.RequestException as e:
self.logger.warning(
'A Connection error occurred sending Threema Gateway:%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 = self.url_parameters(privacy=privacy, *args, **kwargs)
schemaStr = \
'{schema}://{gatewayid}@{secret}/{targets}?{params}'
return schemaStr.format(
schema=self.secure_protocol,
gatewayid=NotifyThreema.quote(self.user),
secret=self.pprint(
self.secret, privacy, mode=PrivacyMode.Secret, safe=''),
targets='/'.join(chain(
[NotifyThreema.quote(x[1], safe='@+') for x in self.targets],
[NotifyThreema.quote(x, safe='@+')
for x in self.invalid_targets])),
params=NotifyThreema.urlencode(params))
def __len__(self):
"""
Returns the number of targets associated with this notification
"""
targets = len(self.targets)
return targets if targets > 0 else 1
@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
results['targets'] = list()
if 'secret' in results['qsd'] and len(results['qsd']['secret']):
results['secret'] = \
NotifyThreema.unquote(results['qsd']['secret'])
else:
results['secret'] = NotifyThreema.unquote(results['host'])
results['targets'] += \
NotifyThreema.split_path(results['fullpath'])
if 'from' in results['qsd'] and len(results['qsd']['from']):
results['user'] = \
NotifyThreema.unquote(results['qsd']['from'])
elif 'gwid' in results['qsd'] and len(results['qsd']['gwid']):
results['user'] = \
NotifyThreema.unquote(results['qsd']['gwid'])
if 'to' in results['qsd'] and len(results['qsd']['to']):
results['targets'] += \
NotifyThreema.parse_list(
results['qsd']['to'], allow_whitespace=False)
return results

15
apprise/utils.py

@ -96,6 +96,9 @@ VALID_QUERY_RE = re.compile(r'^(?P<path>.*[/\\])(?P<query>[^/\\]+)?$')
# This is useful when turning a string into a list # This is useful when turning a string into a list
STRING_DELIMITERS = r'[\[\]\;,\s]+' STRING_DELIMITERS = r'[\[\]\;,\s]+'
# String Delimiters without the whitespace
STRING_DELIMITERS_NO_WS = r'[\[\]\;,]+'
# Pre-Escape content since we reference it so much # Pre-Escape content since we reference it so much
ESCAPED_PATH_SEPARATOR = re.escape('\\/') ESCAPED_PATH_SEPARATOR = re.escape('\\/')
ESCAPED_WIN_PATH_SEPARATOR = re.escape('\\') ESCAPED_WIN_PATH_SEPARATOR = re.escape('\\')
@ -1116,7 +1119,7 @@ def urlencode(query, doseq=False, safe='', encoding=None, errors=None):
errors=errors) errors=errors)
def parse_list(*args, cast=None): def parse_list(*args, cast=None, allow_whitespace=True):
""" """
Take a string list and break it into a delimited Take a string list and break it into a delimited
list of arguments. This funciton also supports list of arguments. This funciton also supports
@ -1143,10 +1146,12 @@ def parse_list(*args, cast=None):
arg = cast(arg) arg = cast(arg)
if isinstance(arg, str): if isinstance(arg, str):
result += re.split(STRING_DELIMITERS, arg) result += re.split(
STRING_DELIMITERS if allow_whitespace
else STRING_DELIMITERS_NO_WS, arg)
elif isinstance(arg, (set, list, tuple)): elif isinstance(arg, (set, list, tuple)):
result += parse_list(*arg) result += parse_list(*arg, allow_whitespace=allow_whitespace)
# #
# filter() eliminates any empty entries # filter() eliminates any empty entries
@ -1154,7 +1159,9 @@ def parse_list(*args, cast=None):
# Since Python v3 returns a filter (iterator) whereas Python v2 returned # Since Python v3 returns a filter (iterator) whereas Python v2 returned
# a list, we need to change it into a list object to remain compatible with # a list, we need to change it into a list object to remain compatible with
# both distribution types. # both distribution types.
return sorted([x for x in filter(bool, list(set(result)))]) return sorted([x for x in filter(bool, list(set(result)))]) \
if allow_whitespace else sorted(
[x.strip() for x in filter(bool, list(set(result))) if x.strip()])
def is_exclusive_match(logic, data, match_all=common.MATCH_ALL_TAG, def is_exclusive_match(logic, data, match_all=common.MATCH_ALL_TAG,

4
packaging/redhat/python-apprise.spec

@ -49,8 +49,8 @@ OneSignal, Opsgenie, PagerDuty, PagerTree, ParsePlatform, PopcornNotify,
Prowl, Pushalot, PushBullet, Pushjet, PushMe, Pushover, PushSafer, Pushy, Prowl, Pushalot, PushBullet, Pushjet, PushMe, Pushover, PushSafer, Pushy,
PushDeer, Reddit, Rocket.Chat, RSyslog, SendGrid, ServerChan, Signal, PushDeer, Reddit, Rocket.Chat, RSyslog, SendGrid, ServerChan, Signal,
SimplePush, Sinch, Slack, SMSEagle, SMTP2Go, Spontit, SparkPost, Super Toasty, SimplePush, Sinch, Slack, SMSEagle, SMTP2Go, Spontit, SparkPost, Super Toasty,
Streamlabs, Stride, Syslog, Techulus Push, Telegram, Twilio, Twitter, Twist, Streamlabs, Stride, Syslog, Techulus Push, Telegram, Threema Gateway, Twilio,
XBMC, Voipms, Vonage, WhatsApp, Webex Teams} Twitter, Twist, XBMC, Voipms, Vonage, WhatsApp, Webex Teams}
Name: python-%{pypi_name} Name: python-%{pypi_name}
Version: 1.6.0 Version: 1.6.0

178
test/test_plugin_threema.py

@ -0,0 +1,178 @@
# -*- coding: utf-8 -*-
# BSD 2-Clause License
#
# Apprise - Push Notification Library.
# Copyright (c) 2023, 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
from apprise.plugins.NotifyThreema import NotifyThreema
from helpers import AppriseURLTester
# Disable logging for a cleaner testing output
import logging
logging.disable(logging.CRITICAL)
# Our Testing URLs
apprise_url_tests = (
('threema://', {
# No user/secret specified
'instance': TypeError,
}),
('threema://@:', {
# Invalid url
'instance': TypeError,
}),
('threema://user@secret', {
# gateway id must be 8 characters in len
'instance': TypeError,
}),
('threema://*THEGWID@secret/{targets}/'.format(
targets='/'.join(['2222'])), {
# Invalid target phone number
'instance': NotifyThreema,
'notify_response': False,
'privacy_url': 'threema://%2ATHEGWID@****/2222',
}),
('threema://*THEGWID@secret/{targets}/'.format(
targets='/'.join(['16134442222'])), {
# Valid
'instance': NotifyThreema,
'privacy_url': 'threema://%2ATHEGWID@****/16134442222',
}),
('threema://*THEGWID@secret/{targets}/'.format(
targets='/'.join(['16134442222', '16134443333'])), {
# Valid multiple targets
'instance': NotifyThreema,
'privacy_url': 'threema://%2ATHEGWID@****/16134442222/16134443333',
}),
('threema:///?secret=secret&from=*THEGWID&to={targets}'.format(
targets=','.join(['16134448888', 'user1@gmail.com', 'abcd1234'])), {
# Valid
'instance': NotifyThreema,
}),
('threema:///?secret=secret&gwid=*THEGWID&to={targets}'.format(
targets=','.join(['16134448888', 'user2@gmail.com', 'abcd1234'])), {
# Valid
'instance': NotifyThreema,
}),
('threema://*THEGWID@secret', {
'instance': NotifyThreema,
# No targets specified
'notify_response': False,
}),
('threema://*THEGWID@secret/16134443333', {
'instance': NotifyThreema,
# throw a bizzare code forcing us to fail to look it up
'response': False,
'requests_response_code': 999,
}),
('threema://*THEGWID@secret/16134443333', {
'instance': NotifyThreema,
# Throws a series of errors
'test_requests_exceptions': True,
}),
)
def test_plugin_threema():
"""
NotifyThreema() Apprise URLs
"""
# Run our general tests
AppriseURLTester(tests=apprise_url_tests).run_all()
@ mock.patch('requests.post')
def test_plugin_threema_edge_cases(mock_post):
"""
NotifyThreema() Edge Cases
"""
# 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
gwid = '*THEGWID'
secret = 'mysecret'
targets = '+1 (555) 123-9876'
# No email specified
with pytest.raises(TypeError):
NotifyThreema(user=gwid, secret=None, targets=targets)
results = NotifyThreema.parse_url(
f'threema://?gwid={gwid}&secret={secret}&to={targets}')
assert isinstance(results, dict)
assert results['user'] == gwid
assert results['secret'] == secret
assert results['password'] is None
assert results['port'] is None
assert results['host'] == ''
assert results['fullpath'] == '/'
assert results['path'] == '/'
assert results['query'] is None
assert results['schema'] == 'threema'
assert results['url'] == 'threema:///'
assert isinstance(results['targets'], list) is True
assert len(results['targets']) == 1
assert results['targets'][0] == '+1 (555) 123-9876'
instance = NotifyThreema(**results)
assert len(instance.targets) == 1
assert instance.targets[0] == ('phone', '15551239876')
assert isinstance(instance, NotifyThreema)
response = instance.send(title='title', body='body 😊')
assert response is True
assert mock_post.call_count == 1
details = mock_post.call_args_list[0]
assert details[0][0] == 'https://msgapi.threema.ch/send_simple'
assert details[1]['headers']['User-Agent'] == 'Apprise'
assert details[1]['headers']['Accept'] == '*/*'
assert details[1]['headers']['Content-Type'] == \
'application/x-www-form-urlencoded; charset=utf-8'
assert details[1]['params']['secret'] == secret
assert details[1]['params']['from'] == gwid
assert details[1]['params']['phone'] == '15551239876'
assert details[1]['params']['text'] == 'body 😊'.encode('utf-8')
Loading…
Cancel
Save