From 770013f0ae45e361ab1c193d96172ba29f4d0759 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Susana=20V=C3=A1zquez?= <3016283+svazquezco@users.noreply.github.com> Date: Thu, 9 Jul 2026 13:22:51 +0200 Subject: [PATCH 1/3] fix(auth): buffer streamed bodies before 401 retry in extension framework auth ExtensionFrameworkAuthentication retries requests on 401 Unauthorized, but one-shot streamed bodies (raw iterators, non-seekable files) were consumed by the first send, so the retry re-sent an empty or broken body: multipart parts backed by non-seekable files were silently truncated and raw iterator bodies raised httpx.StreamConsumed. Buffer the request body at the start of both auth flows (httpx only honours requires_request_body in its default flow wrappers, which this class overrides) so the retried request is always replayable, and add sync and async regression tests that stream a body through a 401/200 sequence and assert the retry carries the fresh token and intact body. MPT-22959 Co-Authored-By: Claude Fable 5 --- docs/usage.md | 5 +- mpt_api_client/auth/extension_framework.py | 8 +++ tests/unit/auth/test_extension_framework.py | 64 +++++++++++++++++++++ 3 files changed, 75 insertions(+), 2 deletions(-) diff --git a/docs/usage.md b/docs/usage.md index 48edc5bf..a09a2cb8 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -46,8 +46,9 @@ implementations are available: - `ExtensionFrameworkAuthentication` — a short-lived installation or account-scoped token fetched from an extension secret via `POST /installations/-/token`. It refreshes proactively once the token nears its JWT `exp` (default leeway 60s) and reactively on - `401`. Pass `account_id` to request a token scoped to a specific account - (`?account.id=`); use one provider instance per account scope. + `401`. Request bodies are buffered in memory before sending so the `401` retry can + replay one-shot streamed bodies intact. Pass `account_id` to request a token scoped to + a specific account (`?account.id=`); use one provider instance per account scope. ## Instantiate The Client diff --git a/mpt_api_client/auth/extension_framework.py b/mpt_api_client/auth/extension_framework.py index 5d52d484..e5ff88da 100644 --- a/mpt_api_client/auth/extension_framework.py +++ b/mpt_api_client/auth/extension_framework.py @@ -40,6 +40,12 @@ class ExtensionFrameworkAuthentication(Authentication): account scopes should construct one provider per scope. """ + # The 401 retry re-sends the request, so one-shot streamed bodies (raw iterators, + # non-seekable files) must be buffered before the first send; otherwise the retry + # replays an already-consumed stream. httpx only honours this flag in the default + # ``auth_flow`` wrappers, so the overridden flows below also buffer explicitly. + requires_request_body = True + def __init__( self, secret: str, @@ -76,6 +82,7 @@ def sync_auth_flow( self, request: httpx.Request ) -> Generator[httpx.Request, httpx.Response, None]: """Attach a cached token, refreshing it proactively and on 401.""" + request.read() if self._token is None or self._is_expired(): self._refresh_sync() request.headers["Authorization"] = f"Bearer {self._token}" @@ -90,6 +97,7 @@ async def async_auth_flow( self, request: httpx.Request ) -> AsyncGenerator[httpx.Request, httpx.Response]: """Attach a cached token, refreshing it proactively and on 401.""" + await request.aread() if self._token is None or self._is_expired(): await self._refresh_async() request.headers["Authorization"] = f"Bearer {self._token}" diff --git a/tests/unit/auth/test_extension_framework.py b/tests/unit/auth/test_extension_framework.py index 898e5723..eccc38be 100644 --- a/tests/unit/auth/test_extension_framework.py +++ b/tests/unit/auth/test_extension_framework.py @@ -1,6 +1,7 @@ import base64 import datetime as dt import json +from collections.abc import AsyncGenerator import httpx import pytest @@ -105,6 +106,34 @@ def test_extension_framework_retries_non_idempotent_request_on_unauthorized(): assert target_request.headers["Authorization"] == "Bearer fresh-token" +@respx.mock +def test_extension_framework_retry_resends_streamed_body(): + token_route = respx.post(TOKEN_URL).mock( + side_effect=[ + httpx.Response(200, json={"token": "stale-token"}), + httpx.Response(200, json={"token": "fresh-token"}), + ] + ) + target_route = respx.post(ORDERS_URL).mock( + side_effect=[ + httpx.Response(401, json={"error": "expired"}), + httpx.Response(201, json={"id": "ORD-1"}), + ] + ) + client = HTTPClient(base_url=API_URL, authentication=ExtensionFrameworkAuthentication(SECRET)) + + response = client.httpx_client.request( # act + "POST", "/orders", content=iter([b"chunk-1", b"chunk-2"]) + ) + + target_request = target_route.calls.last.request + assert response.status_code == httpx.codes.CREATED + assert token_route.call_count == 2 + assert target_route.call_count == 2 + assert target_request.headers["Authorization"] == "Bearer fresh-token" + assert target_request.content == b"chunk-1chunk-2" + + @respx.mock def test_extension_framework_surfaces_repeated_unauthorized(): respx.post(TOKEN_URL).mock(return_value=httpx.Response(200, json={"token": "any-token"})) @@ -207,6 +236,41 @@ async def test_extension_framework_retries_non_idempotent_request_on_unauthorize assert target_request.headers["Authorization"] == "Bearer fresh-token" +@respx.mock +async def test_extension_framework_retry_resends_streamed_body_async(): + token_route = respx.post(TOKEN_URL).mock( + side_effect=[ + httpx.Response(200, json={"token": "stale-token"}), + httpx.Response(200, json={"token": "fresh-token"}), + ] + ) + target_route = respx.post(ORDERS_URL).mock( + side_effect=[ + httpx.Response(401, json={"error": "expired"}), + httpx.Response(201, json={"id": "ORD-1"}), + ] + ) + client = AsyncHTTPClient( + base_url=API_URL, authentication=ExtensionFrameworkAuthentication(SECRET) + ) + + # httpx AsyncClient requires an async iterable for streamed content + async def stream_body() -> AsyncGenerator[bytes]: # noqa: RUF029 + yield b"chunk-1" + yield b"chunk-2" + + response = await client.httpx_client.request( # act + "POST", "/orders", content=stream_body() + ) + + target_request = target_route.calls.last.request + assert response.status_code == httpx.codes.CREATED + assert token_route.call_count == 2 + assert target_route.call_count == 2 + assert target_request.headers["Authorization"] == "Bearer fresh-token" + assert target_request.content == b"chunk-1chunk-2" + + @respx.mock async def test_extension_framework_surfaces_repeated_unauthorized_async(): respx.post(TOKEN_URL).mock(return_value=httpx.Response(200, json={"token": "any-token"})) From fed5e13d7255de38ad5cd7075b4e23eca0ac642a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Susana=20V=C3=A1zquez?= <3016283+svazquezco@users.noreply.github.com> Date: Thu, 9 Jul 2026 13:44:06 +0200 Subject: [PATCH 2/3] test(auth): use fixtures and result naming in streamed-body retry tests Address review feedback: add extension-framework client fixtures for the new regression tests and rename the act variable to result, dropping the redundant act markers. MPT-22959 Co-Authored-By: Claude Fable 5 --- tests/unit/auth/test_extension_framework.py | 28 +++++++++++++-------- 1 file changed, 18 insertions(+), 10 deletions(-) diff --git a/tests/unit/auth/test_extension_framework.py b/tests/unit/auth/test_extension_framework.py index eccc38be..5ba60c89 100644 --- a/tests/unit/auth/test_extension_framework.py +++ b/tests/unit/auth/test_extension_framework.py @@ -17,6 +17,18 @@ ORDERS_URL = f"{API_URL}/orders" +@pytest.fixture +def extension_http_client() -> HTTPClient: + return HTTPClient(base_url=API_URL, authentication=ExtensionFrameworkAuthentication(SECRET)) + + +@pytest.fixture +def async_extension_http_client() -> AsyncHTTPClient: + return AsyncHTTPClient( + base_url=API_URL, authentication=ExtensionFrameworkAuthentication(SECRET) + ) + + def _jwt_with_exp(expires_at: dt.datetime, subject: str = "token") -> str: def encode(payload: object) -> str: raw = json.dumps(payload).encode("utf-8") @@ -107,7 +119,7 @@ def test_extension_framework_retries_non_idempotent_request_on_unauthorized(): @respx.mock -def test_extension_framework_retry_resends_streamed_body(): +def test_extension_framework_retry_resends_streamed_body(extension_http_client): token_route = respx.post(TOKEN_URL).mock( side_effect=[ httpx.Response(200, json={"token": "stale-token"}), @@ -120,14 +132,13 @@ def test_extension_framework_retry_resends_streamed_body(): httpx.Response(201, json={"id": "ORD-1"}), ] ) - client = HTTPClient(base_url=API_URL, authentication=ExtensionFrameworkAuthentication(SECRET)) - response = client.httpx_client.request( # act + result = extension_http_client.httpx_client.request( "POST", "/orders", content=iter([b"chunk-1", b"chunk-2"]) ) target_request = target_route.calls.last.request - assert response.status_code == httpx.codes.CREATED + assert result.status_code == httpx.codes.CREATED assert token_route.call_count == 2 assert target_route.call_count == 2 assert target_request.headers["Authorization"] == "Bearer fresh-token" @@ -237,7 +248,7 @@ async def test_extension_framework_retries_non_idempotent_request_on_unauthorize @respx.mock -async def test_extension_framework_retry_resends_streamed_body_async(): +async def test_extension_framework_retry_resends_streamed_body_async(async_extension_http_client): token_route = respx.post(TOKEN_URL).mock( side_effect=[ httpx.Response(200, json={"token": "stale-token"}), @@ -250,21 +261,18 @@ async def test_extension_framework_retry_resends_streamed_body_async(): httpx.Response(201, json={"id": "ORD-1"}), ] ) - client = AsyncHTTPClient( - base_url=API_URL, authentication=ExtensionFrameworkAuthentication(SECRET) - ) # httpx AsyncClient requires an async iterable for streamed content async def stream_body() -> AsyncGenerator[bytes]: # noqa: RUF029 yield b"chunk-1" yield b"chunk-2" - response = await client.httpx_client.request( # act + result = await async_extension_http_client.httpx_client.request( "POST", "/orders", content=stream_body() ) target_request = target_route.calls.last.request - assert response.status_code == httpx.codes.CREATED + assert result.status_code == httpx.codes.CREATED assert token_route.call_count == 2 assert target_route.call_count == 2 assert target_request.headers["Authorization"] == "Bearer fresh-token" From d0ad4ff056f78f0ec779f4f540fece638c70874d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Susana=20V=C3=A1zquez?= <3016283+svazquezco@users.noreply.github.com> Date: Thu, 9 Jul 2026 13:48:48 +0200 Subject: [PATCH 3/3] test(auth): reuse extension client fixtures across the test module Address review feedback: migrate all tests that built the extension framework client inline to the shared extension_http_client and async_extension_http_client fixtures; only the account-scoped test keeps an inline client because it needs a different provider config. MPT-22959 Co-Authored-By: Claude Fable 5 --- tests/unit/auth/test_extension_framework.py | 96 +++++++++------------ 1 file changed, 40 insertions(+), 56 deletions(-) diff --git a/tests/unit/auth/test_extension_framework.py b/tests/unit/auth/test_extension_framework.py index 5ba60c89..77136fdc 100644 --- a/tests/unit/auth/test_extension_framework.py +++ b/tests/unit/auth/test_extension_framework.py @@ -39,14 +39,13 @@ def encode(payload: object) -> str: @respx.mock -def test_extension_framework_fetches_and_applies_token(): +def test_extension_framework_fetches_and_applies_token(extension_http_client): token_route = respx.post(TOKEN_URL).mock( return_value=httpx.Response(200, json={"token": "installation-token"}) ) target_route = respx.get(ORDERS_URL).mock(return_value=httpx.Response(200, json={"data": []})) - client = HTTPClient(base_url=API_URL, authentication=ExtensionFrameworkAuthentication(SECRET)) - client.request("GET", "/orders") # act + extension_http_client.request("GET", "/orders") # act token_request = token_route.calls.last.request target_request = target_route.calls.last.request @@ -55,22 +54,21 @@ def test_extension_framework_fetches_and_applies_token(): @respx.mock -def test_extension_framework_caches_token_across_requests(): +def test_extension_framework_caches_token_across_requests(extension_http_client): token_route = respx.post(TOKEN_URL).mock( return_value=httpx.Response(200, json={"token": "installation-token"}) ) respx.get(ORDERS_URL).mock(return_value=httpx.Response(200, json={"data": []})) - client = HTTPClient(base_url=API_URL, authentication=ExtensionFrameworkAuthentication(SECRET)) - client.request("GET", "/orders") # first call populates the cache + extension_http_client.request("GET", "/orders") # first call populates the cache - client.request("GET", "/orders") # act + extension_http_client.request("GET", "/orders") # act assert token_route.call_count == 1 @respx.mock -def test_extension_framework_refreshes_expired_token(): +def test_extension_framework_refreshes_expired_token(extension_http_client): token_route = respx.post(TOKEN_URL).mock( side_effect=[ httpx.Response(200, json={"token": "stale-token"}), @@ -83,9 +81,8 @@ def test_extension_framework_refreshes_expired_token(): httpx.Response(200, json={"data": []}), ] ) - client = HTTPClient(base_url=API_URL, authentication=ExtensionFrameworkAuthentication(SECRET)) - response = client.request("GET", "/orders") # act + response = extension_http_client.request("GET", "/orders") # act target_request = target_route.calls.last.request assert response.status_code == httpx.codes.OK @@ -94,7 +91,9 @@ def test_extension_framework_refreshes_expired_token(): @respx.mock -def test_extension_framework_retries_non_idempotent_request_on_unauthorized(): +def test_extension_framework_retries_non_idempotent_request_on_unauthorized( + extension_http_client, +): token_route = respx.post(TOKEN_URL).mock( side_effect=[ httpx.Response(200, json={"token": "stale-token"}), @@ -107,9 +106,8 @@ def test_extension_framework_retries_non_idempotent_request_on_unauthorized(): httpx.Response(201, json={"id": "ORD-1"}), ] ) - client = HTTPClient(base_url=API_URL, authentication=ExtensionFrameworkAuthentication(SECRET)) - response = client.request("POST", "/orders", json={"x": 1}) # act + response = extension_http_client.request("POST", "/orders", json={"x": 1}) # act target_request = target_route.calls.last.request assert response.status_code == httpx.codes.CREATED @@ -146,31 +144,27 @@ def test_extension_framework_retry_resends_streamed_body(extension_http_client): @respx.mock -def test_extension_framework_surfaces_repeated_unauthorized(): +def test_extension_framework_surfaces_repeated_unauthorized(extension_http_client): respx.post(TOKEN_URL).mock(return_value=httpx.Response(200, json={"token": "any-token"})) target_route = respx.get(ORDERS_URL).mock( return_value=httpx.Response(401, json={"error": "nope"}) ) - client = HTTPClient(base_url=API_URL, authentication=ExtensionFrameworkAuthentication(SECRET)) with pytest.raises(MPTAPIError) as exc_info: # act - client.request("GET", "/orders") + extension_http_client.request("GET", "/orders") assert exc_info.value.status_code == httpx.codes.UNAUTHORIZED assert target_route.call_count == 2 # original + exactly one retry, then surfaced @respx.mock -async def test_extension_framework_works_with_async_client(): +async def test_extension_framework_works_with_async_client(async_extension_http_client): token_route = respx.post(TOKEN_URL).mock( return_value=httpx.Response(200, json={"token": "installation-token"}) ) target_route = respx.get(ORDERS_URL).mock(return_value=httpx.Response(200, json={"data": []})) - client = AsyncHTTPClient( - base_url=API_URL, authentication=ExtensionFrameworkAuthentication(SECRET) - ) - await client.request("GET", "/orders") # act + await async_extension_http_client.request("GET", "/orders") # act target_request = target_route.calls.last.request assert token_route.called @@ -178,24 +172,23 @@ async def test_extension_framework_works_with_async_client(): @respx.mock -async def test_extension_framework_caches_token_across_requests_async(): +async def test_extension_framework_caches_token_across_requests_async( + async_extension_http_client, +): token_route = respx.post(TOKEN_URL).mock( return_value=httpx.Response(200, json={"token": "installation-token"}) ) respx.get(ORDERS_URL).mock(return_value=httpx.Response(200, json={"data": []})) - client = AsyncHTTPClient( - base_url=API_URL, authentication=ExtensionFrameworkAuthentication(SECRET) - ) - await client.request("GET", "/orders") # first call populates the cache + await async_extension_http_client.request("GET", "/orders") # first call populates the cache - await client.request("GET", "/orders") # act + await async_extension_http_client.request("GET", "/orders") # act assert token_route.call_count == 1 @respx.mock -async def test_extension_framework_refreshes_expired_token_async(): +async def test_extension_framework_refreshes_expired_token_async(async_extension_http_client): token_route = respx.post(TOKEN_URL).mock( side_effect=[ httpx.Response(200, json={"token": "stale-token"}), @@ -208,11 +201,8 @@ async def test_extension_framework_refreshes_expired_token_async(): httpx.Response(200, json={"data": []}), ] ) - client = AsyncHTTPClient( - base_url=API_URL, authentication=ExtensionFrameworkAuthentication(SECRET) - ) - response = await client.request("GET", "/orders") # act + response = await async_extension_http_client.request("GET", "/orders") # act target_request = target_route.calls.last.request assert response.status_code == httpx.codes.OK @@ -221,7 +211,9 @@ async def test_extension_framework_refreshes_expired_token_async(): @respx.mock -async def test_extension_framework_retries_non_idempotent_request_on_unauthorized_async(): +async def test_extension_framework_retries_non_idempotent_request_on_unauthorized_async( + async_extension_http_client, +): token_route = respx.post(TOKEN_URL).mock( side_effect=[ httpx.Response(200, json={"token": "stale-token"}), @@ -234,11 +226,8 @@ async def test_extension_framework_retries_non_idempotent_request_on_unauthorize httpx.Response(201, json={"id": "ORD-1"}), ] ) - client = AsyncHTTPClient( - base_url=API_URL, authentication=ExtensionFrameworkAuthentication(SECRET) - ) - response = await client.request("POST", "/orders", json={"x": 1}) # act + response = await async_extension_http_client.request("POST", "/orders", json={"x": 1}) # act target_request = target_route.calls.last.request assert response.status_code == httpx.codes.CREATED @@ -280,29 +269,27 @@ async def stream_body() -> AsyncGenerator[bytes]: # noqa: RUF029 @respx.mock -async def test_extension_framework_surfaces_repeated_unauthorized_async(): +async def test_extension_framework_surfaces_repeated_unauthorized_async( + async_extension_http_client, +): respx.post(TOKEN_URL).mock(return_value=httpx.Response(200, json={"token": "any-token"})) target_route = respx.get(ORDERS_URL).mock( return_value=httpx.Response(401, json={"error": "nope"}) ) - client = AsyncHTTPClient( - base_url=API_URL, authentication=ExtensionFrameworkAuthentication(SECRET) - ) with pytest.raises(MPTAPIError) as exc_info: # act - await client.request("GET", "/orders") + await async_extension_http_client.request("GET", "/orders") assert exc_info.value.status_code == httpx.codes.UNAUTHORIZED assert target_route.call_count == 2 # original + exactly one retry, then surfaced @respx.mock -def test_extension_framework_token_error(): +def test_extension_framework_token_error(extension_http_client): respx.post(TOKEN_URL).mock(return_value=httpx.Response(403, json={"error": "forbidden"})) - client = HTTPClient(base_url=API_URL, authentication=ExtensionFrameworkAuthentication(SECRET)) with pytest.raises(MPTError): # act - client.request("GET", "/orders") + extension_http_client.request("GET", "/orders") def test_extension_framework_requires_configuration(): @@ -314,12 +301,11 @@ def test_extension_framework_requires_configuration(): @respx.mock -def test_extension_framework_rejects_empty_token(): +def test_extension_framework_rejects_empty_token(extension_http_client): respx.post(TOKEN_URL).mock(return_value=httpx.Response(200, json={"token": None})) - client = HTTPClient(base_url=API_URL, authentication=ExtensionFrameworkAuthentication(SECRET)) with pytest.raises(MPTError): # act - client.request("GET", "/orders") + extension_http_client.request("GET", "/orders") @respx.mock @@ -340,7 +326,7 @@ def test_extension_framework_account_scoped_token_request(): @respx.mock -def test_extension_framework_refreshes_proactively_before_expiry(): +def test_extension_framework_refreshes_proactively_before_expiry(extension_http_client): near_expiry = dt.datetime.now(dt.UTC) + dt.timedelta(seconds=10) far_expiry = dt.datetime.now(dt.UTC) + dt.timedelta(hours=1) token_route = respx.post(TOKEN_URL).mock( @@ -350,26 +336,24 @@ def test_extension_framework_refreshes_proactively_before_expiry(): ] ) respx.get(ORDERS_URL).mock(return_value=httpx.Response(200, json={"data": []})) - client = HTTPClient(base_url=API_URL, authentication=ExtensionFrameworkAuthentication(SECRET)) - client.request("GET", "/orders") # first call caches a near-expiry token + extension_http_client.request("GET", "/orders") # first call caches a near-expiry token - client.request("GET", "/orders") # act: token within leeway -> proactive refresh + extension_http_client.request("GET", "/orders") # act: token within leeway -> refresh assert token_route.call_count == 2 @respx.mock -def test_extension_framework_reuses_unexpired_token(): +def test_extension_framework_reuses_unexpired_token(extension_http_client): far_expiry = dt.datetime.now(dt.UTC) + dt.timedelta(hours=1) token_route = respx.post(TOKEN_URL).mock( return_value=httpx.Response(200, json={"token": _jwt_with_exp(far_expiry)}) ) respx.get(ORDERS_URL).mock(return_value=httpx.Response(200, json={"data": []})) - client = HTTPClient(base_url=API_URL, authentication=ExtensionFrameworkAuthentication(SECRET)) - client.request("GET", "/orders") # caches a long-lived token + extension_http_client.request("GET", "/orders") # caches a long-lived token - client.request("GET", "/orders") # act + extension_http_client.request("GET", "/orders") # act assert token_route.call_count == 1