Added support for interpretable escapes via CLI (#349)

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

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

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,
 

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.')

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
+    for async_mode in (True, False):
-    assert a.add('throw://localhost') is True
+        # Create an Asset object
-    assert a.add('fail://localhost') is True
+        asset = AppriseAsset(theme='default', async_mode=async_mode)
-    assert len(a) == 3
 
-    # Test when our notify both throws an exception and or just
+        # We can load the device using our asset
-    # simply returns False
+        a = Apprise(asset=asset)
-    assert a.notify(title="present", body="present") is False
+
+        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):

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',

test/test_escapes.py

@@ -0,0 +1,282 @@
+# -*- 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.
+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