diff --git a/apprise/utils/parse.py b/apprise/utils/parse.py index 1d506d2ad..99fba9d8a 100644 --- a/apprise/utils/parse.py +++ b/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'))), ) diff --git a/test/test_apprise_utils.py b/test/test_apprise_utils.py index 75920d062..86cede661 100644 --- a/test/test_apprise_utils.py +++ b/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 """ diff --git a/test/test_decorator_notify.py b/test/test_decorator_notify.py index 61825042b..4328940d7 100644 --- a/test/test_decorator_notify.py +++ b/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