Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 9 additions & 2 deletions eodag/plugins/authentication/token.py
Original file line number Diff line number Diff line change
Expand Up @@ -101,6 +101,7 @@ def __init__(self, provider: str, config: PluginConfig) -> None:
self.refresh_token = ""
self.token_expiration = datetime.now()
self.auth_lock = Lock()
self._unformatted_auth_uri: Optional[str] = None

def __getstate__(self):
"""Exclude attributes that can't be pickled from serialization."""
Expand All @@ -117,9 +118,14 @@ def __setstate__(self, state):
def validate_config_credentials(self) -> None:
"""Validate configured credentials"""
super(TokenAuth, self).validate_config_credentials()

# keep the unformatted auth_uri to be able to reformat it with credentials when needed
if self._unformatted_auth_uri is None:
self._unformatted_auth_uri = self.config.auth_uri

try:
# format auth_uri using credentials if needed
self.config.auth_uri = self.config.auth_uri.format(
self.config.auth_uri = self._unformatted_auth_uri.format(
**self.config.credentials
)

Expand Down Expand Up @@ -162,7 +168,8 @@ def authenticate(self) -> AuthBase:
DEFAULT_TOKEN_EXPIRATION_MARGIN,
)
)
self.validate_config_credentials()
if self.token == "":
self.validate_config_credentials()
if (
self.token
and self.token_expiration - datetime.now() > expiration_margin
Expand Down
17 changes: 17 additions & 0 deletions tests/units/test_auth_plugins.py
Original file line number Diff line number Diff line change
Expand Up @@ -204,6 +204,20 @@ def test_plugins_auth_tokenauth_validate_credentials_format_url_ok(self):
auth_plugin.config.credentials = {"foo": "bar", "username": "john"}
auth_plugin.validate_config_credentials()

@mock.patch(
"eodag.plugins.authentication.token.requests.Session.request", autospec=True
)
def test_plugins_auth_tokenauth_authenticate_format_url_ok(self, mock_request):
"""TokenAuth.authenticate 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": "jo{hn"}
auth_plugin.authenticate()

# validate_config_credentials should have been called a 2nd time in authenticate
# to format already parsed auth_uri (would fail as credentials contain a non-escaped '{' character)
auth_plugin.authenticate()

@mock.patch(
"eodag.plugins.authentication.token.requests.Session.request", autospec=True
)
Expand All @@ -221,6 +235,9 @@ def test_plugins_auth_tokenauth_text_token_authenticate(self, mock_requests_post
auth = auth_plugin.authenticate()
self.assertTrue(isinstance(auth, AuthBase))

# token was set in the plugin
self.assertEqual(auth_plugin.token, "this_is_test_token")

# check token post request call arguments
args, kwargs = mock_requests_post.call_args
self.assertEqual(kwargs["url"], auth_plugin.config.auth_uri)
Expand Down
Loading