URLBase() supports calls to url() for generic responses (#973)

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

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({

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

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):

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

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()

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