Better support for reverse-url construction containing whitespace (#1267)

Author: caronc

apprise/utils/parse.py

@@ -920,25 +920,31 @@ def parse_emails(*args, store_unparseable=True, **kwargs):
     return result
 
 
-def url_assembly(**kwargs):
+def url_assembly(encode=False, **kwargs):
     """
     This function reverses the parse_url() function by taking in the provided
     result set and re-assembling a URL
 
     """
+    def _no_encode(content, *args, **kwargs):
+        # dummy function that does nothing to content
+        return content
+
+    _quote = quote if encode else _no_encode
+
     # Determine Authentication
     auth = ''
     if kwargs.get('user') is not None and \
             kwargs.get('password') is not None:
 
         auth = '{user}:{password}@'.format(
-            user=quote(kwargs.get('user'), safe=''),
-            password=quote(kwargs.get('password'), safe=''),
+            user=_quote(kwargs.get('user'), safe=''),
+            password=_quote(kwargs.get('password'), safe=''),
         )
 
     elif kwargs.get('user') is not None:
         auth = '{user}@'.format(
-            user=quote(kwargs.get('user'), safe=''),
+            user=_quote(kwargs.get('user'), safe=''),
         )
 
     return '{schema}://{auth}{hostname}{port}{fullpath}{params}'.format(
@@ -948,7 +954,7 @@ def url_assembly(**kwargs):
         hostname='' if not kwargs.get('host') else kwargs.get('host', ''),
         port='' if not kwargs.get('port')
         else ':{}'.format(kwargs.get('port')),
-        fullpath=quote(kwargs.get('fullpath', ''), safe='/'),
+        fullpath=_quote(kwargs.get('fullpath', ''), safe='/'),
         params='' if not kwargs.get('qsd')
         else '?{}'.format(urlencode(kwargs.get('qsd'))),
     )

test/test_apprise_utils.py

@@ -773,6 +773,27 @@ def test_parse_url_general():
     assert result['qsd']['+KeY'] == result['qsd+']['KeY']
     assert result['qsd']['-kEy'] == result['qsd-']['kEy']
 
+    # Testing Defect 1264 - whitespaces in url
+    result = utils.parse.parse_url(
+        'posts://example.com/my endpoint?-token=ab cdefg')
+
+    assert len(result['qsd-']) == 1
+    assert len(result['qsd+']) == 0
+    assert len(result['qsd']) == 1
+    assert len(result['qsd:']) == 0
+
+    assert result['schema'] == 'posts'
+    assert result['host'] == 'example.com'
+    assert result['port'] is None
+    assert result['user'] is None
+    assert result['password'] is None
+    assert result['fullpath'] == '/my%20endpoint'
+    assert result['path'] == '/'
+    assert result['query'] == "my%20endpoint"
+    assert result['url'] == 'posts://example.com/my%20endpoint'
+    assert '-token' in result['qsd']
+    assert result['qsd-']['token'] == 'ab cdefg'
+
 
 def test_parse_url_simple():
     "utils: parse_url() testing """
@@ -1181,6 +1202,38 @@ def test_url_assembly():
     assert utils.parse.url_assembly(
         **utils.parse.parse_url(url, verify_host=False)) == url
 
+    # When spaces and special characters are introduced, the URL
+    # is hard to mimic what was entered. Instead it is normalized
+    url = 'schema://hostname:10/a space/file.php?' \
+        'arg=a+space&arg2=a%20space&arg3=a space'
+    assert utils.parse.url_assembly(
+        **utils.parse.parse_url(url, verify_host=False)) == \
+        'schema://hostname:10/a%20space/file.php?' \
+        'arg=a%2Bspace&arg2=a+space&arg3=a+space'
+
+    # encode=True should only be used if you're passing in un-assembled
+    # content... hence the following is likely not what is expected:
+    assert utils.parse.url_assembly(
+        **utils.parse.parse_url(url, verify_host=False), encode=True) == \
+        'schema://hostname:10/a%2520space/file.php?' \
+        'arg=a%2Bspace&arg2=a+space&arg3=a+space'
+
+    # But the following utilizes the encode=True and produces the
+    # desired effects:
+    content = {
+        'host': 'hostname',
+        # Note that fullpath requires escaping in this case
+        'fullpath': '/a space/file.php',
+        'path': '/a space/',
+        'query': 'file.php',
+        'schema': 'schema',
+        # our query arguments also require escaping as well
+        'qsd': {'arg': 'a+space', 'arg2': 'a space', 'arg3': 'a space'},
+    }
+    assert utils.parse.url_assembly(**content, encode=True) == \
+        'schema://hostname/a%20space/file.php?' \
+        'arg=a%2Bspace&arg2=a+space&arg3=a+space'
+
 
 def test_parse_bool():
     "utils: parse_bool() testing """

test/test_decorator_notify.py

@@ -431,6 +431,81 @@ def my_inline_notify_wrapper(
     N_MGR.remove('utiltest')
 
 
+def test_notify_decorator_urls_with_space():
+    """decorators: URLs containing spaces
+    """
+    # This is in relation to https://github.com/caronc/apprise/issues/1264
+
+    # Verify our schema we're about to declare doesn't already exist
+    # in our schema map:
+    assert 'post' not in N_MGR
+
+    verify_obj = []
+
+    @notify(on="posts")
+    def apprise_custom_api_call_wrapper(
+            body, title, notify_type, attach, meta, *args, **kwargs):
+
+        # Track what is added
+        verify_obj.append({
+            'body': body,
+            'title': title,
+            'notify_type': notify_type,
+            'attach': attach,
+            'meta': meta,
+            'args': args,
+            'kwargs': kwargs,
+        })
+
+    assert 'posts' in N_MGR
+
+    # Create ourselves an apprise object
+    aobj = Apprise()
+
+    # Add our configuration
+    aobj.add("posts://example.com/my endpoint?-token=ab cdefg")
+
+    # We loaded 1 item
+    assert len(aobj) == 1
+
+    # Nothing stored yet in our object
+    assert len(verify_obj) == 0
+
+    # Send utf-8 characters
+    assert aobj.notify("ツ".encode('utf-8'), title="My Title") is True
+
+    # Service notified
+    assert len(verify_obj) == 1
+
+    # Extract our object
+    obj = verify_obj.pop()
+
+    assert obj.get('body') == 'ツ'
+    assert obj.get('title') == 'My Title'
+    assert obj.get('notify_type') == 'info'
+    assert obj.get('attach') is None
+    assert isinstance(obj.get('args'), tuple)
+    assert len(obj.get('args')) == 0
+    assert obj.get('kwargs') == {'body_format': None}
+    meta = obj.get('meta')
+    assert isinstance(meta, dict)
+
+    assert meta.get('schema') == 'posts'
+    assert meta.get('url') == \
+        'posts://example.com/my%20endpoint?-token=ab+cdefg'
+    assert meta.get('qsd') == {'-token': 'ab cdefg'}
+    assert meta.get('host') == 'example.com'
+    assert meta.get('fullpath') == '/my%20endpoint'
+    assert meta.get('path') == '/'
+    assert meta.get('query') == 'my%20endpoint'
+    assert isinstance(meta.get('tag'), set)
+    assert len(meta.get('tag')) == 0
+    assert isinstance(meta.get('asset'), AppriseAsset)
+
+    # Tidy
+    N_MGR.remove('posts')
+
+
 def test_notify_multi_instance_decoration(tmpdir):
     """decorators: Test multi-instance @notify
     """
@@ -481,7 +556,7 @@ def my_inline_notify_wrapper(
     # The number of configuration files that exist
     assert len(ac) == 1
 
-    # no notifications are loaded
+    # 2 notification endpoints are loaded
     assert len(ac.servers()) == 2
 
     # Nothing stored yet in our object