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

Custom over-rides added to JSON and XML queries (#547)

Browse Source
This commit is contained in:
Chris Caron 2022-03-28 19:57:46 -04:00 committed by GitHub
parent 51dfdb5013
commit 4b4ec6bf30
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 165 additions and 127 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
apprise/plugins/NotifyJSON.py
View File

@ -122,9 +122,13 @@ class NotifyJSON(NotifyBase):
'name': _('HTTP Header'),
'prefix': '+',
},
'payload': {
'name': _('Payload Extras'),
'prefix': ':',
},
}
def __init__(self, headers=None, method=None, **kwargs):
def __init__(self, headers=None, method=None, payload=None, **kwargs):
"""
Initialize JSON Object
@ -151,6 +155,11 @@ class NotifyJSON(NotifyBase):
# Store our extra headers
self.headers.update(headers)
self.payload_extras = {}
if payload:
# Store our extra payload entries
self.payload_extras.update(payload)
return
def url(self, privacy=False, *args, **kwargs):
@ -169,6 +178,10 @@ class NotifyJSON(NotifyBase):
# Append our headers into our parameters
params.update({'+{}'.format(k): v for k, v in self.headers.items()})
# Append our payload extra's into our parameters
params.update(
{':{}'.format(k): v for k, v in self.payload_extras.items()})
# Determine Authentication
auth = ''
if self.user and self.password:
@ -252,6 +265,9 @@ class NotifyJSON(NotifyBase):
'type': notify_type,
}
# Apply any/all payload over-rides defined
payload.update(self.payload_extras)
auth = None
if self.user:
auth = (self.user, self.password)
@ -340,6 +356,10 @@ class NotifyJSON(NotifyBase):
# We're done early as we couldn't load the results
return results
# store any additional payload extra's defined
results['payload'] = {NotifyJSON.unquote(x): NotifyJSON.unquote(y)
for x, y in results['qsd:'].items()}
# Add our headers that the user can potentially over-ride if they wish
# to to our returned result set
results['headers'] = results['qsd+']

65
apprise/plugins/NotifyXML.py
View File

@ -127,9 +127,13 @@ class NotifyXML(NotifyBase):
'name': _('HTTP Header'),
'prefix': '+',
},
'payload': {
'name': _('Payload Extras'),
'prefix': ':',
},
}
def __init__(self, headers=None, method=None, **kwargs):
def __init__(self, headers=None, method=None, payload=None, **kwargs):
"""
Initialize XML Object
@ -145,12 +149,9 @@ class NotifyXML(NotifyBase):
xmlns:xsd="http://www.w3.org/2001/XMLSchema"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
<soapenv:Body>
<Notification xmlns:xsi="{XSD_URL}">
<Version>{XSD_VER}</Version>
<Subject>{SUBJECT}</Subject>
<MessageType>{MESSAGE_TYPE}</MessageType>
<Message>{MESSAGE}</Message>
{ATTACHMENTS}
<Notification xmlns:xsi="{{XSD_URL}}">
{{CORE}}
{{ATTACHMENTS}}
</Notification>
</soapenv:Body>
</soapenv:Envelope>"""
@ -172,6 +173,19 @@ class NotifyXML(NotifyBase):
# Store our extra headers
self.headers.update(headers)
self.payload_extras = {}
if payload:
# Store our extra payload entries (but tidy them up since they will
# become XML Keys (they can't contain certain characters
for k, v in payload.items():
key = re.sub(r'[^A-Za-z0-9_-]*', '', k)
if not key:
self.logger.warning(
'Ignoring invalid XML Stanza element name({})'
.format(k))
continue
self.payload_extras[key] = v
return
def url(self, privacy=False, *args, **kwargs):
@ -190,6 +204,10 @@ class NotifyXML(NotifyBase):
# Append our headers into our parameters
params.update({'+{}'.format(k): v for k, v in self.headers.items()})
# Append our payload extra's into our parameters
params.update(
{':{}'.format(k): v for k, v in self.payload_extras.items()})
# Determine Authentication
auth = ''
if self.user and self.password:
@ -235,7 +253,24 @@ class NotifyXML(NotifyBase):
# Our XML Attachmement subsitution
xml_attachments = ''
# Track our potential attachments
# Our Payload Base
payload_base = {
'Version': self.xsd_ver,
'Subject': NotifyXML.escape_html(title, whitespace=False),
'MessageType': NotifyXML.escape_html(
notify_type, whitespace=False),
'Message': NotifyXML.escape_html(body, whitespace=False),
}
# Apply our payload extras
payload_base.update(
{k: NotifyXML.escape_html(v, whitespace=False)
for k, v in self.payload_extras.items()})
# Base Entres
xml_base = ''.join(
['<{}>{}</{}>'.format(k, v, k) for k, v in payload_base.items()])
attachments = []
if attach:
for attachment in attach:
@ -274,13 +309,9 @@ class NotifyXML(NotifyBase):
''.join(attachments) + '</Attachments>'
re_map = {
'{XSD_VER}': self.xsd_ver,
'{XSD_URL}': self.xsd_url.format(version=self.xsd_ver),
'{MESSAGE_TYPE}': NotifyXML.escape_html(
notify_type, whitespace=False),
'{SUBJECT}': NotifyXML.escape_html(title, whitespace=False),
'{MESSAGE}': NotifyXML.escape_html(body, whitespace=False),
'{ATTACHMENTS}': xml_attachments,
'{{XSD_URL}}': self.xsd_url.format(version=self.xsd_ver),
'{{ATTACHMENTS}}': xml_attachments,
'{{CORE}}': xml_base,
}
# Iterate over above list and store content accordingly
@ -379,6 +410,10 @@ class NotifyXML(NotifyBase):
# We're done early as we couldn't load the results
return results
# store any additional payload extra's defined
results['payload'] = {NotifyXML.unquote(x): NotifyXML.unquote(y)
for x, y in results['qsd:'].items()}
# Add our headers that the user can potentially over-ride if they wish
# to to our returned result set
results['headers'] = results['qsd+']

108
test/test_custom_xml_plugin.py
View File

@ -1,108 +0,0 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2021 Chris Caron <lead2gold@gmail.com>
# All rights reserved.
#
# This code is licensed under the MIT License.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files(the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and / or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions :
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
import os
import sys
import mock
import requests
from apprise import plugins
from apprise import Apprise
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')
@mock.patch('requests.post')
def test_notify_xml_plugin_attachments(mock_post):
"""
NotifyXML() 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('xml://localhost.localdomain/')
assert isinstance(obj, plugins.NotifyXML)
# 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('xml://no-reply@example.com/')
assert isinstance(obj, plugins.NotifyXML)
# 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

3
test/test_plugin_custom_json.py
View File

@ -182,7 +182,8 @@ def test_plugin_custom_json_edge_cases(mock_get, mock_post):
dataset = json.loads(details[1]['data'])
assert dataset['title'] == 'title'
assert 'message' in dataset
assert dataset['message'] == 'body'
# message over-ride was provided
assert dataset['message'] == 'test'
assert instance.url(privacy=False).startswith(
'json://localhost:8080/command?')

94
test/test_plugin_custom_xml.py
View File

@ -22,15 +22,25 @@
# 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
import re
import mock
import requests
from apprise import plugins
from apprise import Apprise
from apprise import AppriseAttachment
from apprise import NotifyType
from helpers import AppriseURLTester
# 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 = (
('xml://:@/', {
@ -149,6 +159,74 @@ def test_plugin_custom_xml_urls():
AppriseURLTester(tests=apprise_url_tests).run_all()
@mock.patch('requests.post')
def test_notify_xml_plugin_attachments(mock_post):
"""
NotifyXML() 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('xml://localhost.localdomain/')
assert isinstance(obj, plugins.NotifyXML)
# 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('xml://no-reply@example.com/')
assert isinstance(obj, plugins.NotifyXML)
# 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
@mock.patch('requests.post')
@mock.patch('requests.get')
def test_plugin_custom_xml_edge_cases(mock_get, mock_post):
@ -168,7 +246,8 @@ def test_plugin_custom_xml_edge_cases(mock_get, mock_post):
mock_get.return_value = response
results = plugins.NotifyXML.parse_url(
'xml://localhost:8080/command?:message=test&method=GET')
'xml://localhost:8080/command?:Message=test&method=GET'
'&:Key=value&:,=invalid')
assert isinstance(results, dict)
assert results['user'] is None
@ -181,7 +260,9 @@ def test_plugin_custom_xml_edge_cases(mock_get, mock_post):
assert results['schema'] == 'xml'
assert results['url'] == 'xml://localhost:8080/command'
assert isinstance(results['qsd:'], dict) is True
assert results['qsd:']['message'] == 'test'
assert results['qsd:']['Message'] == 'test'
assert results['qsd:']['Key'] == 'value'
assert results['qsd:'][','] == 'invalid'
instance = plugins.NotifyXML(**results)
assert isinstance(instance, plugins.NotifyXML)
@ -201,3 +282,12 @@ def test_plugin_custom_xml_edge_cases(mock_get, mock_post):
for k in ('user', 'password', 'port', 'host', 'fullpath', 'path', 'query',
'schema', 'url', 'method'):
assert new_results[k] == results[k]
# Test our data set for our key/value pair
assert re.search('<Version>[1-9]+\.[0-9]+</Version>', details[1]['data'])
assert re.search('<MessageType>info</MessageType>', details[1]['data'])
assert re.search('<Subject>title</Subject>', details[1]['data'])
# Custom entry Message acts as Over-ride and kicks in here
assert re.search('<Message>test</Message>', details[1]['data'])
# Custom entry
assert re.search('<Key>value</Key>', details[1]['data'])