From 480d0e0bbcfe51ac622d885c9a91fc3d5a498fff Mon Sep 17 00:00:00 2001 From: Chris Caron Date: Sun, 8 Oct 2023 14:15:58 -0400 Subject: [PATCH] URLBase() supports calls to url() for generic responses (#973) --- apprise/URLBase.py | 63 ++++++++++++++++++++++++++++- apprise/plugins/NotifyAppriseAPI.py | 8 +--- test/test_api.py | 43 ++++++++++++++++++++ test/test_attach_base.py | 5 +-- test/test_notify_base.py | 11 +---- test/test_plugin_apprise_api.py | 6 +++ test/test_plugin_matrix.py | 37 +++++++++++++---- 7 files changed, 147 insertions(+), 26 deletions(-) diff --git a/apprise/URLBase.py b/apprise/URLBase.py index 998c5c46..b69ba9fc 100644 --- a/apprise/URLBase.py +++ b/apprise/URLBase.py @@ -228,6 +228,11 @@ class URLBase: # Always unquote the password if it exists self.password = URLBase.unquote(self.password) + # Store our full path consistently ensuring it ends with a `/' + self.fullpath = URLBase.unquote(kwargs.get('fullpath')) + if not isinstance(self.fullpath, str) or not self.fullpath: + self.fullpath = '/' + # Store our Timeout Variables if 'rto' in kwargs: try: @@ -307,7 +312,36 @@ class URLBase: arguments provied. """ - raise NotImplementedError("url() is implimented by the child class.") + + # Our default parameters + params = self.url_parameters(privacy=privacy, *args, **kwargs) + + # Determine Authentication + auth = '' + if self.user and self.password: + auth = '{user}:{password}@'.format( + user=URLBase.quote(self.user, safe=''), + password=self.pprint( + self.password, privacy, mode=PrivacyMode.Secret, safe=''), + ) + elif self.user: + auth = '{user}@'.format( + user=URLBase.quote(self.user, safe=''), + ) + + default_port = 443 if self.secure else 80 + + return '{schema}://{auth}{hostname}{port}{fullpath}?{params}'.format( + schema='https' if self.secure else 'http', + auth=auth, + # never encode hostname since we're expecting it to be a valid one + hostname=self.host, + port='' if self.port is None or self.port == default_port + else ':{}'.format(self.port), + fullpath=URLBase.quote(self.fullpath, safe='/') + if self.fullpath else '/', + params=URLBase.urlencode(params), + ) def __contains__(self, tags): """ @@ -583,6 +617,33 @@ class URLBase: """ return (self.socket_connect_timeout, self.socket_read_timeout) + @property + def request_auth(self): + """This is primarily used to fullfill the `auth` keyword argument + that is used by requests.get() and requests.put() calls. + """ + return (self.user, self.password) if self.user else None + + @property + def request_url(self): + """ + Assemble a simple URL that can be used by the requests library + + """ + + # Acquire our schema + schema = 'https' if self.secure else 'http' + + # Prepare our URL + url = '%s://%s' % (schema, self.host) + + # Apply Port information if present + if isinstance(self.port, int): + url += ':%d' % self.port + + # Append our full path + return url + self.fullpath + def url_parameters(self, *args, **kwargs): """ Provides a default set of args to work with. This can greatly diff --git a/apprise/plugins/NotifyAppriseAPI.py b/apprise/plugins/NotifyAppriseAPI.py index a1247c24..b18c1be3 100644 --- a/apprise/plugins/NotifyAppriseAPI.py +++ b/apprise/plugins/NotifyAppriseAPI.py @@ -167,10 +167,6 @@ class NotifyAppriseAPI(NotifyBase): """ super().__init__(**kwargs) - self.fullpath = kwargs.get('fullpath') - if not isinstance(self.fullpath, str): - self.fullpath = '/' - self.token = validate_regex( token, *self.template_tokens['token']['regex']) if not self.token: @@ -334,8 +330,8 @@ class NotifyAppriseAPI(NotifyBase): url += ':%d' % self.port fullpath = self.fullpath.strip('/') - url += '/{}/'.format(fullpath) if fullpath else '/' - url += 'notify/{}'.format(self.token) + url += '{}'.format('/' + fullpath) if fullpath else '' + url += '/notify/{}'.format(self.token) # Some entries can not be over-ridden headers.update({ diff --git a/test/test_api.py b/test/test_api.py index c37d214b..41dfdf5d 100644 --- a/test/test_api.py +++ b/test/test_api.py @@ -741,6 +741,49 @@ def test_apprise_schemas(tmpdir): assert len(schemas) == 0 +def test_apprise_urlbase_object(): + """ + API: Apprise() URLBase object testing + + """ + results = URLBase.parse_url('https://localhost/path/?cto=3.0&verify=no') + assert results.get('user') is None + assert results.get('password') is None + assert results.get('path') == '/path/' + assert results.get('secure') is True + assert results.get('verify') is False + base = URLBase(**results) + assert base.request_timeout == (3.0, 4.0) + assert base.request_auth is None + assert base.request_url == 'https://localhost/path/' + assert base.url().startswith('https://localhost/') + + results = URLBase.parse_url( + 'http://user:pass@localhost:34/path/here?rto=3.0&verify=yes') + assert results.get('user') == 'user' + assert results.get('password') == 'pass' + assert results.get('fullpath') == '/path/here' + assert results.get('secure') is False + assert results.get('verify') is True + base = URLBase(**results) + assert base.request_timeout == (4.0, 3.0) + assert base.request_auth == ('user', 'pass') + assert base.request_url == 'http://localhost:34/path/here' + assert base.url().startswith('http://user:pass@localhost:34/path/here') + + results = URLBase.parse_url('http://user@127.0.0.1/path/') + assert results.get('user') == 'user' + assert results.get('password') is None + assert results.get('fullpath') == '/path/' + assert results.get('secure') is False + assert results.get('verify') is True + base = URLBase(**results) + assert base.request_timeout == (4.0, 4.0) + assert base.request_auth == ('user', None) + assert base.request_url == 'http://127.0.0.1/path/' + assert base.url().startswith('http://user@127.0.0.1/path/') + + def test_apprise_notify_formats(tmpdir): """ API: Apprise() Input Formats tests diff --git a/test/test_attach_base.py b/test/test_attach_base.py index 340d59e7..1e4b8026 100644 --- a/test/test_attach_base.py +++ b/test/test_attach_base.py @@ -70,9 +70,8 @@ def test_attach_base(): # Create an object with no mimetype over-ride obj = AttachBase() - # Get our string object - with pytest.raises(NotImplementedError): - str(obj) + # Get our url object + str(obj) # We can not process name/path/mimetype at a Base level with pytest.raises(NotImplementedError): diff --git a/test/test_notify_base.py b/test/test_notify_base.py index 1650a519..c03722b0 100644 --- a/test/test_notify_base.py +++ b/test/test_notify_base.py @@ -65,15 +65,8 @@ def test_notify_base(): nb = NotifyBase(port=10) assert nb.port == 10 - try: - nb.url() - assert False - - except NotImplementedError: - # Each sub-module is that inherits this as a parent is required to - # over-ride this function. So direct calls to this throws a not - # implemented error intentionally - assert True + assert isinstance(nb.url(), str) + assert str(nb) == nb.url() try: nb.send('test message') diff --git a/test/test_plugin_apprise_api.py b/test/test_plugin_apprise_api.py index 3eb9cd50..2a835d53 100644 --- a/test/test_plugin_apprise_api.py +++ b/test/test_plugin_apprise_api.py @@ -265,4 +265,10 @@ def test_notify_apprise_api_attachments(mock_post): body='body', title='title', notify_type=NotifyType.INFO, attach=attach) is True assert mock_post.call_count == 1 + + details = mock_post.call_args_list[0] + assert details[0][0] == 'http://localhost/notify/mytoken1' + assert obj.url(privacy=False).startswith( + 'apprise://user@localhost/mytoken1/') + mock_post.reset_mock() diff --git a/test/test_plugin_matrix.py b/test/test_plugin_matrix.py index c5658fa7..f73210f0 100644 --- a/test/test_plugin_matrix.py +++ b/test/test_plugin_matrix.py @@ -769,6 +769,9 @@ def test_plugin_matrix_rooms(mock_post, mock_get): obj._room_cache = {} assert obj._room_id('#abc123:localhost') is None + # Force a object removal (thus a logout call) + del obj + def test_plugin_matrix_url_parsing(): """ @@ -840,6 +843,9 @@ def test_plugin_matrix_image_errors(mock_post, mock_get): # post was okay assert obj.notify('test', 'test') is True + # Force a object removal (thus a logout call) + del obj + def mock_function_handing(url, data, **kwargs): """ dummy function for handling image posts (successfully) @@ -873,6 +879,9 @@ def test_plugin_matrix_image_errors(mock_post, mock_get): assert obj.notify('test', 'test') is True + # Force a object removal (thus a logout call) + del obj + @mock.patch('requests.get') @mock.patch('requests.post') @@ -957,11 +966,14 @@ def test_plugin_matrix_attachments_api_v3(mock_post, mock_get): # handle a bad response bad_response = mock.Mock() bad_response.status_code = requests.codes.internal_server_error - mock_post.side_effect = [response, bad_response] + mock_post.side_effect = [response, bad_response, response] # We'll fail now because of an internal exception assert obj.send(body="test", attach=attach) is False + # Force a object removal (thus a logout call) + del obj + @mock.patch('requests.get') @mock.patch('requests.post') @@ -1053,15 +1065,23 @@ def test_plugin_matrix_attachments_api_v2(mock_post, mock_get): # Throw an exception on the first call to requests.post() for side_effect in (requests.RequestException(), OSError(), bad_response): - mock_post.side_effect = [side_effect] - mock_get.side_effect = [side_effect] + # Reset our value + mock_post.reset_mock() + mock_get.reset_mock() + + mock_post.side_effect = [side_effect, response] + mock_get.side_effect = [side_effect, response] assert obj.send(body="test", attach=attach) is False # Throw an exception on the second call to requests.post() for side_effect in (requests.RequestException(), OSError(), bad_response): - mock_post.side_effect = [response, side_effect, side_effect] - mock_get.side_effect = [side_effect, side_effect] + # Reset our value + mock_post.reset_mock() + mock_get.reset_mock() + + mock_post.side_effect = [response, side_effect, side_effect, response] + mock_get.side_effect = [side_effect, side_effect, response] # We'll fail now because of our error handling assert obj.send(body="test", attach=attach) is False @@ -1070,9 +1090,12 @@ def test_plugin_matrix_attachments_api_v2(mock_post, mock_get): bad_response = mock.Mock() bad_response.status_code = requests.codes.internal_server_error mock_post.side_effect = \ - [response, bad_response, response, response, response] + [response, bad_response, response, response, response, response] mock_get.side_effect = \ - [response, bad_response, response, response, response] + [response, bad_response, response, response, response, response] # We'll fail now because of an internal exception assert obj.send(body="test", attach=attach) is False + + # Force __del__() call + del obj