Fixturify _credentials_basic_handler

Author: soxofaan

openeo/rest/auth/testing.py

@@ -3,12 +3,11 @@
 """
 
 import base64
-import contextlib
+import dataclasses
 import json
 import urllib.parse
 import uuid
 from typing import List, Optional, Union
-from unittest import mock
 
 import requests
 import requests_mock.request
@@ -290,3 +289,36 @@ def get_request_history(
             for r in self.requests_mock.request_history
             if (method is None or method.lower() == r.method.lower()) and (url is None or url == r.url)
         ]
+
+
+def build_basic_auth_header(username: str, password: str) -> str:
+    """Generate basic auth header from given username and password."""
+    return "Basic " + base64.b64encode(f"{username}:{password}".encode("utf-8")).decode("utf-8")
+
+
+@dataclasses.dataclass(frozen=True)
+class SimpleBasicAuthMocker:
+    """
+    Helper to create a test fixture for simple basic auth handling in openEO context:
+    set up `/credentials/basic` handling with a fixed username/password/access_token combo.
+    """
+
+    username: str = "john"
+    password: str = "j0hn!"
+    access_token: str = "6cc3570k3n"
+
+    def expected_auth_header(self) -> str:
+        return build_basic_auth_header(username=self.username, password=self.password)
+
+    def setup_credentials_basic_handler(self, *, api_root: str, requests_mock):
+        """Set up `requests_mock` handler for `/credentials/basic` endpoint."""
+        expected_auth_header = self.expected_auth_header()
+
+        def credentials_basic_handler(request, context):
+            assert request.headers["Authorization"] == expected_auth_header
+            return json.dumps({"access_token": self.access_token})
+
+        return requests_mock.get(
+            url_join(api_root, "/credentials/basic"),
+            text=credentials_basic_handler,
+        )

tests/rest/auth/test_testing.py

@@ -3,7 +3,7 @@
     OidcClientInfo,
     OidcProviderInfo,
 )
-from openeo.rest.auth.testing import OidcMock
+from openeo.rest.auth.testing import OidcMock, build_basic_auth_header
 
 
 class TestOidcMock:
@@ -33,3 +33,7 @@ def test_request_history(self, requests_mock):
         assert [r.url for r in oidc_mock.get_request_history("/token")] == [
             "https://oidc.test/token"
         ]
+
+
+def test_build_basic_auth_header():
+    assert build_basic_auth_header("john", "56(r61!?") == "Basic am9objo1NihyNjEhPw=="

tests/rest/conftest.py

@@ -1,4 +1,6 @@
 import contextlib
+import dataclasses
+import json
 import re
 import typing
 from unittest import mock
@@ -7,7 +9,9 @@
 import time_machine
 
 from openeo.rest._testing import DummyBackend, build_capabilities
+from openeo.rest.auth.testing import SimpleBasicAuthMocker, build_basic_auth_header
 from openeo.rest.connection import Connection
+from openeo.util import url_join
 
 API_URL = "https://oeo.test/"
 
@@ -119,3 +123,14 @@ def another_dummy_backend(requests_mock) -> DummyBackend:
     another_dummy_backend.setup_file_format("GTiff")
     another_dummy_backend.setup_file_format("netCDF")
     return another_dummy_backend
+
+
+@pytest.fixture
+def basic_auth(requests_mock) -> SimpleBasicAuthMocker:
+    """
+    Fixture for simple basic auth handling in openEO context:
+    set up /credentials/basic handling with a fixed username/password/access_token combo.
+    """
+    fixture = SimpleBasicAuthMocker()
+    fixture.setup_credentials_basic_handler(api_root=API_URL, requests_mock=requests_mock)
+    return fixture

tests/rest/test_connection.py

@@ -11,7 +11,7 @@
 from pathlib import Path
 
 import pytest
-import requests.auth
+import requests
 import requests_mock
 import shapely.geometry
 
@@ -30,7 +30,7 @@
 from openeo.rest._testing import DummyBackend, build_capabilities
 from openeo.rest.auth.auth import BearerAuth, NullAuth
 from openeo.rest.auth.oidc import OidcException
-from openeo.rest.auth.testing import ABSENT, OidcMock
+from openeo.rest.auth.testing import ABSENT, OidcMock, SimpleBasicAuthMocker
 from openeo.rest.connection import (
     DEFAULT_TIMEOUT,
     DEFAULT_TIMEOUT_SYNCHRONOUS_EXECUTE,
@@ -564,10 +564,8 @@ def _get_capabilities_auth_dependent(request, context):
     return capabilities
 
 
-def test_capabilities_caching_after_authenticate_basic(requests_mock):
-    user, pwd = "john262", "J0hndo3"
+def test_capabilities_caching_after_authenticate_basic(requests_mock, basic_auth):
     get_capabilities_mock = requests_mock.get(API_URL, json=_get_capabilities_auth_dependent)
-    requests_mock.get(API_URL + 'credentials/basic', text=_credentials_basic_handler(user, pwd))
 
     con = Connection(API_URL)
     assert con.capabilities().capabilities["endpoints"] == [
@@ -578,7 +576,7 @@ def test_capabilities_caching_after_authenticate_basic(requests_mock):
     con.capabilities()
     assert get_capabilities_mock.call_count == 1
 
-    con.authenticate_basic(username=user, password=pwd)
+    con.authenticate_basic(username=basic_auth.username, password=basic_auth.password)
     assert get_capabilities_mock.call_count == 1
     assert con.capabilities().capabilities["endpoints"] == [
         {"methods": ["GET"], "path": "/credentials/basic"},
@@ -715,30 +713,17 @@ def test_api_error_non_json(requests_mock):
     assert exc.message == "olapola"
 
 
-def _credentials_basic_handler(username, password, access_token="w3lc0m3"):
-    # TODO: better reuse of this helper
-    expected_auth = requests.auth._basic_auth_str(username=username, password=password)
-
-    def handler(request, context):
-        assert request.headers["Authorization"] == expected_auth
-        return json.dumps({"access_token": access_token})
-
-    return handler
-
-
-def test_create_connection_lazy_auth_config(requests_mock, api_version):
-    user, pwd = "john262", "J0hndo3"
+def test_create_connection_lazy_auth_config(requests_mock, api_version, basic_auth):
     requests_mock.get(API_URL, json={"api_version": api_version, "endpoints": BASIC_ENDPOINTS})
-    requests_mock.get(API_URL + 'credentials/basic', text=_credentials_basic_handler(user, pwd))
 
     with mock.patch('openeo.rest.connection.AuthConfig') as AuthConfig:
         # Don't create default AuthConfig when not necessary
         conn = Connection(API_URL)
         assert AuthConfig.call_count == 0
-        conn.authenticate_basic(user, pwd)
+        conn.authenticate_basic(basic_auth.username, basic_auth.password)
         assert AuthConfig.call_count == 0
         # call `authenticate_basic` so that fallback AuthConfig is created/used lazily
-        AuthConfig.return_value.get_basic_auth.return_value = (user, pwd)
+        AuthConfig.return_value.get_basic_auth.return_value = (basic_auth.username, basic_auth.password)
         conn.authenticate_basic()
         assert AuthConfig.call_count == 1
         conn.authenticate_basic()
@@ -785,29 +770,25 @@ def test_authenticate_basic_no_support(requests_mock, api_version):
     assert isinstance(conn.auth, NullAuth)
 
 
-def test_authenticate_basic(requests_mock, api_version):
-    user, pwd = "john262", "J0hndo3"
+def test_authenticate_basic(requests_mock, api_version, basic_auth):
     requests_mock.get(API_URL, json={"api_version": api_version, "endpoints": BASIC_ENDPOINTS})
-    requests_mock.get(API_URL + 'credentials/basic', text=_credentials_basic_handler(user, pwd))
 
     conn = Connection(API_URL)
     assert isinstance(conn.auth, NullAuth)
-    conn.authenticate_basic(username=user, password=pwd)
+    conn.authenticate_basic(username=basic_auth.username, password=basic_auth.password)
     assert isinstance(conn.auth, BearerAuth)
-    assert conn.auth.bearer == "basic//w3lc0m3"
+    assert conn.auth.bearer == "basic//6cc3570k3n"
 
 
-def test_authenticate_basic_from_config(requests_mock, api_version, auth_config):
-    user, pwd = "john281", "J0hndo3"
+def test_authenticate_basic_from_config(requests_mock, api_version, auth_config, basic_auth):
     requests_mock.get(API_URL, json={"api_version": api_version, "endpoints": BASIC_ENDPOINTS})
-    requests_mock.get(API_URL + 'credentials/basic', text=_credentials_basic_handler(user, pwd))
-    auth_config.set_basic_auth(backend=API_URL, username=user, password=pwd)
+    auth_config.set_basic_auth(backend=API_URL, username=basic_auth.username, password=basic_auth.password)
 
     conn = Connection(API_URL)
     assert isinstance(conn.auth, NullAuth)
     conn.authenticate_basic()
     assert isinstance(conn.auth, BearerAuth)
-    assert conn.auth.bearer == "basic//w3lc0m3"
+    assert conn.auth.bearer == "basic//6cc3570k3n"
 
 
 @pytest.mark.slow
@@ -3966,9 +3947,10 @@ def test_connect_auto_auth_from_config_basic(
     """))
     user, pwd = "john", "j0hn"
     for u, a in [(default, "Hell0!"), (other, "Wazz6!")]:
-        auth_config.set_basic_auth(backend=u, username=user, password=pwd)
+        basic_auth_mocker = SimpleBasicAuthMocker(username=user, password=pwd, access_token=a)
+        auth_config.set_basic_auth(backend=u, username=basic_auth_mocker.username, password=basic_auth_mocker.password)
         requests_mock.get(u, json={"api_version": "1.0.0", "endpoints": BASIC_ENDPOINTS})
-        requests_mock.get(f"{u}/credentials/basic", text=_credentials_basic_handler(user, pwd, access_token=a))
+        basic_auth_mocker.setup_credentials_basic_handler(api_root=u, requests_mock=requests_mock)
 
     if use_default:
         # Without arguments: use default

tests/rest/test_job.py

@@ -13,8 +13,6 @@
 from openeo.rest.job import BatchJob, ResultAsset
 from openeo.rest.models.general import Link
 
-from .test_connection import _credentials_basic_handler
-
 API_URL = "https://oeo.test"
 
 TIFF_CONTENT = b'T1f7D6t6l0l' * 1000
@@ -719,26 +717,18 @@ def download_tiff(request, context):
         assert f.read() == TIFF_CONTENT
 
 
+
 @pytest.mark.parametrize(
     ["list_jobs_kwargs", "expected_qs"],
     [
         ({}, {}),
         ({"limit": 123}, {"limit": ["123"]}),
     ],
 )
-def test_list_jobs(con100, requests_mock, list_jobs_kwargs, expected_qs):
-    username = "john"
-    password = "j0hn!"
-    access_token = "6cc35!"
-    requests_mock.get(
-        API_URL + "/credentials/basic",
-        text=_credentials_basic_handler(
-            username=username, password=password, access_token=access_token
-        ),
-    )
+def test_list_jobs(con100, requests_mock, list_jobs_kwargs, expected_qs, basic_auth):
 
     def get_jobs(request, context):
-        assert request.headers["Authorization"] == f"Bearer basic//{access_token}"
+        assert request.headers["Authorization"] == f"Bearer basic//{basic_auth.access_token}"
         assert request.qs == expected_qs
         return {
             "jobs": [
@@ -757,26 +747,18 @@ def get_jobs(request, context):
 
     requests_mock.get(API_URL + "/jobs", json=get_jobs)
 
-    con100.authenticate_basic(username, password)
+    con100.authenticate_basic(basic_auth.username, basic_auth.password)
     jobs = con100.list_jobs(**list_jobs_kwargs)
     assert jobs == [
         {"id": "job123", "status": "running", "created": "2021-02-22T09:00:00Z"},
         {"id": "job456", "status": "created", "created": "2021-03-22T10:00:00Z"},
     ]
 
 
-def test_list_jobs_extra_metadata(con100, requests_mock, caplog):
-    # TODO: avoid this boilerplate duplication
-    username = "john"
-    password = "j0hn!"
-    access_token = "6cc35!"
-    requests_mock.get(
-        API_URL + "/credentials/basic",
-        text=_credentials_basic_handler(username=username, password=password, access_token=access_token),
-    )
+def test_list_jobs_extra_metadata(con100, requests_mock, caplog, basic_auth):
 
     def get_jobs(request, context):
-        assert request.headers["Authorization"] == f"Bearer basic//{access_token}"
+        assert request.headers["Authorization"] == f"Bearer basic//{basic_auth.access_token}"
         return {
             "jobs": [
                 {
@@ -798,7 +780,7 @@ def get_jobs(request, context):
 
     requests_mock.get(API_URL + "/jobs", json=get_jobs)
 
-    con100.authenticate_basic(username, password)
+    con100.authenticate_basic(basic_auth.username, basic_auth.password)
     jobs = con100.list_jobs()
     assert jobs == [
         {"id": "job123", "status": "running", "created": "2021-02-22T09:00:00Z"},