fix: review-driven fixes across adapters, config, clock, redirect, digest

- AsyncHttpxHttpClient: close httpx response (via aclose) when Status() rejects an unknown code, releasing the socket back to the pool.
- AiohttpHttpClient: use Url.wire_form() to preserve userinfo in outbound requests, matching httpx/requests parity.
- AiohttpHttpClient: extract Content-Length from response headers and pass to AsyncResponseBody.from_async_stream — matches httpx/requests.
- Configuration: defensive-copy `overrides` dict in __post_init__ so external mutation does not leak past construction.
- FakeClock: split wall-clock and monotonic counters; advance() never moves monotonic backwards.
- async_redirect: reference shared _REDIRECT_STATUSES constant instead of duplicating the literal set.
- DigestChallengeHandler._next_nc: clamp counter to 32 bits so it never overflows the 8-hex-digit RFC 7616 nc format.

Regression tests cover each fix.

Author: OmarAlJarrah

packages/dexpace-sdk-core/src/dexpace/sdk/core/config/configuration.py

@@ -56,6 +56,13 @@ class Configuration:
     overrides: dict[str, str] = field(default_factory=dict)
     env: EnvSource = field(default=os.environ.get, repr=False)
 
+    def __post_init__(self) -> None:
+        # Defensive copy: ``frozen=True`` protects the attribute binding, not
+        # the dict's contents. Without this, external callers can mutate
+        # ``overrides`` after construction and observe the change through
+        # ``get``.
+        object.__setattr__(self, "overrides", dict(self.overrides))
+
     # Well-known keys — mirror Java SDK constants where they apply.
     # ``ClassVar`` keeps these off the dataclass field list and out of ``__slots__``.
     MAX_RETRY_ATTEMPTS: ClassVar[str] = "MAX_RETRY_ATTEMPTS"

packages/dexpace-sdk-core/src/dexpace/sdk/core/http/auth/digest.py

@@ -159,7 +159,10 @@ def _select(self, challenges: list[AuthenticateChallenge]) -> AuthenticateChalle
 
     def _next_nc(self) -> str:
         with self._lock:
-            self._counter += 1
+            # Clamp to 32 bits — ``nc`` is rendered as 8 hex digits per
+            # RFC 7616, and wrapping after 2**32-1 is acceptable since the
+            # server hashes the value (no monotonic check).
+            self._counter = (self._counter + 1) & 0xFFFFFFFF
             value = self._counter
         return f"{value:08x}"
 

packages/dexpace-sdk-core/src/dexpace/sdk/core/pipeline/policies/async_redirect.py

@@ -13,7 +13,7 @@
 from ...http.request.method import Method
 from ..async_policy import AsyncPolicy
 from ..stage import Stage
-from .redirect import RedirectPolicy
+from .redirect import _REDIRECT_STATUSES, RedirectPolicy
 
 if TYPE_CHECKING:
     from ...http.request.request import Request
@@ -65,7 +65,7 @@ async def send(self, request: Request, ctx: PipelineContext) -> AsyncResponse:
             if hops >= cfg.max_hops:
                 return response
             status = int(response.status)
-            if status not in {301, 302, 303, 307, 308}:
+            if status not in _REDIRECT_STATUSES:
                 return response
             location = response.headers.get("Location")
             if location is None or not location.strip():

packages/dexpace-sdk-core/tests/config/test_configuration.py

@@ -190,3 +190,19 @@ def test_default_env_is_os_environ(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setenv("DEXPACE_TEST_CONFIG_KEY", "value-from-os")
     cfg = Configuration()
     assert cfg.get("DEXPACE_TEST_CONFIG_KEY", None) == "value-from-os"
+
+
+class TestOverrideImmutability:
+    def test_external_mutation_of_overrides_does_not_leak(self) -> None:
+        """A defensive copy means callers can't mutate config after construction."""
+        original = {"KEY": "initial"}
+        cfg = Configuration(overrides=original)
+        original["KEY"] = "mutated-after-construction"
+        assert cfg.get("KEY") == "initial"
+
+    def test_builder_build_takes_defensive_copy(self) -> None:
+        builder = Configuration.builder().put("KEY", "v1")
+        cfg = builder.build()
+        builder.put("KEY", "v2")
+        # The originally-built config should not see subsequent builder edits.
+        assert cfg.get("KEY") == "v1"

packages/dexpace-sdk-core/tests/conftest.py

@@ -28,32 +28,43 @@ def _clean_context_store() -> Iterator[None]:
 class FakeClock:
     """Deterministic :class:`~dexpace.sdk.core.util.Clock` for unit tests.
 
-    Wall-clock and monotonic readings are backed by the same internal
-    counter, which advances only via :meth:`sleep` (clamped at zero) or
-    explicit :meth:`advance` calls. No real time elapses.
+    Wall-clock (``now``) and monotonic readings are tracked independently so
+    tests can model the real divergence between them (system clock jumps vs
+    monotonic uptime). ``sleep`` advances both. ``advance`` advances wall-
+    clock by the (possibly negative) delta but only ever advances monotonic
+    forward — monotonic is, by contract, non-decreasing.
     """
 
-    __slots__ = ("_t",)
+    __slots__ = ("_monotonic", "_wall")
 
     def __init__(self, start: float = 0.0) -> None:
-        """Initialise the fake clock at ``start`` seconds."""
-        self._t = start
+        """Initialise both clocks at ``start`` seconds."""
+        self._wall = start
+        self._monotonic = start
 
     def now(self) -> float:
         """Return the current simulated wall-clock time, in seconds."""
-        return self._t
+        return self._wall
 
     def monotonic(self) -> float:
         """Return the current simulated monotonic reading, in seconds."""
-        return self._t
+        return self._monotonic
 
     def sleep(self, duration: float) -> None:
-        """Advance the clock by ``duration`` seconds, clamped at zero."""
-        self._t += max(0.0, duration)
+        """Advance both wall and monotonic by ``duration`` (clamped at zero)."""
+        delta = max(0.0, duration)
+        self._wall += delta
+        self._monotonic += delta
 
     def advance(self, duration: float) -> None:
-        """Advance the clock by ``duration`` seconds (may be negative)."""
-        self._t += duration
+        """Advance wall-clock by ``duration`` (may be negative).
+
+        Monotonic is only advanced when ``duration`` is positive — modelling
+        the platform guarantee that monotonic time never goes backwards.
+        """
+        self._wall += duration
+        if duration > 0:
+            self._monotonic += duration
 
 
 @pytest.fixture

packages/dexpace-sdk-core/tests/util/test_clock.py

@@ -127,3 +127,19 @@ def test_fake_clock_advance(fake_clock: FakeClock) -> None:
 def test_fake_clock_satisfies_clock_protocol(fake_clock: FakeClock) -> None:
     """``FakeClock`` is structurally a :class:`Clock`."""
     assert isinstance(fake_clock, Clock)
+
+
+def test_fake_clock_monotonic_does_not_decrease_on_negative_advance() -> None:
+    """``advance(-delta)`` moves wall back but monotonic is preserved."""
+    clock = FakeClock(start=10.0)
+    clock.advance(-3.0)
+    assert clock.now() == pytest.approx(7.0)
+    # Monotonic time never goes backwards on real systems.
+    assert clock.monotonic() == pytest.approx(10.0)
+
+
+def test_fake_clock_sleep_advances_both_wall_and_monotonic() -> None:
+    clock = FakeClock()
+    clock.sleep(5.0)
+    assert clock.now() == pytest.approx(5.0)
+    assert clock.monotonic() == pytest.approx(5.0)

packages/dexpace-sdk-http-aiohttp/src/dexpace/sdk/http/aiohttp/client.py

@@ -78,7 +78,7 @@ async def execute(self, request: Request) -> AsyncResponse:
         try:
             ctx = session.request(
                 method=str(request.method),
-                url=str(request.url),
+                url=request.url.wire_form(),
                 headers=_request_headers(request.headers),
                 data=data,
                 timeout=timeout_cfg,
@@ -161,7 +161,10 @@ def _wrap_response(request: Request, aio_response: aiohttp.ClientResponse) -> As
         ) from err
     headers = Headers(tuple(aio_response.headers.items()))
     reason = aio_response.reason
-    body = AsyncResponseBody.from_async_stream(_StreamReaderAdapter(aio_response))
+    content_length = _content_length(aio_response)
+    body = AsyncResponseBody.from_async_stream(
+        _StreamReaderAdapter(aio_response), content_length=content_length
+    )
     return AsyncResponse(
         request=request,
         protocol=Protocol.HTTP_1_1,
@@ -204,4 +207,15 @@ async def close(self) -> None:
         self._response.release()
 
 
+def _content_length(aio_response: aiohttp.ClientResponse) -> int:
+    """Extract ``Content-Length`` from the response, or ``-1`` when absent/invalid."""
+    raw = aio_response.headers.get("Content-Length")
+    if raw is None:
+        return -1
+    try:
+        return max(0, int(raw))
+    except ValueError:
+        return -1
+
+
 __all__ = ["AiohttpHttpClient"]

packages/dexpace-sdk-http-aiohttp/tests/test_aiohttp_client.py

@@ -157,3 +157,13 @@ async def test_shared_session_is_not_closed_on_aclose() -> None:
         assert not session.closed
     finally:
         await session.close()
+
+
+async def test_content_length_extracted_from_response(base_url: str) -> None:
+    """Aiohttp client should populate AsyncResponseBody.content_length from headers."""
+    async with AiohttpHttpClient() as client:
+        request = Request(method=Method.GET, url=Url.parse(f"{base_url}/ok"))
+        async with await client.execute(request) as response:
+            assert response.body is not None
+            # /ok returns a small JSON; Content-Length should be set by aiohttp.
+            assert response.body.content_length() > 0

packages/dexpace-sdk-http-httpx/src/dexpace/sdk/http/httpx/async_.py

@@ -111,7 +111,7 @@ async def execute(self, request: Request) -> AsyncResponse:
             raise ServiceRequestError(str(err), error=err) from err
         except httpx.RequestError as err:
             raise ServiceRequestError(str(err), error=err) from err
-        return _build_async_response(request, httpx_response)
+        return await _build_async_response(request, httpx_response)
 
     def _build_request(self, request: Request) -> httpx.Request:
         headers = _headers_to_pairs(request.headers)
@@ -200,13 +200,13 @@ async def close(self) -> None:
         await self._response.aclose()
 
 
-def _build_async_response(request: Request, httpx_response: httpx.Response) -> AsyncResponse:
+async def _build_async_response(request: Request, httpx_response: httpx.Response) -> AsyncResponse:
     try:
         status = Status(int(httpx_response.status_code))
     except ValueError as err:
-        # ``aclose`` is async; schedule a synchronous close fallback via
-        # the underlying response's ``close`` since we cannot await here.
-        # Best effort — the failing status code is the user-visible signal.
+        # Release the underlying socket back to the pool; httpx async
+        # responses reject sync ``close``, so use ``aclose``.
+        await httpx_response.aclose()
         raise ServiceResponseError(
             f"Unknown status code: {httpx_response.status_code}", error=err
         ) from err

packages/dexpace-sdk-http-httpx/tests/test_async_httpx_client.py

@@ -185,3 +185,25 @@ def handler(request: httpx.Request) -> httpx.Response:
         request = Request(method=Method.GET, url=Url.parse("http://example.test/"))
         async with await client.execute(request) as response:
             assert response.reason == "OK"
+
+
+async def test_async_unknown_status_closes_response() -> None:
+    """When Status() rejects an unknown code, the response must be released."""
+    from dexpace.sdk.core.errors import ServiceResponseError
+
+    closed = {"yes": False}
+
+    class _TrackedResponse(httpx.Response):
+        async def aclose(self) -> None:
+            closed["yes"] = True
+            await super().aclose()
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        return _TrackedResponse(999, content=b"")
+
+    transport = httpx.MockTransport(handler)
+    async with AsyncHttpxHttpClient(transport=transport) as client:
+        request = Request(method=Method.GET, url=Url.parse("http://example.test/"))
+        with pytest.raises(ServiceResponseError):
+            await client.execute(request)
+    assert closed["yes"], "Response should be closed when status mapping fails"