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..77136fdc 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 @@ -16,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") @@ -26,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 @@ -42,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"}), @@ -70,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 @@ -81,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"}), @@ -94,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 @@ -106,31 +117,54 @@ def test_extension_framework_retries_non_idempotent_request_on_unauthorized(): @respx.mock -def test_extension_framework_surfaces_repeated_unauthorized(): +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"}), + 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"}), + ] + ) + + 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 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" + assert target_request.content == b"chunk-1chunk-2" + + +@respx.mock +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 @@ -138,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"}), @@ -168,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 @@ -181,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"}), @@ -194,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 @@ -208,29 +237,59 @@ async def test_extension_framework_retries_non_idempotent_request_on_unauthorize @respx.mock -async def test_extension_framework_surfaces_repeated_unauthorized_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"}), + 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"}), + ] + ) + + # 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" + + result = await async_extension_http_client.httpx_client.request( + "POST", "/orders", content=stream_body() + ) + + target_request = target_route.calls.last.request + 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" + assert target_request.content == b"chunk-1chunk-2" + + +@respx.mock +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(): @@ -242,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 @@ -268,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( @@ -278,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