[AT-4915] Settings detection based on credentials (#570)

* Settings detection based on credentials

* Version and CHANGELOG.md

* Address comments

* Apply hooks

CHANGELOG.md

@@ -29,6 +29,11 @@ You can check your current version with the following command:
     ```
 
 For more information, see [UP42 Python package description](https://pypi.org/project/up42-py/).
+## 0.37.1a2
+
+**March 19, 2024**
+
+- Detection of credentials settings based on supplied credentials.
 
 ## 0.37.1a1
 

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "up42-py"
-version = "0.37.1a1"
+version = "0.37.1a2"
 description = "Python SDK for UP42, the geospatial marketplace and developer platform."
 authors = ["UP42 GmbH <[email protected]>"]
 license = "https://github.com/up42/up42-py/blob/master/LICENSE"

tests/http/test_oauth.py

@@ -1,7 +1,10 @@
 import base64
+import dataclasses
 import time
+from typing import Tuple
 
 import mock
+import pytest
 import requests
 import requests_mock as req_mock
 
@@ -11,8 +14,8 @@
 TOKEN_VALUE = "some-token"
 TOKEN_URL = "https://localhost/oauth/token"
 project_credentials = config.ProjectCredentialsSettings(
-    client_id="client_id",
-    client_secret="client_secret",
+    project_id="some-id",
+    project_api_key="some-key",
 )
 account_credentials = config.AccountCredentialsSettings(username="some-user", password="some-pass")
 
@@ -22,7 +25,7 @@ def basic_auth(username, password):
     return f'Basic {token.decode("ascii")}'
 
 
-basic_client_auth = basic_auth(project_credentials.client_id, project_credentials.client_secret)
+basic_client_auth = basic_auth(project_credentials.project_id, project_credentials.project_api_key)
 basic_auth_headers = {"Authorization": basic_client_auth}
 
 
@@ -94,3 +97,26 @@ def test_should_fetch_token_when_expired(self):
         assert up42_auth.token.access_token == second_token
         assert TOKEN_URL, HTTP_TIMEOUT == retrieve.call_args.args[1:]
         assert retrieve.call_count == 2
+
+
+class TestDetectSettings:
+    def test_should_detect_project_credentials(self):
+        assert oauth.detect_settings(dataclasses.asdict(project_credentials)) == project_credentials
+
+    def test_should_detect_account_credentials(self):
+        assert oauth.detect_settings(dataclasses.asdict(account_credentials)) == account_credentials
+
+    @pytest.mark.parametrize("keys", [("project_id", "project_api_key"), ("username", "password")])
+    def test_should_accept_empty_credentials(self, keys: Tuple[str, str]):
+        credentials = dict(zip(keys, [None] * 2))
+        assert not oauth.detect_settings(credentials)
+
+    def test_fails_if_credentials_are_incomplete(self):
+        credentials = {"key1": "value1", "key2": None}
+        with pytest.raises(oauth.IncompleteCredentials):
+            oauth.detect_settings(credentials)
+
+    def test_fails_if_credentials_are_invalid(self):
+        credentials = {"key1": "value1", "key2": "value2"}
+        with pytest.raises(oauth.InvalidCredentials):
+            oauth.detect_settings(credentials)

up42/http/config.py

@@ -1,4 +1,5 @@
 import dataclasses as dc
+from typing import Union
 
 
 @dc.dataclass(eq=True, frozen=True)
@@ -17,11 +18,14 @@ class TokenProviderSettings:
 
 @dc.dataclass(eq=True, frozen=True)
 class ProjectCredentialsSettings:
-    client_id: str
-    client_secret: str
+    project_id: str
+    project_api_key: str
 
 
 @dc.dataclass(eq=True, frozen=True)
 class AccountCredentialsSettings:
     username: str
     password: str
+
+
+CredentialsSettings = Union[AccountCredentialsSettings, ProjectCredentialsSettings]

up42/http/oauth.py

@@ -1,6 +1,7 @@
 import dataclasses as dc
 import datetime as dt
-from typing import Protocol
+import warnings
+from typing import Optional, Protocol
 
 import requests
 from requests import auth
@@ -26,8 +27,8 @@ def __call__(self, session: requests.Session, token_url: str, timeout: int) -> s
 class ProjectTokenRetriever:
     def __init__(self, supply_credentials_settings=config.ProjectCredentialsSettings):
         credentials_settings = supply_credentials_settings()
-        self.client_id = credentials_settings.client_id
-        self.client_secret = credentials_settings.client_secret
+        self.client_id = credentials_settings.project_id
+        self.client_secret = credentials_settings.project_api_key
 
     def __call__(self, session: requests.Session, token_url: str, timeout: int) -> str:
         basic_auth = auth.HTTPBasicAuth(self.client_id, self.client_secret)
@@ -93,3 +94,28 @@ def token(self) -> Token:
         if self._token.has_expired:
             self._token = self._fetch_token()
         return self._token
+
+
+def detect_settings(credentials: dict) -> Optional[config.CredentialsSettings]:
+    if all(credentials.values()):
+        keys = credentials.keys()
+        if keys == {"project_id", "project_api_key"}:
+            warnings.warn(
+                "Project based authentication will be deprecated."
+                "Please follow authentication guidelines (/docs/authentication.md)."
+            )
+            return config.ProjectCredentialsSettings(**credentials)
+        if keys == {"username", "password"}:
+            return config.AccountCredentialsSettings(**credentials)
+        raise InvalidCredentials
+    elif any(credentials.values()):
+        raise IncompleteCredentials
+    return None
+
+
+class InvalidCredentials(ValueError):
+    pass
+
+
+class IncompleteCredentials(ValueError):
+    pass