diff --git a/.changelog/httpx-client-adapters.md b/.changelog/httpx-client-adapters.md new file mode 100644 index 0000000..8fef834 --- /dev/null +++ b/.changelog/httpx-client-adapters.md @@ -0,0 +1,5 @@ +--- +pympp: minor +--- + +Added fail-fast per-client adapters for existing HTTPX 0.27.x and 0.28.x clients. diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 495c877..7515064 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -17,7 +17,7 @@ jobs: ci-gate: name: CI Gate if: always() - needs: [test, integration, package, install-smoke] + needs: [test, httpx-compat, integration, package, install-smoke] runs-on: ubuntu-latest steps: - env: @@ -90,6 +90,38 @@ jobs: htmlcov/ if-no-files-found: ignore + httpx-compat: + name: HTTPX ${{ matrix.httpx-version }} + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + httpx-version: ["0.27.0", "0.28.1"] + + steps: + - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 + with: + persist-credentials: false + + - name: Install uv + uses: astral-sh/setup-uv@fac544c07dec837d0ccb6301d7b5580bf5edae39 # v8.2.0 + with: + enable-cache: true + cache-dependency-glob: "pyproject.toml" + + - name: Set up Python + run: uv python install 3.12 + + - name: Install dependencies + run: | + uv venv --python 3.12 + uv pip install --python .venv/bin/python -e . "httpx==$HTTPX_VERSION" pytest pytest-asyncio + env: + HTTPX_VERSION: ${{ matrix.httpx-version }} + + - name: Run HTTPX adapter tests + run: .venv/bin/python -m pytest --noconftest -q tests/test_httpx_adapters.py + integration: name: Integration Test runs-on: ubuntu-latest diff --git a/README.md b/README.md index a7d2ac6..36006bc 100644 --- a/README.md +++ b/README.md @@ -94,6 +94,20 @@ with OwnedPaymentRuntime([method]) as runtime: For loop-bound resources, pass async-context-manager factories with `method_factories=` so they are created and closed on the owned loop. +Existing HTTPX clients can be made payment-aware without replacing their +configured transports: + +```python +with OwnedPaymentRuntime([method]) as runtime: + with runtime.wrap_client(httpx.Client()) as client: + response = client.get("https://api.example.com/paid") +``` + +Use `runtime.wrap_async_client(httpx.AsyncClient())` for asynchronous clients. +These adapters support HTTPX 0.27.x and 0.28.x and fail before changing a client +when its version or private send seam is incompatible. Explicit `PaymentTransport` +and `SyncPaymentTransport` integrations do not depend on those private seams. + ## Examples | Example | Description | diff --git a/src/mpp/_httpx.py b/src/mpp/_httpx.py new file mode 100644 index 0000000..70feafa --- /dev/null +++ b/src/mpp/_httpx.py @@ -0,0 +1,275 @@ +"""Fail-fast adapters for existing HTTPX clients.""" + +from __future__ import annotations + +import inspect +import threading +from contextlib import contextmanager +from contextvars import ContextVar +from dataclasses import dataclass +from functools import wraps +from importlib.metadata import version +from typing import TYPE_CHECKING, Any + +import httpx + +if TYPE_CHECKING: + from collections.abc import Awaitable, Callable + + from mpp.runtime import OwnedPaymentRuntime + +HTTPX_ADAPTER_VERSIONS = ">=0.27,<0.29" +_SUPPORTED_HTTPX_MINORS = {(0, 27), (0, 28)} +_MARKER = "_mpp_httpx_adapter" +_MISSING = object() +_LOCK = threading.RLock() +_POSITIONAL = inspect.Parameter.POSITIONAL_OR_KEYWORD +_KEYWORD_ONLY = inspect.Parameter.KEYWORD_ONLY +_PRIVATE = (("self", _POSITIONAL), ("request", _POSITIONAL)) +_BOUND_PRIVATE = _PRIVATE[1:] +_PUBLIC = ( + ("self", _POSITIONAL), + ("request", _POSITIONAL), + ("stream", _KEYWORD_ONLY), + ("auth", _KEYWORD_ONLY), + ("follow_redirects", _KEYWORD_ONLY), +) +_BOUND_PUBLIC = _PUBLIC[1:] + + +class HttpxCompatibilityError(RuntimeError): + """The installed HTTPX client cannot be safely adapted.""" + + +@dataclass(frozen=True, slots=True) +class _Adapter: + runtime: OwnedPaymentRuntime + private: Any + public: Any + + +@dataclass(slots=True) +class _SendOperation: + paid: bool = False + + +_SEND_OPERATION: ContextVar[_SendOperation | None] = ContextVar( + "mpp_httpx_send_operation", + default=None, +) + + +@contextmanager +def _send_operation(): + token = _SEND_OPERATION.set(_SendOperation()) + try: + yield + finally: + _SEND_OPERATION.reset(token) + + +@contextmanager +def _payment_budget(request: httpx.Request): + from mpp.client._http import _PAYMENT_SENT + + operation = _SEND_OPERATION.get() + if operation is not None: + source = request.extensions.get(_PAYMENT_SENT) + if operation.paid: + request.extensions[_PAYMENT_SENT] = 0 + elif isinstance(source, int): + request.extensions.pop(_PAYMENT_SENT, None) + try: + yield + finally: + if operation is not None and isinstance(request.extensions.get(_PAYMENT_SENT), int): + operation.paid = True + + +class _SyncSend(httpx.BaseTransport): + def __init__(self, send: Callable[[httpx.Request], httpx.Response]) -> None: + self._send = send + + def handle_request(self, request: httpx.Request) -> httpx.Response: + return self._send(request) + + +class _AsyncSend(httpx.AsyncBaseTransport): + def __init__( + self, + send: Callable[[httpx.Request], Awaitable[httpx.Response]], + ) -> None: + self._send = send + + async def handle_async_request(self, request: httpx.Request) -> httpx.Response: + return await self._send(request) + + +def wrap_client(client: httpx.Client, runtime: OwnedPaymentRuntime) -> httpx.Client: + """Make one synchronous HTTPX client payment-aware.""" + if not isinstance(client, httpx.Client): + raise TypeError("wrap_client requires an httpx.Client") + return _wrap(client, runtime, asynchronous=False) + + +def wrap_async_client( + client: httpx.AsyncClient, + runtime: OwnedPaymentRuntime, +) -> httpx.AsyncClient: + """Make one asynchronous HTTPX client payment-aware.""" + if not isinstance(client, httpx.AsyncClient): + raise TypeError("wrap_async_client requires an httpx.AsyncClient") + return _wrap(client, runtime, asynchronous=True) + + +def _wrap(client: Any, runtime: OwnedPaymentRuntime, *, asynchronous: bool) -> Any: + with _LOCK: + existing = client.__dict__.get(_MARKER, _MISSING) + if existing is not _MISSING: + if ( + not isinstance(existing, _Adapter) + or inspect.getattr_static(client, "_send_single_request") is not existing.private + or inspect.getattr_static(client, "send") is not existing.public + ): + raise HttpxCompatibilityError("HTTPX client adapter was replaced or corrupted") + if existing.runtime is not runtime: + raise RuntimeError("HTTPX client is already bound to another payment runtime") + return client + + original_private, original_public = _validate_client( + client, + asynchronous=asynchronous, + ) + if asynchronous: + from mpp.client import PaymentTransport + + transport = PaymentTransport(inner=_AsyncSend(original_private), runtime=runtime) + + @wraps(original_private) + async def async_private(request: httpx.Request) -> httpx.Response: + with _payment_budget(request): + return await transport.handle_async_request(request) + + @wraps(original_public) + async def async_public( + request: httpx.Request, + *args: Any, + **kwargs: Any, + ) -> httpx.Response: + with _send_operation(): + return await original_public(request, *args, **kwargs) + + private, public = async_private, async_public + else: + from mpp.client import SyncPaymentTransport + + transport = SyncPaymentTransport(inner=_SyncSend(original_private), runtime=runtime) + + @wraps(original_private) + def sync_private(request: httpx.Request) -> httpx.Response: + with _payment_budget(request): + return transport.handle_request(request) + + @wraps(original_public) + def sync_public( + request: httpx.Request, + *args: Any, + **kwargs: Any, + ) -> httpx.Response: + with _send_operation(): + return original_public(request, *args, **kwargs) + + private, public = sync_private, sync_public + + _set_attributes( + client, + _send_single_request=private, + send=public, + **{_MARKER: _Adapter(runtime, private, public)}, + ) + return client + + +def _validate_client(client: Any, *, asynchronous: bool) -> tuple[Any, Any]: + _validate_version() + owner = type(client) + bound = [] + for name, expected, bound_expected in ( + ("_send_single_request", _PRIVATE, _BOUND_PRIVATE), + ("send", _PUBLIC, _BOUND_PUBLIC), + ): + try: + seam = inspect.getattr_static(owner, name) + except AttributeError as error: + raise HttpxCompatibilityError( + f"HTTPX adapter seam {owner.__name__}.{name} is missing" + ) from error + _validate_seam(f"{owner.__name__}.{name}", seam, expected, asynchronous) + value = getattr(client, name) + _validate_seam( + f"{owner.__name__} instance.{name}", + value, + bound_expected, + asynchronous, + ) + bound.append(value) + return bound[0], bound[1] + + +def _validate_version() -> None: + installed = version("httpx") + try: + supported = tuple(map(int, installed.split(".")[:2])) in _SUPPORTED_HTTPX_MINORS + except ValueError as error: + raise HttpxCompatibilityError( + f"Cannot determine HTTPX compatibility from version {installed!r}" + ) from error + if not supported: + raise HttpxCompatibilityError( + f"HTTPX {installed} is unsupported by pympp HTTPX adapters " + f"(supported: {HTTPX_ADAPTER_VERSIONS}). " + "Use PaymentTransport or SyncPaymentTransport explicitly, or upgrade pympp." + ) + + +def _validate_seam( + name: str, + seam: Any, + expected: tuple[tuple[str, inspect._ParameterKind], ...], + asynchronous: bool, +) -> None: + try: + shape = tuple( + (parameter.name, parameter.kind) + for parameter in inspect.signature(seam).parameters.values() + ) + except (TypeError, ValueError) as error: + raise HttpxCompatibilityError(f"HTTPX adapter seam {name} is not inspectable") from error + if not callable(seam) or shape != expected: + raise HttpxCompatibilityError(f"HTTPX adapter seam {name} has an unsupported signature") + if inspect.iscoroutinefunction(seam) is not asynchronous: + kind = "async" if asynchronous else "sync" + raise HttpxCompatibilityError(f"HTTPX adapter seam {name} is not {kind}") + + +def _set_attributes(client: Any, **values: Any) -> None: + previous = {name: client.__dict__.get(name, _MISSING) for name in values} + changed: list[str] = [] + try: + for name, value in values.items(): + changed.append(name) + setattr(client, name, value) + except BaseException as error: + rollback_error: BaseException | None = None + for name in reversed(changed): + try: + if previous[name] is _MISSING: + if name in client.__dict__: + delattr(client, name) + else: + setattr(client, name, previous[name]) + except BaseException as cause: + rollback_error = rollback_error or cause + if rollback_error is not None: + raise RuntimeError("Failed to roll back HTTPX client adapter") from error + raise diff --git a/src/mpp/runtime.py b/src/mpp/runtime.py index be193c6..19fabfd 100644 --- a/src/mpp/runtime.py +++ b/src/mpp/runtime.py @@ -22,6 +22,8 @@ from anyio.from_thread import BlockingPortal, start_blocking_portal from mpp import Challenge, Credential +from mpp._httpx import HTTPX_ADAPTER_VERSIONS as HTTPX_ADAPTER_VERSIONS +from mpp._httpx import HttpxCompatibilityError as HttpxCompatibilityError from mpp.events import ( CHALLENGE_RECEIVED, CREDENTIAL_CREATED, @@ -542,6 +544,18 @@ def emit_event_sync(self, name: str, payload: EventPayload) -> Any: def allows_http_payment(self, url: httpx.URL) -> bool: return self._allowed_origins.allows(url) + def wrap_client(self, client: httpx.Client) -> httpx.Client: + """Make one existing synchronous HTTPX client payment-aware.""" + from mpp._httpx import wrap_client + + return wrap_client(client, self) + + def wrap_async_client(self, client: httpx.AsyncClient) -> httpx.AsyncClient: + """Make one existing asynchronous HTTPX client payment-aware.""" + from mpp._httpx import wrap_async_client + + return wrap_async_client(client, self) + def reset_unknown_outcomes(self, *, reconciled: bool) -> None: self._core().reset_unknown_outcomes(reconciled=reconciled) diff --git a/tests/test_httpx_adapters.py b/tests/test_httpx_adapters.py new file mode 100644 index 0000000..7be1510 --- /dev/null +++ b/tests/test_httpx_adapters.py @@ -0,0 +1,582 @@ +from __future__ import annotations + +import inspect +import io +import threading +from contextlib import asynccontextmanager +from typing import Any, Literal, cast +from unittest.mock import AsyncMock + +import httpx +import pytest + +import mpp._httpx as httpx_adapter +from mpp import Challenge, Credential +from mpp.errors import PaymentError, PaymentOutcomeUnknownError +from mpp.runtime import HTTPX_ADAPTER_VERSIONS, HttpxCompatibilityError, OwnedPaymentRuntime + +Kind = Literal["sync", "async"] + + +class Method: + name = "tempo" + _intents = {"charge": True} + + def __init__(self) -> None: + self.create_credential = AsyncMock(side_effect=self._create_credential) + + async def _create_credential(self, challenge: Challenge) -> Credential: + return Credential(challenge=challenge.to_echo(), payload={"hash": "0xabc"}) + + +def required(challenge_id: str = "challenge", **kwargs: Any) -> httpx.Response: + challenge = Challenge(id=challenge_id, method="tempo", intent="charge", request={}) + headers = httpx.Headers(kwargs.pop("headers", {})) + headers["www-authenticate"] = challenge.to_www_authenticate("example.com") + return httpx.Response(402, headers=headers, **kwargs) + + +def make_client( + kind: Kind, + runtime: OwnedPaymentRuntime, + handler: Any, + **kwargs: Any, +) -> httpx.Client | httpx.AsyncClient: + if kind == "sync": + return runtime.wrap_client(httpx.Client(transport=httpx.MockTransport(handler), **kwargs)) + return runtime.wrap_async_client( + httpx.AsyncClient(transport=httpx.MockTransport(handler), **kwargs) + ) + + +async def send( + kind: Kind, + client: httpx.Client | httpx.AsyncClient, + method: str = "GET", + url: str = "https://example.com/paid", + **kwargs: Any, +) -> httpx.Response: + if kind == "sync": + return cast(httpx.Client, client).request(method, url, **kwargs) + return await cast(httpx.AsyncClient, client).request(method, url, **kwargs) + + +async def send_request( + kind: Kind, + client: httpx.Client | httpx.AsyncClient, + request: httpx.Request, +) -> httpx.Response: + if kind == "sync": + return cast(httpx.Client, client).send(request) + return await cast(httpx.AsyncClient, client).send(request) + + +async def close(kind: Kind, client: httpx.Client | httpx.AsyncClient) -> None: + if kind == "sync": + cast(httpx.Client, client).close() + else: + await cast(httpx.AsyncClient, client).aclose() + + +class TwoRequestAuth(httpx.Auth): + def auth_flow(self, request: httpx.Request): + response = yield request + assert response.status_code == 200 + yield httpx.Request("GET", "https://example.com/second") + + +@pytest.mark.parametrize("kind", ["sync", "async"]) +async def test_one_public_send_never_pays_twice(kind: Kind) -> None: + method = Method() + runtime = OwnedPaymentRuntime([method]) + + def handler(request: httpx.Request) -> httpx.Response: + if request.headers.get("authorization", "").startswith("Payment "): + return httpx.Response(200) + return required(request.url.path) + + client = make_client(kind, runtime, handler, auth=TwoRequestAuth()) + try: + response = await send(kind, client, url="https://example.com/first") + finally: + await close(kind, client) + runtime.close() + + assert response.status_code == 402 + method.create_credential.assert_awaited_once() + + +@pytest.mark.parametrize("kind", ["sync", "async"]) +async def test_each_public_send_gets_a_fresh_payment_budget(kind: Kind) -> None: + method = Method() + runtime = OwnedPaymentRuntime([method]) + + def handler(request: httpx.Request) -> httpx.Response: + if request.headers.get("authorization", "").startswith("Payment "): + if request.url.path == "/first": + return httpx.Response(302, headers={"location": "/second"}) + return httpx.Response(200) + return required(request.url.path) + + client = make_client(kind, runtime, handler) + request = client.build_request("GET", "https://example.com/first") + try: + first = await send_request(kind, client, request) + assert first.status_code == 302 + assert first.next_request is not None + assert (await send_request(kind, client, first.next_request)).is_success + assert (await send_request(kind, client, request)).status_code == 302 + finally: + await close(kind, client) + runtime.close() + + assert method.create_credential.await_count == 3 + + +@pytest.mark.parametrize("kind", ["sync", "async"]) +async def test_auth_redirect_cookies_and_hooks(kind: Kind) -> None: + method = Method() + runtime = OwnedPaymentRuntime([method]) + requests: list[tuple[str, str, str]] = [] + hooks: list[int] = [] + + def handler(request: httpx.Request) -> httpx.Response: + requests.append( + ( + request.url.path, + request.headers.get("authorization", ""), + request.headers.get("cookie", ""), + ) + ) + if request.url.path == "/start": + return httpx.Response( + 302, + headers={"location": "/paid", "set-cookie": "redirect=yes; Path=/"}, + ) + if request.headers.get("authorization", "").startswith("Payment "): + return httpx.Response(200, content=b"paid") + return required(headers={"set-cookie": "challenge=yes; Path=/"}) + + if kind == "sync": + + def sync_hook(response: httpx.Response) -> None: + hooks.append(response.status_code) + + hook = sync_hook + else: + + async def async_hook(response: httpx.Response) -> None: + hooks.append(response.status_code) + + hook = async_hook + client = make_client( + kind, + runtime, + handler, + auth=("user", "password"), + event_hooks={"response": [hook]}, + follow_redirects=True, + ) + try: + response = await send(kind, client, url="https://example.com/start") + finally: + await close(kind, client) + runtime.close() + + assert response.content == b"paid" + assert [item[0] for item in requests] == ["/start", "/paid", "/paid"] + assert requests[0][1].startswith("Basic ") + assert requests[1][1].startswith("Basic ") + assert requests[2][1].startswith("Payment ") + assert "redirect=yes" in requests[1][2] + assert "challenge=yes" in requests[2][2] + assert hooks == [302, 200] + + +@pytest.mark.parametrize("kind", ["sync", "async"]) +async def test_free_and_disallowed_requests_do_not_start_runtime(kind: Kind) -> None: + entered: list[bool] = [] + + @asynccontextmanager + async def factory(): + entered.append(True) + yield Method() + + runtime = OwnedPaymentRuntime( + method_factories=[factory], + allowed_origins=["https://allowed.example"], + ) + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(200) if request.url.path == "/free" else required() + + client = make_client(kind, runtime, handler) + try: + assert (await send(kind, client, url="https://allowed.example/free")).status_code == 200 + assert (await send(kind, client, url="https://blocked.example/paid")).status_code == 402 + assert entered == [] + finally: + await close(kind, client) + runtime.close() + + +@pytest.mark.parametrize("kind", ["sync", "async"]) +async def test_multipart_is_replayed_byte_for_byte(kind: Kind) -> None: + method = Method() + runtime = OwnedPaymentRuntime([method]) + bodies: list[bytes] = [] + + def handler(request: httpx.Request) -> httpx.Response: + bodies.append(request.content) + return httpx.Response(200, content=b"paid") if len(bodies) == 2 else required() + + client = make_client(kind, runtime, handler) + try: + response = await send( + kind, + client, + "POST", + files={"upload": ("data.txt", io.BytesIO(b"file-content"))}, + ) + finally: + await close(kind, client) + runtime.close() + + assert response.content == b"paid" + assert bodies[0] == bodies[1] + assert b"file-content" in bodies[0] + + +class ConsumingSyncTransport(httpx.BaseTransport): + def __init__(self, bodies: list[bytes]) -> None: + self.bodies = bodies + + def handle_request(self, request: httpx.Request) -> httpx.Response: + self.bodies.append(b"".join(cast(httpx.SyncByteStream, request.stream))) + return required() + + +class ConsumingAsyncTransport(httpx.AsyncBaseTransport): + def __init__(self, bodies: list[bytes]) -> None: + self.bodies = bodies + + async def handle_async_request(self, request: httpx.Request) -> httpx.Response: + stream = cast(httpx.AsyncByteStream, request.stream) + self.bodies.append(b"".join([chunk async for chunk in stream])) + return required() + + +@pytest.mark.parametrize("kind", ["sync", "async"]) +async def test_one_shot_paid_upload_fails_before_credential(kind: Kind) -> None: + method = Method() + runtime = OwnedPaymentRuntime([method]) + bodies: list[bytes] = [] + if kind == "sync": + client: httpx.Client | httpx.AsyncClient = httpx.Client( + transport=ConsumingSyncTransport(bodies) + ) + + def sync_content(): + yield b"one-shot" + + content = sync_content + client = runtime.wrap_client(client) + else: + client = httpx.AsyncClient(transport=ConsumingAsyncTransport(bodies)) + + async def async_content(): + yield b"one-shot" + + content = async_content + client = runtime.wrap_async_client(client) + + try: + with pytest.raises(PaymentError, match="cannot be replayed"): + await send(kind, client, "POST", content=content()) + finally: + await close(kind, client) + runtime.close() + + assert bodies == [b"one-shot"] + method.create_credential.assert_not_awaited() + + +@pytest.mark.parametrize("kind", ["sync", "async"]) +async def test_hook_failure_does_not_change_known_payment(kind: Kind) -> None: + method = Method() + runtime = OwnedPaymentRuntime([method]) + + def handler(request: httpx.Request) -> httpx.Response: + if "authorization" in request.headers: + return httpx.Response(200, content=b"paid") + return required() + + if kind == "sync": + + def sync_hook(response: httpx.Response) -> None: + response.read() + raise RuntimeError("hook failed") + + hook = sync_hook + else: + + async def async_hook(response: httpx.Response) -> None: + await response.aread() + raise RuntimeError("hook failed") + + hook = async_hook + client = make_client(kind, runtime, handler, event_hooks={"response": [hook]}) + try: + with pytest.raises(RuntimeError, match="hook failed"): + await send(kind, client) + client.event_hooks["response"].clear() + assert (await send(kind, client)).is_success + finally: + await close(kind, client) + runtime.close() + assert method.create_credential.await_count == 2 + + +class BrokenSyncStream(httpx.SyncByteStream): + def __iter__(self): + raise httpx.ReadError("body lost") + yield b"" # pragma: no cover + + +class BrokenAsyncStream(httpx.AsyncByteStream): + async def __aiter__(self): + raise httpx.ReadError("body lost") + yield b"" # pragma: no cover + + +@pytest.mark.parametrize("kind", ["sync", "async"]) +async def test_only_send_failure_retains_unknown_outcome( + kind: Kind, +) -> None: + method = Method() + runtime = OwnedPaymentRuntime([method]) + failure = "body" + + def handler(request: httpx.Request) -> httpx.Response: + if "authorization" not in request.headers: + return required() + if failure == "send": + raise httpx.ReadTimeout("response lost", request=request) + if failure == "body": + stream = BrokenSyncStream() if kind == "sync" else BrokenAsyncStream() + return httpx.Response(200, stream=stream) + return httpx.Response(200, content=b"paid") + + client = make_client(kind, runtime, handler) + try: + request = client.build_request("GET", "https://example.com/paid") + if kind == "sync": + response = cast(httpx.Client, client).send(request, stream=True) + with pytest.raises(httpx.ReadError, match="body lost"): + response.read() + response.close() + else: + response = await cast(httpx.AsyncClient, client).send(request, stream=True) + with pytest.raises(httpx.ReadError, match="body lost"): + await response.aread() + await response.aclose() + failure = "ok" + assert (await send(kind, client)).is_success + failure = "send" + with pytest.raises(PaymentOutcomeUnknownError): + await send(kind, client) + with pytest.raises(PaymentOutcomeUnknownError): + await send(kind, client) + finally: + await close(kind, client) + runtime.close() + assert method.create_credential.await_count == 3 + + +@pytest.mark.parametrize("kind", ["sync", "async"]) +async def test_repeated_challenge_and_redirect_never_pay_twice(kind: Kind) -> None: + method = Method() + runtime = OwnedPaymentRuntime([method]) + mode = "repeat" + + def handler(request: httpx.Request) -> httpx.Response: + if mode == "repeat": + return required("repeat") + if request.url.path == "/first": + return ( + httpx.Response(302, headers={"location": "/second"}) + if "authorization" in request.headers + else required("first") + ) + return required("second") + + client = make_client(kind, runtime, handler, follow_redirects=True) + try: + with pytest.raises(PaymentOutcomeUnknownError): + await send(kind, client) + runtime.reset_unknown_outcomes(reconciled=True) + mode = "redirect" + response = await send(kind, client, url="https://example.com/first") + assert response.status_code == 402 + finally: + await close(kind, client) + runtime.close() + assert method.create_credential.await_count == 2 + + +@pytest.mark.parametrize("kind", ["sync", "async"]) +async def test_nested_hook_send_is_a_fresh_operation(kind: Kind) -> None: + method = Method() + runtime = OwnedPaymentRuntime([method]) + client: httpx.Client | httpx.AsyncClient + + def handler(request: httpx.Request) -> httpx.Response: + return ( + httpx.Response(200, content=request.url.path.encode()) + if "authorization" in request.headers + else required(request.url.path) + ) + + if kind == "sync": + + def sync_hook(response: httpx.Response) -> None: + if response.request.url.path == "/outer": + assert cast(httpx.Client, client).get("https://example.com/inner").is_success + + hook = sync_hook + else: + + async def async_hook(response: httpx.Response) -> None: + if response.request.url.path == "/outer": + nested = await cast(httpx.AsyncClient, client).get("https://example.com/inner") + assert nested.is_success + + hook = async_hook + client = make_client(kind, runtime, handler, event_hooks={"response": [hook]}) + try: + assert (await send(kind, client, url="https://example.com/outer")).is_success + finally: + await close(kind, client) + runtime.close() + assert method.create_credential.await_count == 2 + + +def test_idempotence_signatures_and_stale_adapter_rejection() -> None: + runtime = OwnedPaymentRuntime() + other = OwnedPaymentRuntime() + client = httpx.Client(transport=httpx.MockTransport(lambda _: httpx.Response(200))) + signatures = ( + inspect.signature(client._send_single_request), + inspect.signature(client.send), + ) + original_public = inspect.getattr_static(client, "send") + client = runtime.wrap_client(client) + private = inspect.getattr_static(client, "_send_single_request") + public = inspect.getattr_static(client, "send") + try: + assert runtime.wrap_client(client) is client + assert inspect.signature(client._send_single_request) == signatures[0] + assert inspect.signature(client.send) == signatures[1] + assert public is not original_public + with pytest.raises(RuntimeError, match="another payment runtime"): + other.wrap_client(client) + client._send_single_request = lambda _request: httpx.Response(200) # type: ignore[method-assign] + with pytest.raises(HttpxCompatibilityError, match="replaced"): + runtime.wrap_client(client) + finally: + client._send_single_request = private + client.send = public + client.close() + runtime.close() + other.close() + + +def test_concurrent_rebind_has_one_fixed_runtime() -> None: + methods = [Method(), Method()] + runtimes = [OwnedPaymentRuntime([method]) for method in methods] + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(200) if "authorization" in request.headers else required() + + client = httpx.Client(transport=httpx.MockTransport(handler)) + barrier = threading.Barrier(3) + outcomes: list[int | Exception] = [] + + def bind(index: int) -> None: + barrier.wait() + try: + runtimes[index].wrap_client(client) + outcomes.append(index) + except Exception as error: + outcomes.append(error) + + threads = [threading.Thread(target=bind, args=(index,)) for index in range(2)] + for thread in threads: + thread.start() + barrier.wait() + for thread in threads: + thread.join() + try: + assert client.get("https://example.com").is_success + finally: + client.close() + for runtime in runtimes: + runtime.close() + + winner = next(cast(int, value) for value in outcomes if isinstance(value, int)) + assert sum(isinstance(value, RuntimeError) for value in outcomes) == 1 + assert methods[winner].create_credential.await_count == 1 + assert methods[1 - winner].create_credential.await_count == 0 + + +def test_compatibility_failures_do_not_mutate_client(monkeypatch: pytest.MonkeyPatch) -> None: + runtime = OwnedPaymentRuntime() + client = httpx.Client() + before = { + name: inspect.getattr_static(client, name) for name in ("send", "_send_single_request") + } + monkeypatch.setattr(httpx_adapter, "version", lambda _: "0.29.0") + try: + with pytest.raises(HttpxCompatibilityError, match=r">=0.27,<0.29"): + runtime.wrap_client(client) + finally: + client.close() + runtime.close() + assert HTTPX_ADAPTER_VERSIONS == ">=0.27,<0.29" + assert all(inspect.getattr_static(client, name) is value for name, value in before.items()) + assert httpx_adapter._MARKER not in client.__dict__ + + +def test_seam_and_assignment_failures_roll_back() -> None: + runtime = OwnedPaymentRuntime() + client = httpx.Client() + original_private = inspect.getattr_static(client, "_send_single_request") + + def incompatible(request: httpx.Request, extra: object) -> httpx.Response: + raise AssertionError + + client._send_single_request = incompatible # type: ignore[method-assign] + with pytest.raises(HttpxCompatibilityError, match="instance._send_single_request"): + runtime.wrap_client(client) + assert inspect.getattr_static(client, "_send_single_request") is incompatible + client._send_single_request = original_private + client.close() + + class RejectingClient(httpx.Client): + def __setattr__(self, name: str, value: object) -> None: + if name == httpx_adapter._MARKER: + raise RuntimeError("assignment rejected") + super().__setattr__(name, value) + + rejecting = RejectingClient() + before = { + name: inspect.getattr_static(rejecting, name) for name in ("send", "_send_single_request") + } + try: + with pytest.raises(RuntimeError, match="assignment rejected"): + runtime.wrap_client(rejecting) + finally: + rejecting.close() + runtime.close() + assert all(inspect.getattr_static(rejecting, name) is value for name, value in before.items()) + assert httpx_adapter._MARKER not in rejecting.__dict__