extern/apprise
1
0
Fork 0
mirror of https://github.com/caronc/apprise.git synced 2024-11-22 08:04:02 +01:00
Code Issues Packages Projects Releases Wiki Activity

Signal API Attachment and Group Support Added (#580)

Browse Source
This commit is contained in:
Chris Caron 2022-05-11 11:11:51 -04:00 committed by GitHub
parent ca0c8460f1
commit 9c145a842e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 242 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

85
apprise/plugins/NotifySignalAPI.py
View File

@ -1,6 +1,6 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2020 Chris Caron <lead2gold@gmail.com>
# Copyright (C) 2022 Chris Caron <lead2gold@gmail.com>
# All rights reserved.
#
# This code is licensed under the MIT License.
@ -23,8 +23,10 @@
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
import re
import requests
from json import dumps
import base64
from .NotifyBase import NotifyBase
from ..common import NotifyType
@ -35,6 +37,10 @@ from ..URLBase import PrivacyMode
from ..AppriseLocale import gettext_lazy as _
GROUP_REGEX = re.compile(
r'^\s*((\@|\%40)?(group\.)|\@|\%40)(?P<group>[a-z0-9_-]+)', re.I)
class NotifySignalAPI(NotifyBase):
"""
A wrapper for SignalAPI Notifications
@ -113,6 +119,13 @@ class NotifySignalAPI(NotifyBase):
'regex': (r'^[0-9\s)(+-]+$', 'i'),
'map_to': 'targets',
},
'target_channel': {
'name': _('Target Group ID'),
'type': 'string',
'prefix': '@',
'regex': (r'^[a-z0-9_-]+$', 'i'),
'map_to': 'targets',
},
'targets': {
'name': _('Targets'),
'type': 'list:string',
@ -173,23 +186,33 @@ class NotifySignalAPI(NotifyBase):
for target in parse_phone_no(targets):
# Validate targets and drop bad ones:
result = is_phone_no(target)
if not result:
if result:
# store valid phone number
self.targets.append('+{}'.format(result['full']))
continue
result = GROUP_REGEX.match(target)
if result:
# Just store group information
self.targets.append(
'group.{}'.format(result.group('group')))
continue
self.logger.warning(
'Dropped invalid phone # '
'Dropped invalid phone/group '
'({}) 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):
def send(self, body, title='', notify_type=NotifyType.INFO, attach=None,
**kwargs):
"""
Perform Signal API Notification
"""
@ -203,12 +226,50 @@ class NotifySignalAPI(NotifyBase):
# error tracking (used for function return)
has_error = False
attachments = []
if attach:
for attachment in attach:
# Perform some simple error checking
if not attachment:
# We could not access the attachment
self.logger.error(
'Could not access attachment {}.'.format(
attachment.url(privacy=True)))
return False
try:
with open(attachment.path, 'rb') as f:
# Prepare our Attachment in Base64
attachments.append(
base64.b64encode(f.read()).decode('utf-8'))
except (OSError, IOError) as e:
self.logger.warning(
'An I/O error occurred while reading {}.'.format(
attachment.name if attachment else 'attachment'))
self.logger.debug('I/O Exception: %s' % str(e))
return False
# Prepare our headers
headers = {
'User-Agent': self.app_id,
'Content-Type': 'application/json',
}
# Format defined here:
# https://bbernhard.github.io/signal-cli-rest-api\
# /#/Messages/post_v2_send
# Example:
# {
# "base64_attachments": [
# "string"
# ],
# "message": "string",
# "number": "string",
# "recipients": [
# "string"
# ]
# }
# Prepare our payload
payload = {
'message': "{}{}".format(
@ -218,6 +279,10 @@ class NotifySignalAPI(NotifyBase):
"recipients": []
}
if attachments:
# Store our attachments
payload['base64_attachments'] = attachments
# Determine Authentication
auth = None
if self.user:
@ -339,7 +404,11 @@ class NotifySignalAPI(NotifyBase):
targets = self.invalid_targets
else:
targets = list(self.targets)
# append @ to non-phone number entries as they are groups
# Remove group. prefix as well
targets = \
['@{}'.format(x[6:]) if x[0] != '+'
else x for x in self.targets]
return '{schema}://{auth}{hostname}{port}/{src}/{dst}?{params}'.format(
schema=self.secure_protocol if self.secure else self.protocol,
@ -350,7 +419,7 @@ class NotifySignalAPI(NotifyBase):
else ':{}'.format(self.port),
src=self.source,
dst='/'.join(
[NotifySignalAPI.quote(x, safe='') for x in targets]),
[NotifySignalAPI.quote(x, safe='@+') for x in targets]),
params=NotifySignalAPI.urlencode(params),
)

2
apprise/plugins/NotifySpontit.py
View File

@ -52,7 +52,7 @@ from ..AppriseLocale import gettext_lazy as _
# specified. If not, we use the user of the person sending the notification
# Finally the channel identifier is detected
CHANNEL_REGEX = re.compile(
r'^\s*(#|%23)?((@|%40)?(?P<user>[a-z0-9_]+)([/\\]|%2F))?'
r'^\s*(\#|\%23)?((\@|\%40)?(?P<user>[a-z0-9_]+)([/\\]|\%2F))?'
r'(?P<channel>[a-z0-9_-]+)\s*$', re.I)

3
apprise/plugins/NotifyXML.py
View File

@ -284,8 +284,7 @@ class NotifyXML(NotifyBase):
try:
with open(attachment.path, 'rb') as f:
# Output must be in a DataURL format (that's what
# PushSafer calls it):
# Prepare our Attachment in Base64
entry = \
'<Attachment filename="{}" mimetype="{}">'.format(
NotifyXML.escape_html(

161
test/test_plugin_signal.py
View File

@ -22,6 +22,8 @@
# 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 os
import sys
from json import loads
import mock
import pytest
@ -29,11 +31,16 @@ import requests
from apprise import plugins
from apprise import Apprise
from helpers import AppriseURLTester
from apprise import AppriseAttachment
from apprise import NotifyType
# Disable logging for a cleaner testing output
import logging
logging.disable(logging.CRITICAL)
# Attachment Directory
TEST_VAR_DIR = os.path.join(os.path.dirname(__file__), 'var')
# Our Testing URLs
apprise_url_tests = (
('signal://', {
@ -70,6 +77,18 @@ apprise_url_tests = (
'instance': plugins.NotifySignalAPI,
}),
('signal://localhost:8082/+{}/@group.abcd/'.format('2' * 11), {
# a valid group
'instance': plugins.NotifySignalAPI,
# Our expected url(privacy=True) startswith() response:
'privacy_url': 'signal://localhost:8082/+{}/@abcd'.format('2' * 11),
}),
('signal://localhost:8080/+{}/group.abcd/'.format('1' * 11), {
# another valid group (without @ symbol)
'instance': plugins.NotifySignalAPI,
# Our expected url(privacy=True) startswith() response:
'privacy_url': 'signal://localhost:8080/+{}/@abcd'.format('1' * 11),
}),
('signal://localhost:8080/?from={}&to={},{}'.format(
'1' * 11, '2' * 11, '3' * 11), {
# use get args to acomplish the same thing
@ -203,10 +222,13 @@ def test_plugin_signal_test_based_on_feedback(mock_post):
title = "My Title"
aobj = Apprise()
aobj.add('signal://10.0.0.112:8080/+12512222222/+12513333333')
aobj.add(
'signal://10.0.0.112:8080/+12512222222/+12513333333/'
'12514444444?batch=yes')
assert aobj.notify(title=title, body=body)
# If a batch, there is only 1 post
assert mock_post.call_count == 1
details = mock_post.call_args_list[0]
@ -214,4 +236,139 @@ def test_plugin_signal_test_based_on_feedback(mock_post):
payload = loads(details[1]['data'])
assert payload['message'] == 'My Title\r\ntest body'
assert payload['number'] == "+12512222222"
assert payload['recipients'] == ["+12513333333"]
assert len(payload['recipients']) == 2
assert "+12513333333" in payload['recipients']
# The + is appended
assert "+12514444444" in payload['recipients']
# Reset our test and turn batch mode off
mock_post.reset_mock()
aobj = Apprise()
aobj.add(
'signal://10.0.0.112:8080/+12512222222/+12513333333/'
'12514444444?batch=no')
assert aobj.notify(title=title, body=body)
# If a batch, there is only 1 post
assert mock_post.call_count == 2
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 len(payload['recipients']) == 1
assert "+12513333333" in payload['recipients']
details = mock_post.call_args_list[1]
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 len(payload['recipients']) == 1
# The + is appended
assert "+12514444444" in payload['recipients']
mock_post.reset_mock()
# Test group names
aobj = Apprise()
aobj.add(
'signal://10.0.0.112:8080/+12513333333/@group1/@group2/'
'12514444444?batch=yes')
assert aobj.notify(title=title, body=body)
# If a batch, there is only 1 post
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'] == "+12513333333"
assert len(payload['recipients']) == 3
assert "+12514444444" in payload['recipients']
# our groups
assert "group.group1" in payload['recipients']
assert "group.group2" in payload['recipients']
# Groups are stored properly
assert '/@group1' in aobj[0].url()
assert '/@group2' in aobj[0].url()
# Our target phone number is also in the path
assert '/+12514444444' in aobj[0].url()
@mock.patch('requests.post')
def test_notify_signal_plugin_attachments(mock_post):
"""
NotifySignalAPI() Attachments
"""
# Disable Throttling to speed testing
plugins.NotifyBase.request_rate_per_sec = 0
okay_response = requests.Request()
okay_response.status_code = requests.codes.ok
okay_response.content = ""
# Assign our mock object our return value
mock_post.return_value = okay_response
obj = Apprise.instantiate(
'signal://10.0.0.112:8080/+12512222222/+12513333333/'
'12514444444?batch=no')
assert isinstance(obj, plugins.NotifySignalAPI)
# Test Valid Attachment
path = os.path.join(TEST_VAR_DIR, 'apprise-test.gif')
attach = AppriseAttachment(path)
assert obj.notify(
body='body', title='title', notify_type=NotifyType.INFO,
attach=attach) is True
# Test invalid attachment
path = os.path.join(TEST_VAR_DIR, '/invalid/path/to/an/invalid/file.jpg')
assert obj.notify(
body='body', title='title', notify_type=NotifyType.INFO,
attach=path) is False
# Get a appropriate "builtin" module name for pythons 2/3.
if sys.version_info.major >= 3:
builtin_open_function = 'builtins.open'
else:
builtin_open_function = '__builtin__.open'
# Test Valid Attachment (load 3)
path = (
os.path.join(TEST_VAR_DIR, 'apprise-test.gif'),
os.path.join(TEST_VAR_DIR, 'apprise-test.gif'),
os.path.join(TEST_VAR_DIR, 'apprise-test.gif'),
)
attach = AppriseAttachment(path)
# Return our good configuration
mock_post.side_effect = None
mock_post.return_value = okay_response
with mock.patch(builtin_open_function, side_effect=OSError()):
# We can't send the message we can't open the attachment for reading
assert obj.notify(
body='body', title='title', notify_type=NotifyType.INFO,
attach=attach) is False
# test the handling of our batch modes
obj = Apprise.instantiate(
'signal://10.0.0.112:8080/+12512222222/+12513333333/'
'12514444444?batch=yes')
assert isinstance(obj, plugins.NotifySignalAPI)
# Now send an attachment normally without issues
mock_post.reset_mock()
assert obj.notify(
body='body', title='title', notify_type=NotifyType.INFO,
attach=attach) is True
assert mock_post.call_count == 1