From fec6de1403ebbb409349b756f022183e77866d68 Mon Sep 17 00:00:00 2001 From: Chris Caron Date: Sun, 31 Jan 2021 17:15:18 -0500 Subject: [PATCH] Added support for interpretable escapes via CLI (#349) --- apprise/Apprise.py | 40 ++++- apprise/AppriseAsset.py | 5 + apprise/cli.py | 7 +- apprise/config/ConfigBase.py | 4 +- test/test_api.py | 31 ++-- test/test_cli.py | 43 ++++++ test/test_escapes.py | 282 +++++++++++++++++++++++++++++++++++ 7 files changed, 399 insertions(+), 13 deletions(-) create mode 100644 test/test_escapes.py diff --git a/apprise/Apprise.py b/apprise/Apprise.py index 9759ef2f..ba3b8914 100644 --- a/apprise/Apprise.py +++ b/apprise/Apprise.py @@ -293,7 +293,8 @@ class Apprise(object): return def notify(self, body, title='', notify_type=NotifyType.INFO, - body_format=None, tag=MATCH_ALL_TAG, attach=None): + body_format=None, tag=MATCH_ALL_TAG, attach=None, + interpret_escapes=None): """ Send a notification to all of the plugins previously loaded. @@ -313,6 +314,9 @@ class Apprise(object): Attach can contain a list of attachment URLs. attach can also be represented by a an AttachBase() (or list of) object(s). This identifies the products you wish to notify + + Set interpret_escapes to True if you want to pre-escape a string + such as turning a \n into an actual new line, etc. """ if len(self) == 0: @@ -343,6 +347,10 @@ class Apprise(object): body_format = self.asset.body_format \ if body_format is None else body_format + # Allow Asset default value + interpret_escapes = self.asset.interpret_escapes \ + if interpret_escapes is None else interpret_escapes + # for asyncio support; we track a list of our servers to notify coroutines = [] @@ -403,6 +411,36 @@ class Apprise(object): # Store entry directly conversion_map[server.notify_format] = body + if interpret_escapes: + # + # Escape our content + # + + try: + # Added overhead requrired due to Python 3 Encoding Bug + # idenified here: https://bugs.python.org/issue21331 + conversion_map[server.notify_format] = \ + conversion_map[server.notify_format]\ + .encode('ascii', 'backslashreplace')\ + .decode('unicode-escape') + + except AttributeError: + # Must be of string type + logger.error('Failed to escape message body') + return False + + try: + # Added overhead requrired due to Python 3 Encoding Bug + # idenified here: https://bugs.python.org/issue21331 + title = title\ + .encode('ascii', 'backslashreplace')\ + .decode('unicode-escape') + + except AttributeError: + # Must be of string type + logger.error('Failed to escape message title') + return False + if ASYNCIO_SUPPORT and server.asset.async_mode: # Build a list of servers requiring notification # that will be triggered asynchronously afterwards diff --git a/apprise/AppriseAsset.py b/apprise/AppriseAsset.py index 123da722..1e085592 100644 --- a/apprise/AppriseAsset.py +++ b/apprise/AppriseAsset.py @@ -105,6 +105,11 @@ class AppriseAsset(object): # notifications are sent sequentially (one after another) async_mode = True + # Whether or not to interpret escapes found within the input text prior + # to passing it upstream. Such as converting \t to an actual tab and \n + # to a new line. + interpret_escapes = False + def __init__(self, **kwargs): """ Asset Initialization diff --git a/apprise/cli.py b/apprise/cli.py index 4c5cc517..8b6819b0 100644 --- a/apprise/cli.py +++ b/apprise/cli.py @@ -144,6 +144,8 @@ def print_version_msg(): @click.option('--verbose', '-v', count=True, help='Makes the operation more talkative. Use multiple v to ' 'increase the verbosity. I.e.: -vvvv') +@click.option('--interpret-escapes', '-e', is_flag=True, + help='Enable interpretation of backslash escapes') @click.option('--debug', '-D', is_flag=True, help='Debug mode') @click.option('--version', '-V', is_flag=True, help='Display the apprise version and exit.') @@ -151,7 +153,7 @@ def print_version_msg(): metavar='SERVER_URL [SERVER_URL2 [SERVER_URL3]]',) def main(body, title, config, attach, urls, notification_type, theme, tag, input_format, dry_run, recursion_depth, verbose, disable_async, - debug, version): + interpret_escapes, debug, version): """ Send a notification to all of the specified servers identified by their URLs the content provided within the title, body and notification-type. @@ -230,6 +232,9 @@ def main(body, title, config, attach, urls, notification_type, theme, tag, # Our body format body_format=input_format, + # Interpret Escapes + interpret_escapes=interpret_escapes, + # Set the theme theme=theme, diff --git a/apprise/config/ConfigBase.py b/apprise/config/ConfigBase.py index e828cc18..07ef6892 100644 --- a/apprise/config/ConfigBase.py +++ b/apprise/config/ConfigBase.py @@ -637,7 +637,9 @@ class ConfigBase(URLBase): # Load our data (safely) result = yaml.load(content, Loader=yaml.SafeLoader) - except (AttributeError, yaml.error.MarkedYAMLError) as e: + except (AttributeError, + yaml.parser.ParserError, + yaml.error.MarkedYAMLError) as e: # Invalid content ConfigBase.logger.error( 'Invalid Apprise YAML data specified.') diff --git a/test/test_api.py b/test/test_api.py index 1fb8eab7..bc87c20e 100644 --- a/test/test_api.py +++ b/test/test_api.py @@ -201,6 +201,13 @@ def test_apprise(): assert a.add('bad://localhost') is False assert len(a) == 0 + # We'll fail because we've got nothing to notify + assert a.notify( + title="my title", body="my body") is False + + # Clear our server listings again + a.clear() + assert a.add('good://localhost') is True assert len(a) == 1 @@ -252,9 +259,6 @@ def test_apprise(): body='body', title='test', notify_type=NotifyType.INFO, attach='invalid://') is False - # Clear our server listings again - a.clear() - class ThrowNotification(NotifyBase): def notify(self, **kwargs): # Pretend everything is okay @@ -292,14 +296,21 @@ def test_apprise(): # Store our good notification in our schema map SCHEMA_MAP['runtime'] = RuntimeNotification - assert a.add('runtime://localhost') is True - assert a.add('throw://localhost') is True - assert a.add('fail://localhost') is True - assert len(a) == 3 + for async_mode in (True, False): + # Create an Asset object + asset = AppriseAsset(theme='default', async_mode=async_mode) - # Test when our notify both throws an exception and or just - # simply returns False - assert a.notify(title="present", body="present") is False + # We can load the device using our asset + a = Apprise(asset=asset) + + assert a.add('runtime://localhost') is True + assert a.add('throw://localhost') is True + assert a.add('fail://localhost') is True + assert len(a) == 3 + + # Test when our notify both throws an exception and or just + # simply returns False + assert a.notify(title="present", body="present") is False # Create a Notification that throws an unexected exception class ThrowInstantiateNotification(NotifyBase): diff --git a/test/test_cli.py b/test/test_cli.py index 9aeeeaf6..cb85f1cd 100644 --- a/test/test_cli.py +++ b/test/test_cli.py @@ -25,6 +25,8 @@ from __future__ import print_function import re import mock +import requests +import json from os.path import dirname from os.path import join from apprise import cli @@ -112,6 +114,47 @@ def test_apprise_cli_nux_env(tmpdir): ]) assert result.exit_code == 0 + with mock.patch('requests.post') as mock_post: + # Prepare Mock + mock_post.return_value = requests.Request() + mock_post.return_value.status_code = requests.codes.ok + + result = runner.invoke(cli.main, [ + '-t', 'test title', + '-b', 'test body\\nsNewLine', + # Test using interpret escapes + '-e', + # Use our JSON query + 'json://localhost', + ]) + assert result.exit_code == 0 + + # Test our call count + assert mock_post.call_count == 1 + + # Our string is now escaped correctly + json.loads(mock_post.call_args_list[0][1]['data'])\ + .get('message', '') == 'test body\nsNewLine' + + # Reset + mock_post.reset_mock() + + result = runner.invoke(cli.main, [ + '-t', 'test title', + '-b', 'test body\\nsNewLine', + # No -e switch at all (so we don't escape the above) + # Use our JSON query + 'json://localhost', + ]) + assert result.exit_code == 0 + + # Test our call count + assert mock_post.call_count == 1 + + # Our string is now escaped correctly + json.loads(mock_post.call_args_list[0][1]['data'])\ + .get('message', '') == 'test body\\nsNewLine' + # Run in synchronous mode result = runner.invoke(cli.main, [ '-t', 'test title', diff --git a/test/test_escapes.py b/test/test_escapes.py new file mode 100644 index 00000000..850335bd --- /dev/null +++ b/test/test_escapes.py @@ -0,0 +1,282 @@ +# -*- coding: utf-8 -*- +# +# Copyright (C) 2021 Chris Caron +# 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. +from __future__ import print_function +import sys +from json import loads +import pytest +import requests +import mock +import apprise + + +@mock.patch('requests.post') +def test_apprise_interpret_escapes(mock_post): + """ + API: Apprise() interpret-escapse tests + """ + + # Prepare Mock + mock_post.return_value = requests.Request() + mock_post.return_value.status_code = requests.codes.ok + + # Default Escapes interpretation Mode is set to disable + asset = apprise.AppriseAsset() + assert asset.interpret_escapes is False + + # Load our asset + a = apprise.Apprise(asset=asset) + + # add a test server + assert a.add("json://localhost") is True + + # Our servers should carry this flag + a[0].asset.interpret_escapes is False + + # Send notification + assert a.notify("ab\\ncd") is True + + # Test our call count + assert mock_post.call_count == 1 + + # content is not escaped + loads(mock_post.call_args_list[0][1]['data'])\ + .get('message', '') == 'ab\\ncd' + + # Reset + mock_post.reset_mock() + + # Send notification and provide override: + assert a.notify("ab\\ncd", interpret_escapes=True) is True + + # Test our call count + assert mock_post.call_count == 1 + + # content IS escaped + loads(mock_post.call_args_list[0][1]['data'])\ + .get('message', '') == 'ab\ncd' + + # Reset + mock_post.reset_mock() + + # + # Now we test the reverse setup where we set the AppriseAsset + # object to True but force it off through the notify() calls + # + + # Default Escapes interpretation Mode is set to disable + asset = apprise.AppriseAsset(interpret_escapes=True) + assert asset.interpret_escapes is True + + # Load our asset + a = apprise.Apprise(asset=asset) + + # add a test server + assert a.add("json://localhost") is True + + # Our servers should carry this flag + a[0].asset.interpret_escapes is True + + # Send notification + assert a.notify("ab\\ncd") is True + + # Test our call count + assert mock_post.call_count == 1 + + # content IS escaped + loads(mock_post.call_args_list[0][1]['data'])\ + .get('message', '') == 'ab\ncd' + + # Reset + mock_post.reset_mock() + + # Send notification and provide override: + assert a.notify("ab\\ncd", interpret_escapes=False) is True + + # Test our call count + assert mock_post.call_count == 1 + + # content is NOT escaped + loads(mock_post.call_args_list[0][1]['data'])\ + .get('message', '') == 'ab\\ncd' + + +@pytest.mark.skipif(sys.version_info.major <= 2, reason="Requires Python 3.x+") +@mock.patch('requests.post') +def test_apprise_escaping_py3(mock_post): + """ + API: Apprise() Python v3.x escaping + + """ + a = apprise.Apprise() + + response = mock.Mock() + response.content = '' + response.status_code = requests.codes.ok + mock_post.return_value = response + + # Create ourselves a test object to work with + a.add('json://localhost') + + # Escape our content + assert a.notify( + title="\\r\\ntitle\\r\\n", body="\\r\\nbody\\r\\n", + interpret_escapes=True) + + # Verify our content was escaped correctly + assert mock_post.call_count == 1 + result = loads(mock_post.call_args_list[0][1]['data']) + assert result['title'] == 'title' + assert result['message'] == '\r\nbody' + + # Reset our mock object + mock_post.reset_mock() + + # + # Support Specially encoded content: + # + + # Escape our content + assert a.notify( + # Google Translated to Arabic: "Let's make the world a better place." + title='دعونا نجعل العالم مكانا أفضل.\\r\\t\\t\\n\\r\\n', + # Google Translated to Hungarian: "One line of code at a time.' + body='Egy sor kódot egyszerre.\\r\\n\\r\\r\\n', + # Our Escape Flag + interpret_escapes=True) + + # Verify our content was escaped correctly + assert mock_post.call_count == 1 + result = loads(mock_post.call_args_list[0][1]['data']) + assert result['title'] == 'دعونا نجعل العالم مكانا أفضل.' + assert result['message'] == 'Egy sor kódot egyszerre.' + + # Error handling + # + # We can't escape the content below + assert a.notify( + title=None, body=4, interpret_escapes=True) is False + assert a.notify( + title=4, body=None, interpret_escapes=True) is False + assert a.notify( + title=object(), body=False, interpret_escapes=True) is False + assert a.notify( + title=False, body=object(), interpret_escapes=True) is False + assert a.notify( + title=b'byte title', body=b'byte body', + interpret_escapes=True) is False + + # The body is proessed first, so the errors thrown above get tested on + # the body only. Now we run similar tests but only make the title + # bad and always mark the body good + assert a.notify( + title=None, body="valid", interpret_escapes=True) is False + assert a.notify( + title=4, body="valid", interpret_escapes=True) is False + assert a.notify( + title=object(), body="valid", interpret_escapes=True) is False + assert a.notify( + title=False, body="valid", interpret_escapes=True) is False + assert a.notify( + title=b'byte title', body="valid", interpret_escapes=True) is False + + +@pytest.mark.skipif(sys.version_info.major >= 3, reason="Requires Python 2.x+") +@mock.patch('requests.post') +def test_apprise_escaping_py2(mock_post): + """ + API: Apprise() Python v2.x escaping + + """ + a = apprise.Apprise() + + response = mock.Mock() + response.content = '' + response.status_code = requests.codes.ok + mock_post.return_value = response + + # Create ourselves a test object to work with + a.add('json://localhost') + + # Escape our content + assert a.notify( + title="\\r\\ntitle\\r\\n", body="\\r\\nbody\\r\\n", + interpret_escapes=True) + + # Verify our content was escaped correctly + assert mock_post.call_count == 1 + result = loads(mock_post.call_args_list[0][1]['data']) + assert result['title'] == 'title' + assert result['message'] == '\r\nbody' + + # Reset our mock object + mock_post.reset_mock() + + # + # Support Specially encoded content: + # + + # Escape our content + assert a.notify( + # Google Translated to Arabic: "Let's make the world a better place." + title='دعونا نجعل العالم مكانا أفضل.\\r\\t\\t\\n\\r\\n', + # Google Translated to Hungarian: "One line of code at a time.' + body='Egy sor kódot egyszerre.\\r\\n\\r\\r\\n', + # Our Escape Flag + interpret_escapes=True) + + # Verify our content was escaped correctly + assert mock_post.call_count == 1 + result = loads(mock_post.call_args_list[0][1]['data']) + assert result['title'] == 'دعونا نجعل العالم مكانا أفضل.' + assert result['message'] == 'Egy sor kódot egyszerre.' + + # Error handling + # + # We can't escape the content below + assert a.notify( + title=None, body=4, interpret_escapes=True) is False + assert a.notify( + title=4, body=None, interpret_escapes=True) is False + assert a.notify( + title=object(), body=False, interpret_escapes=True) is False + assert a.notify( + title=False, body=object(), interpret_escapes=True) is False + assert a.notify( + title=u'unicode title', body=u'unicode body', + interpret_escapes=True) is False + + # The body is proessed first, so the errors thrown above get tested on + # the body only. Now we run similar tests but only make the title + # bad and always mark the body good + assert a.notify( + title=None, body="valid", interpret_escapes=True) is False + assert a.notify( + title=4, body="valid", interpret_escapes=True) is False + assert a.notify( + title=object(), body="valid", interpret_escapes=True) is False + assert a.notify( + title=False, body="valid", interpret_escapes=True) is False + assert a.notify( + title=u'unicode title', body="valid", interpret_escapes=True) is False