Repository: CS-SI/eodag
Pull request #447 (2022-04-28 08:15:54)

feat: allow headers and url formatting in TokenAuth
Using TokenAuth plugin:
- allow formatting in `auth_uri`, like `http://foo.bar?username={username}&password={password}`
- allow specifying headers for token post request


github-actions[bot]: ## Unit Test Results
    2 files  ±  0      2 suites  ±0   11m 35s [:stopwatch:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v1.20/README.md#the-symbols "duration of all tests") + 1m 48s
242 tests +  6  240 [:heavy_check_mark:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v1.20/README.md#the-symbols "passed tests") +  6  2 [:zzz:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v1.20/README.md#the-symbols "skipped / disabled tests") ±0  0 [:x:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v1.20/README.md#the-symbols "failed tests") ±0 
484 runs  +12  480 [:heavy_check_mark:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v1.20/README.md#the-symbols "passed tests") +12  4 [:zzz:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v1.20/README.md#the-symbols "skipped / disabled tests") ±0  0 [:x:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v1.20/README.md#the-symbols "failed tests") ±0 

Results for commit 1e5fa009. ± Comparison against base commit 06dd23d5.

github-actions[bot]: <strong> </strong>

| File | Coverage |   |
| - | :-: | :-: |
| **All files** | `79%` | :white_check_mark: |
| eodag/plugins/authentication/token.py | `91%` | :white_check_mark: |

<strong> </strong>

| File | Coverage |   |
| - | :-: | :-: |
| **All files** | `78%` | :white_check_mark: |
| eodag/plugins/authentication/token.py | `91%` | :white_check_mark: |
| eodag/plugins/authentication/token.py | `91%` | :white_check_mark: |

_Minimum allowed coverage is `70%`_

<p align="right">Generated by :monkey: cobertura-action against 1e5fa0094d5f4355230f454125602842539979ef </p>

Patch:
diff --git a/eodag/plugins/authentication/token.py b/eodag/plugins/authentication/token.py
index 91e3c581..48d5eec3 100644
--- a/eodag/plugins/authentication/token.py
+++ b/eodag/plugins/authentication/token.py
@@ -21,16 +21,38 @@ from requests import HTTPError
 
 from eodag.plugins.authentication.base import Authentication
 from eodag.utils import RequestsTokenAuth
+from eodag.utils.exceptions import MisconfiguredError
 
 
 class TokenAuth(Authentication):
     """TokenAuth authentication plugin"""
 
+    def validate_config_credentials(self):
+        """Validate configured credentials"""
+        super(TokenAuth, self).validate_config_credentials()
+        try:
+            # format auth_uri using credentials if needed
+            self.config.auth_uri = self.config.auth_uri.format(
+                **self.config.credentials
+            )
+        except KeyError as e:
+            raise MisconfiguredError(
+                f"Missing credentials inputs for provider {self.provider}: {e}"
+            )
+
     def authenticate(self):
         """Authenticate"""
-        # First get the token
         self.validate_config_credentials()
-        response = requests.post(self.config.auth_uri, data=self.config.credentials)
+
+        # append headers to req if some are specified in config
+        req_kwargs = (
+            {"headers": self.config.headers} if hasattr(self.config, "headers") else {}
+        )
+
+        # First get the token
+        response = requests.post(
+            self.config.auth_uri, data=self.config.credentials, **req_kwargs
+        )
         try:
             response.raise_for_status()
         except HTTPError as e:
@@ -40,4 +62,5 @@ class TokenAuth(Authentication):
                 token = response.json()[self.config.token_key]
             else:
                 token = response.text
+            # Return auth class set with obtained token
             return RequestsTokenAuth(token, "header")


Test patch:
diff --git a/tests/units/test_auth_plugins.py b/tests/units/test_auth_plugins.py
new file mode 100644
index 00000000..a1ad599b
--- /dev/null
+++ b/tests/units/test_auth_plugins.py
@@ -0,0 +1,169 @@
+# -*- coding: utf-8 -*-
+# Copyright 2018, CS GROUP - France, https://www.csgroup.eu/
+#
+# This file is part of EODAG project
+#     https://www.github.com/CS-SI/EODAG
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import unittest
+from unittest import mock
+
+from requests.auth import AuthBase
+
+from eodag.config import override_config_from_mapping
+from tests.context import MisconfiguredError, PluginManager
+
+
+class BaseAuthPluginTest(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        super(BaseAuthPluginTest, cls).setUpClass()
+        cls.providers_config = {}
+        cls.plugins_manager = PluginManager(cls.providers_config)
+
+    def tearDown(self):
+        super(BaseAuthPluginTest, self).tearDown()
+        # remove credentials set during tests
+        for provider in self.providers_config:
+            self.get_auth_plugin(provider).config.__dict__.pop("credentials", None)
+
+    def get_auth_plugin(self, provider):
+        return self.plugins_manager.get_auth_plugin(provider)
+
+
+class TestAuthPluginTokenAuth(BaseAuthPluginTest):
+    @classmethod
+    def setUpClass(cls):
+        super(TestAuthPluginTokenAuth, cls).setUpClass()
+        override_config_from_mapping(
+            cls.providers_config,
+            {
+                "provider_text_token_simple_url": {
+                    "products": {},
+                    "auth": {
+                        "type": "TokenAuth",
+                        "auth_uri": "http://foo.bar",
+                    },
+                },
+                "provider_text_token_format_url": {
+                    "products": {},
+                    "auth": {
+                        "type": "TokenAuth",
+                        "auth_uri": "http://foo.bar?username={username}",
+                    },
+                },
+                "provider_text_token_header": {
+                    "products": {},
+                    "auth": {
+                        "type": "TokenAuth",
+                        "auth_uri": "http://foo.bar",
+                        "headers": {
+                            "Content-Type": "application/json;charset=UTF-8",
+                            "Accept": "application/json",
+                        },
+                    },
+                },
+                "provider_json_token_simple_url": {
+                    "products": {},
+                    "auth": {
+                        "type": "TokenAuth",
+                        "auth_uri": "http://foo.bar",
+                        "token_type": "json",
+                        "token_key": "token_is_here",
+                    },
+                },
+            },
+        )
+        cls.plugins_manager = PluginManager(cls.providers_config)
+
+    def test_plugins_auth_tokenauth_validate_credentials_empty(self):
+        """TokenAuth.validate_credentials must raise an error on empty credentials"""
+        auth_plugin = self.get_auth_plugin("provider_text_token_simple_url")
+        self.assertRaises(
+            MisconfiguredError,
+            auth_plugin.validate_config_credentials,
+        )
+
+    def test_plugins_auth_tokenauth_validate_credentials_ok(self):
+        """TokenAuth.validate_credentials must be ok on non-empty credentials"""
+        auth_plugin = self.get_auth_plugin("provider_text_token_simple_url")
+
+        auth_plugin.config.credentials = {"foo": "bar"}
+        auth_plugin.validate_config_credentials()
+
+    def test_plugins_auth_tokenauth_validate_credentials_format_url_missing(self):
+        """TokenAuth.validate_credentials must raise an error if cannot format url"""
+        auth_plugin = self.get_auth_plugin("provider_text_token_format_url")
+
+        auth_plugin.config.credentials = {"foo": "bar"}
+        self.assertRaises(
+            MisconfiguredError,
+            auth_plugin.validate_config_credentials,
+        )
+
+    def test_plugins_auth_tokenauth_validate_credentials_format_url_ok(self):
+        """TokenAuth.validate_credentials must be ok if it can format url"""
+        auth_plugin = self.get_auth_plugin("provider_text_token_format_url")
+
+        auth_plugin.config.credentials = {"foo": "bar", "username": "john"}
+        auth_plugin.validate_config_credentials()
+
+    @mock.patch("eodag.plugins.authentication.token.requests.post", autospec=True)
+    def test_plugins_auth_tokenauth_text_token_authenticate(self, mock_requests_post):
+        """TokenAuth.authenticate must return a RequestsTokenAuth object using text token"""
+        auth_plugin = self.get_auth_plugin("provider_text_token_header")
+
+        auth_plugin.config.credentials = {"foo": "bar"}
+
+        # mock token post request response
+        mock_requests_post.return_value = mock.Mock()
+        mock_requests_post.return_value.text = "this_is_test_token"
+
+        # check if returned auth object is an instance of requests.AuthBase
+        auth = auth_plugin.authenticate()
+        assert isinstance(auth, AuthBase)
+
+        # check token post request call arguments
+        args, kwargs = mock_requests_post.call_args
+        assert args[0] == auth_plugin.config.auth_uri
+        assert kwargs["data"] == auth_plugin.config.credentials
+        assert kwargs["headers"] == auth_plugin.config.headers
+
+        # check if token is integrated to the request
+        req = mock.Mock(headers={})
+        auth(req)
+        assert req.headers["Authorization"] == "Bearer this_is_test_token"
+
+    @mock.patch("eodag.plugins.authentication.token.requests.post", autospec=True)
+    def test_plugins_auth_tokenauth_json_token_authenticate(self, mock_requests_post):
+        """TokenAuth.authenticate must return a RequestsTokenAuth object using json token"""
+        auth_plugin = self.get_auth_plugin("provider_json_token_simple_url")
+
+        auth_plugin.config.credentials = {"foo": "bar"}
+
+        # mock token post request response
+        mock_requests_post.return_value = mock.Mock()
+        mock_requests_post.return_value.json.return_value = {
+            "not_token": "this_is_not_test_token",
+            "token_is_here": "this_is_test_token",
+        }
+
+        # check if returned auth object is an instance of requests.AuthBase
+        auth = auth_plugin.authenticate()
+        assert isinstance(auth, AuthBase)
+
+        # check if token is integrated to the request
+        req = mock.Mock(headers={})
+        auth(req)
+        assert req.headers["Authorization"] == "Bearer this_is_test_token"