From 343af19723ff5d8eeab0f00c080a4cf0bd80a8f6 Mon Sep 17 00:00:00 2001 From: Parv Ahuja <17094219+parvahuja@users.noreply.github.com> Date: Mon, 20 Jul 2026 17:51:54 -0700 Subject: [PATCH] Add synchronous payment transport --- .changelog/sync-payment-transport.md | 5 + pyproject.toml | 1 + src/mpp/client/__init__.py | 1 + src/mpp/client/sync_transport.py | 211 +++++++++++++++++ src/mpp/client/transport.py | 103 +++++---- src/mpp/runtime.py | 276 +++++++++++++++++++--- tests/test_sync_client.py | 329 +++++++++++++++++++++++++++ 7 files changed, 855 insertions(+), 71 deletions(-) create mode 100644 .changelog/sync-payment-transport.md create mode 100644 src/mpp/client/sync_transport.py create mode 100644 tests/test_sync_client.py diff --git a/.changelog/sync-payment-transport.md b/.changelog/sync-payment-transport.md new file mode 100644 index 00000000..0a541896 --- /dev/null +++ b/.changelog/sync-payment-transport.md @@ -0,0 +1,5 @@ +--- +pympp: minor +--- + +Added synchronous HTTP payment handling and existing-client wrapping to `PaymentRuntime`. diff --git a/pyproject.toml b/pyproject.toml index ea003b2f..e87d48bc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -59,6 +59,7 @@ dev = [ "pytest-cov>=7.0.0", "pytest-httpx>=0.36.0", "pyright>=1.1", + "openai>=2.24,<3", "eth-abi>=5.0,<6", "build>=1.0", "twine>=6.0", diff --git a/src/mpp/client/__init__.py b/src/mpp/client/__init__.py index 2b000e25..2732e297 100644 --- a/src/mpp/client/__init__.py +++ b/src/mpp/client/__init__.py @@ -17,6 +17,7 @@ """ from mpp import _expires as Expires +from mpp.client.sync_transport import SyncPaymentTransport from mpp.client.transport import Client, PaymentTransport, get, post, request from mpp.events import ( CHALLENGE_RECEIVED, diff --git a/src/mpp/client/sync_transport.py b/src/mpp/client/sync_transport.py new file mode 100644 index 00000000..c552a00f --- /dev/null +++ b/src/mpp/client/sync_transport.py @@ -0,0 +1,211 @@ +"""Synchronous payment-aware httpx transport.""" + +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING + +import httpx + +from mpp.errors import PaymentError +from mpp.events import ( + CHALLENGE_RECEIVED, + CREDENTIAL_CREATED, + PAYMENT_FAILED, + PAYMENT_RESPONSE, + EventDispatcher, + EventHandler, + Unsubscribe, +) +from mpp.runtime import Method, PaymentRuntime + +from .transport import ( + _challenge_is_expired, + _challenged_request, + _client_payment_failed_payload, + _payment_challenges, +) + +if TYPE_CHECKING: + from collections.abc import Sequence + +logger = logging.getLogger(__name__) + + +class SyncPaymentTransport(httpx.BaseTransport): + """httpx transport that synchronously handles 402 payment challenges.""" + + def __init__( + self, + methods: Sequence[Method] | None = None, + inner: httpx.BaseTransport | None = None, + events: EventDispatcher | None = None, + *, + runtime: PaymentRuntime | None = None, + ) -> None: + self._owns_runtime = runtime is None + if runtime is not None: + if methods is not None or events is not None: + raise ValueError("Pass either methods/events or runtime, not both") + self._runtime = runtime + else: + if methods is None: + raise ValueError("Pass methods or runtime") + self._runtime = PaymentRuntime(methods, events=events) + self._inner = inner or httpx.HTTPTransport() + self._events = self._runtime.events + + def on(self, name: str, handler: EventHandler) -> Unsubscribe: + """Register a client payment event handler.""" + return self._events.on(name, handler) + + def on_challenge_received(self, handler: EventHandler) -> Unsubscribe: + return self.on(CHALLENGE_RECEIVED, handler) + + def on_credential_created(self, handler: EventHandler) -> Unsubscribe: + return self.on(CREDENTIAL_CREATED, handler) + + def on_payment_response(self, handler: EventHandler) -> Unsubscribe: + return self.on(PAYMENT_RESPONSE, handler) + + def on_payment_failed(self, handler: EventHandler) -> Unsubscribe: + return self.on(PAYMENT_FAILED, handler) + + def handle_request(self, request: httpx.Request) -> httpx.Response: + """Send a request and retry one 402 with a payment credential.""" + if isinstance(request.stream, httpx.SyncByteStream) and not isinstance( + request.stream, httpx.AsyncByteStream + ): + raise PaymentError( + "Streaming request bodies (generators) are not supported through the " + "payment retry flow. Use a buffered body (bytes, str, files=, or data=) instead." + ) + + request.read() + response = self._inner.handle_request(request) + if response.status_code != 402: + return response + + response.read() + challenged_request = _challenged_request(response, request) + if not self._runtime.allows_http_payment(challenged_request.url): + return response + challenged_request.read() + + challenges, parse_error = _payment_challenges(response) + try: + challenge, method = self._runtime.match_challenge( + challenges, + prefer_method_order=False, + ) + except ValueError: + challenge = None + method = None + + if challenge is None or method is None: + if parse_error is not None or challenges: + self._runtime.emit_event_sync( + PAYMENT_FAILED, + _client_payment_failed_payload( + challenge=None, + challenges=challenges, + credential=None, + error=parse_error + or ValueError("No compatible payment method for challenges"), + method=None, + request=challenged_request, + response=response, + ), + ) + return response + + if _challenge_is_expired(challenge): + logger.warning("Challenge expired at %s, not paying", challenge.expires) + self._runtime.emit_event_sync( + PAYMENT_FAILED, + _client_payment_failed_payload( + challenge=challenge, + challenges=challenges, + credential=None, + error=ValueError(f"Challenge expired at {challenge.expires}"), + method=method, + request=challenged_request, + response=response, + ), + ) + return response + + try: + credential = self._runtime.create_credential_sync( + challenge, + method, + event_payload={ + "challenges": challenges, + "request": challenged_request, + "response": response, + "protocol": "http", + }, + ) + except Exception as error: + self._runtime.emit_event_sync( + PAYMENT_FAILED, + _client_payment_failed_payload( + challenge=challenge, + challenges=challenges, + credential=None, + error=error, + method=method, + request=challenged_request, + response=response, + ), + ) + raise + + headers = httpx.Headers(challenged_request.headers) + headers["Authorization"] = credential.to_authorization() + retry_request = httpx.Request( + method=challenged_request.method, + url=challenged_request.url, + headers=headers, + content=challenged_request.content, + extensions=challenged_request.extensions, + ) + + try: + payment_response = self._inner.handle_request(retry_request) + except Exception as error: + self._runtime.emit_event_sync( + PAYMENT_FAILED, + _client_payment_failed_payload( + challenge=challenge, + challenges=challenges, + credential=credential, + error=error, + method=method, + request=challenged_request, + response=response, + ), + ) + raise + + if payment_response.is_success: + self._runtime.emit_event_sync( + PAYMENT_RESPONSE, + { + "challenge": challenge, + "credential": credential, + "method": method, + "request": challenged_request, + "response": payment_response, + "protocol": "http", + }, + ) + return payment_response + + def close(self) -> None: + """Close the inner transport.""" + try: + self._inner.close() + finally: + if self._owns_runtime: + self._runtime.close() diff --git a/src/mpp/client/transport.py b/src/mpp/client/transport.py index da0d4b8f..ccf7ff91 100644 --- a/src/mpp/client/transport.py +++ b/src/mpp/client/transport.py @@ -57,6 +57,39 @@ def _client_payment_failed_payload( } +def _challenged_request( + response: httpx.Response, + fallback: httpx.Request, +) -> httpx.Request: + try: + return response.request + except RuntimeError: + return fallback + + +def _payment_challenges(response: httpx.Response) -> tuple[list[Challenge], ParseError | None]: + challenges: list[Challenge] = [] + parse_error: ParseError | None = None + for header in response.headers.get_list("www-authenticate"): + if not header.lower().startswith("payment "): + continue + try: + challenges.append(Challenge.from_www_authenticate(header)) + except ParseError as error: + parse_error = error + return challenges, parse_error + + +def _challenge_is_expired(challenge: Challenge) -> bool: + if not challenge.expires: + return False + try: + expires = datetime.fromisoformat(challenge.expires.replace("Z", "+00:00")) + except ValueError: + return False + return expires < datetime.now(UTC) + + class PaymentTransport(httpx.AsyncBaseTransport): """httpx transport that handles 402 Payment Required responses. @@ -84,6 +117,7 @@ def __init__( *, runtime: PaymentRuntime | None = None, ) -> None: + self._owns_runtime = runtime is None if runtime is not None: if methods is not None or events is not None: raise ValueError("Pass either methods/events or runtime, not both") @@ -145,28 +179,12 @@ async def handle_async_request(self, request: httpx.Request) -> httpx.Response: # A high-level send may have followed redirects before returning the # 402. Apply policy and retry against the request that was challenged. - try: - challenged_request = response.request - except RuntimeError: - challenged_request = request + challenged_request = _challenged_request(response, request) if not self._runtime.allows_http_payment(challenged_request.url): return response await challenged_request.aread() - # Handle multiple WWW-Authenticate headers (per RFC 9110) - www_auth_headers = response.headers.get_list("www-authenticate") - - challenges: list[Challenge] = [] - parse_error: ParseError | None = None - for header in www_auth_headers: - if not header.lower().startswith("payment "): - continue - try: - parsed = Challenge.from_www_authenticate(header) - except ParseError as error: - parse_error = error - continue - challenges.append(parsed) + challenges, parse_error = _payment_challenges(response) try: challenge, matched_method = self._runtime.match_challenge( @@ -181,7 +199,7 @@ async def handle_async_request(self, request: httpx.Request) -> httpx.Response: if parse_error is not None or challenges: # Surface parse/method-selection failures to observers while # preserving the original 402 response for the caller. - await self._events.emit( + await self._runtime.emit_event( PAYMENT_FAILED, _client_payment_failed_payload( challenge=None, @@ -197,26 +215,21 @@ async def handle_async_request(self, request: httpx.Request) -> httpx.Response: return response # Check expiry before paying (client-side guardrail) - if challenge.expires: - try: - expires_dt = datetime.fromisoformat(challenge.expires.replace("Z", "+00:00")) - if expires_dt < datetime.now(UTC): - logger.warning("Challenge expired at %s, not paying", challenge.expires) - await self._events.emit( - PAYMENT_FAILED, - _client_payment_failed_payload( - challenge=challenge, - challenges=challenges, - credential=None, - error=ValueError(f"Challenge expired at {challenge.expires}"), - method=matched_method, - request=request, - response=response, - ), - ) - return response - except ValueError: - pass # If we can't parse, let server validate + if _challenge_is_expired(challenge): + logger.warning("Challenge expired at %s, not paying", challenge.expires) + await self._runtime.emit_event( + PAYMENT_FAILED, + _client_payment_failed_payload( + challenge=challenge, + challenges=challenges, + credential=None, + error=ValueError(f"Challenge expired at {challenge.expires}"), + method=matched_method, + request=challenged_request, + response=response, + ), + ) + return response try: credential = await self._runtime.create_credential( @@ -231,7 +244,7 @@ async def handle_async_request(self, request: httpx.Request) -> httpx.Response: ) auth_header = credential.to_authorization() except Exception as error: - await self._events.emit( + await self._runtime.emit_event( PAYMENT_FAILED, _client_payment_failed_payload( challenge=challenge, @@ -259,7 +272,7 @@ async def handle_async_request(self, request: httpx.Request) -> httpx.Response: try: payment_response = await self._inner.handle_async_request(retry_request) except Exception as error: - await self._events.emit( + await self._runtime.emit_event( PAYMENT_FAILED, _client_payment_failed_payload( challenge=challenge, @@ -274,7 +287,7 @@ async def handle_async_request(self, request: httpx.Request) -> httpx.Response: raise if payment_response.is_success: - await self._events.emit( + await self._runtime.emit_event( PAYMENT_RESPONSE, { "challenge": challenge, @@ -290,7 +303,11 @@ async def handle_async_request(self, request: httpx.Request) -> httpx.Response: async def aclose(self) -> None: """Close the inner transport.""" - await self._inner.aclose() + try: + await self._inner.aclose() + finally: + if self._owns_runtime: + await self._runtime.aclose() class Client: diff --git a/src/mpp/runtime.py b/src/mpp/runtime.py index 185d8d9f..2fdb01fd 100644 --- a/src/mpp/runtime.py +++ b/src/mpp/runtime.py @@ -2,7 +2,10 @@ from __future__ import annotations -from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable +import asyncio +import threading +from contextvars import ContextVar, copy_context +from typing import TYPE_CHECKING, Any, Protocol, TypeVar, runtime_checkable from urllib.parse import urlparse import httpx @@ -14,12 +17,21 @@ PAYMENT_FAILED, PAYMENT_RESPONSE, EventDispatcher, + EventPayload, ) if TYPE_CHECKING: - from collections.abc import Callable, Sequence + from collections.abc import Callable, Coroutine, Sequence - from mpp.client import PaymentTransport + from mpp.client import PaymentTransport, SyncPaymentTransport + +_T = TypeVar("_T") +_PAYMENT_FLOW_ACTIVE: ContextVar[bool] = ContextVar("mpp_payment_flow_active", default=False) + + +def payment_flow_active() -> bool: + """Return whether the current context is creating a payment credential.""" + return _PAYMENT_FLOW_ACTIVE.get() @runtime_checkable @@ -49,6 +61,136 @@ async def aclose(self) -> None: return None +class _BoundSyncSendTransport(httpx.BaseTransport): + def __init__(self, send: Any, args: tuple[Any, ...], kwargs: dict[str, Any]) -> None: + self._send = send + self._args = args + self._kwargs = kwargs + + def handle_request(self, request: httpx.Request) -> httpx.Response: + kwargs = dict(self._kwargs) + if request.headers.get("authorization", "").startswith("Payment "): + kwargs["auth"] = None + return self._send(request, *self._args, **kwargs) + + def close(self) -> None: + return None + + +class _AsyncBridge: + """Own one lazy event loop for synchronous payment-method calls.""" + + def __init__(self) -> None: + self._closed = False + self._lock = threading.Lock() + self._loop: asyncio.AbstractEventLoop | None = None + self._ready = threading.Event() + self._start_error: BaseException | None = None + self._thread: threading.Thread | None = None + + def _submit(self, coroutine: Coroutine[Any, Any, _T]) -> Any: + with self._lock: + if self._closed: + raise RuntimeError("PaymentRuntime is closed") + if self._thread is None: + self._thread = threading.Thread( + target=self._run, + name="pympp-payment-runtime", + daemon=True, + ) + self._thread.start() + self._ready.wait() + with self._lock: + if self._closed: + raise RuntimeError("PaymentRuntime is closed") + if self._start_error is not None: + raise RuntimeError("PaymentRuntime background loop failed to start") from ( + self._start_error + ) + if self._loop is None: + raise RuntimeError("PaymentRuntime background loop failed to start") + if threading.current_thread() is self._thread: + raise RuntimeError("Cannot block the PaymentRuntime background loop") + return copy_context().run( + asyncio.run_coroutine_threadsafe, + coroutine, + self._loop, + ) + + def _run(self) -> None: + try: + loop = asyncio.new_event_loop() + except BaseException as error: + self._start_error = error + self._ready.set() + return + self._loop = loop + asyncio.set_event_loop(loop) + self._ready.set() + try: + loop.run_forever() + finally: + pending = asyncio.all_tasks(loop) + for task in pending: + task.cancel() + if pending: + loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True)) + loop.run_until_complete(loop.shutdown_asyncgens()) + loop.run_until_complete(loop.shutdown_default_executor()) + loop.close() + + def run(self, coroutine: Coroutine[Any, Any, _T]) -> _T: + """Run an async payment operation from synchronous code.""" + try: + future = self._submit(coroutine) + except BaseException: + coroutine.close() + raise + try: + return future.result() + except BaseException: + future.cancel() + raise + + async def _cancel_pending(self) -> None: + current = asyncio.current_task() + pending = [task for task in asyncio.all_tasks() if task is not current] + for task in pending: + task.cancel() + if pending: + await asyncio.gather(*pending, return_exceptions=True) + + def close(self) -> None: + """Stop the runtime loop, if it was started.""" + with self._lock: + if self._closed: + return + self._closed = True + thread = self._thread + if thread is None: + return + if self._loop is None: + self._ready.wait() + loop = self._loop + if loop is None: + thread.join() + return + if threading.current_thread() is thread: + + async def shutdown() -> None: + await self._cancel_pending() + loop.stop() + + loop.create_task(shutdown()) + return + if thread.is_alive(): + future = asyncio.run_coroutine_threadsafe(self._cancel_pending(), loop) + future.result() + if loop is not None: + loop.call_soon_threadsafe(loop.stop) + thread.join() + + class PaymentRuntime: """Reusable payment runtime for HTTP and MCP payment handling.""" @@ -62,6 +204,7 @@ def __init__( self.methods = tuple(methods) self.events = events or EventDispatcher() self._allowed = _AllowedOrigins(allowed_origins) + self._bridge = _AsyncBridge() def payment_transport(self, inner: httpx.AsyncBaseTransport | None = None) -> PaymentTransport: """Create an httpx transport using this runtime's payment methods.""" @@ -72,6 +215,31 @@ def payment_transport(self, inner: httpx.AsyncBaseTransport | None = None) -> Pa runtime=self, ) + def sync_payment_transport( + self, inner: httpx.BaseTransport | None = None + ) -> SyncPaymentTransport: + """Create a synchronous httpx transport using this runtime.""" + from mpp.client import SyncPaymentTransport + + return SyncPaymentTransport(inner=inner, runtime=self) + + def wrap_client(self, client: httpx.Client) -> httpx.Client: + """Make one existing Client payment-aware without global instrumentation.""" + client._mpp_payment_runtime = self # type: ignore[attr-defined] + if getattr(client, "_mpp_payment_wrapped", False): + return client + + original_send = client.send + + def send(request: httpx.Request, *args: Any, **kwargs: Any) -> httpx.Response: + runtime = getattr(client, "_mpp_payment_runtime", self) + return runtime.send_httpx_sync(original_send, request, *args, **kwargs) + + client._mpp_payment_original_send = original_send # type: ignore[attr-defined] + client._mpp_payment_wrapped = True # type: ignore[attr-defined] + client.send = send # type: ignore[method-assign] + return client + def wrap_async_client(self, client: httpx.AsyncClient) -> httpx.AsyncClient: """Make one existing AsyncClient payment-aware without global instrumentation.""" client._mpp_payment_runtime = self # type: ignore[attr-defined] @@ -111,6 +279,17 @@ async def send_httpx( transport = _BoundSendTransport(send, args, dict(kwargs)) return await self.payment_transport(inner=transport).handle_async_request(request) + def send_httpx_sync( + self, + send: Any, + request: httpx.Request, + *args: Any, + **kwargs: Any, + ) -> httpx.Response: + """Send one sync httpx request with automatic 402 payment handling.""" + transport = _BoundSyncSendTransport(send, args, dict(kwargs)) + return self.sync_payment_transport(inner=transport).handle_request(request) + async def call_mcp_tool( self, call_tool: Any, @@ -148,7 +327,7 @@ async def call_mcp_tool( error = ValueError( "Server returned malformed payment challenges or disallowed payment origins" ) - await self.events.emit( + await self.emit_event( PAYMENT_FAILED, { "challenge": None, @@ -179,7 +358,7 @@ async def call_mcp_tool( ) mcp_credential = MCPCredential.from_core(core_credential, challenge) except Exception as error: - await self.events.emit( + await self.emit_event( PAYMENT_FAILED, { "challenge": locals().get("core_challenge"), @@ -202,7 +381,7 @@ async def call_mcp_tool( payment_response = await call_tool(name, arguments, *args, **retry_kwargs) except Exception as error: outcome_error = PaymentOutcomeUnknownError(challenge, error) - await self.events.emit( + await self.emit_event( PAYMENT_FAILED, { "challenge": core_challenge, @@ -216,7 +395,7 @@ async def call_mcp_tool( ) raise outcome_error from error - await self.events.emit( + await self.emit_event( PAYMENT_RESPONSE, { "challenge": core_challenge, @@ -270,28 +449,69 @@ async def create_credential( *, event_payload: dict[str, Any] | None = None, ) -> Credential: - """Create a credential and emit shared client lifecycle events.""" - payload = { - "challenge": challenge, - "challenges": [challenge], - "method": method, - **(event_payload or {}), - } - event_credential = await self.events.emit( - CHALLENGE_RECEIVED, - payload, - first_result=True, - ) - credential = ( - event_credential - if isinstance(event_credential, Credential) - else await method.create_credential(challenge) - ) - await self.events.emit( - CREDENTIAL_CREATED, - {**payload, "credential": credential}, + """Create a credential on the caller event loop.""" + return await self._create_credential(challenge, method, event_payload=event_payload) + + def create_credential_sync( + self, + challenge: Challenge, + method: Method, + *, + event_payload: dict[str, Any] | None = None, + ) -> Credential: + """Synchronously create a credential on the runtime-owned event loop.""" + return self._bridge.run( + self._create_credential(challenge, method, event_payload=event_payload) ) - return credential + + async def _create_credential( + self, + challenge: Challenge, + method: Method, + *, + event_payload: dict[str, Any] | None = None, + ) -> Credential: + token = _PAYMENT_FLOW_ACTIVE.set(True) + try: + payload = { + "challenge": challenge, + "challenges": [challenge], + "method": method, + **(event_payload or {}), + } + event_credential = await self.events.emit( + CHALLENGE_RECEIVED, + payload, + first_result=True, + ) + credential = ( + event_credential + if isinstance(event_credential, Credential) + else await method.create_credential(challenge) + ) + await self.events.emit( + CREDENTIAL_CREATED, + {**payload, "credential": credential}, + ) + return credential + finally: + _PAYMENT_FLOW_ACTIVE.reset(token) + + async def emit_event(self, name: str, payload: EventPayload) -> Any: + """Emit an asynchronous lifecycle event on the caller event loop.""" + return await self.events.emit(name, payload) + + def emit_event_sync(self, name: str, payload: EventPayload) -> Any: + """Synchronously emit a lifecycle event on the runtime-owned event loop.""" + return self._bridge.run(self.events.emit(name, payload)) + + def close(self) -> None: + """Release the runtime background loop.""" + self._bridge.close() + + async def aclose(self) -> None: + """Asynchronously release the runtime background loop.""" + await asyncio.to_thread(self.close) def allows_http_payment(self, url: httpx.URL) -> bool: """Return whether credentials may be created for an HTTP origin.""" diff --git a/tests/test_sync_client.py b/tests/test_sync_client.py new file mode 100644 index 00000000..c16cd6d2 --- /dev/null +++ b/tests/test_sync_client.py @@ -0,0 +1,329 @@ +"""Tests for synchronous payment-aware HTTP clients.""" + +from __future__ import annotations + +import asyncio +import threading +from concurrent.futures import ThreadPoolExecutor +from typing import Any +from unittest.mock import AsyncMock + +import httpx +import pytest + +from mpp import Challenge +from mpp.client import SyncPaymentTransport +from mpp.errors import PaymentError +from mpp.runtime import PaymentRuntime +from tests import make_credential + + +class MockMethod: + name = "tempo" + _intents = {"charge": True} + + def __init__(self) -> None: + self.loops: list[asyncio.AbstractEventLoop] = [] + self.create_credential = AsyncMock(side_effect=self._create_credential) + + async def _create_credential(self, challenge: Challenge): + self.loops.append(asyncio.get_running_loop()) + return make_credential({"hash": "0xabc"}, challenge_id=challenge.id) + + +class MockTransport(httpx.BaseTransport): + def __init__(self, responses: list[httpx.Response]) -> None: + self.responses = responses + self.requests: list[httpx.Request] = [] + self.closed = False + + def handle_request(self, request: httpx.Request) -> httpx.Response: + self.requests.append(request) + return self.responses.pop(0) + + def close(self) -> None: + self.closed = True + + +class TrackingStream(httpx.SyncByteStream): + def __init__(self, chunks: list[bytes]) -> None: + self.chunks = chunks + self.started = False + self.closed = False + + def __iter__(self): + self.started = True + yield from self.chunks + + def close(self) -> None: + self.closed = True + + +def challenge() -> Challenge: + return Challenge(id="test-id", method="tempo", intent="charge", request={}) + + +def payment_required() -> httpx.Response: + return httpx.Response( + 402, + headers={"www-authenticate": challenge().to_www_authenticate("example.com")}, + ) + + +class TestSyncPaymentTransport: + def test_passes_through_free_response(self) -> None: + inner = MockTransport([httpx.Response(200, content=b"ok")]) + transport = SyncPaymentTransport(methods=[], inner=inner) + try: + response = transport.handle_request(httpx.Request("GET", "https://example.com")) + finally: + transport.close() + + assert response.content == b"ok" + assert len(inner.requests) == 1 + + def test_replays_bytes_and_multipart_bodies(self) -> None: + requests = [ + httpx.Request("POST", "https://example.com", content=b'{"hello":"world"}'), + httpx.Request( + "POST", + "https://example.com", + files={"file": ("hello.txt", b"hello", "text/plain")}, + ), + ] + + for request in requests: + inner = MockTransport([payment_required(), httpx.Response(200, content=b"paid")]) + transport = SyncPaymentTransport(methods=[MockMethod()], inner=inner) + try: + response = transport.handle_request(request) + finally: + transport.close() + + assert response.status_code == 200 + assert inner.requests[1].content == inner.requests[0].content + assert inner.requests[1].headers["authorization"].startswith("Payment ") + + def test_rejects_generator_body_before_send(self) -> None: + def body(): + yield b"one-shot" + + inner = MockTransport([]) + transport = SyncPaymentTransport(methods=[MockMethod()], inner=inner) + try: + with pytest.raises(PaymentError, match="Streaming request bodies"): + transport.handle_request( + httpx.Request("POST", "https://example.com", content=body()) + ) + finally: + transport.close() + + assert inner.requests == [] + + def test_paid_stream_remains_lazy(self) -> None: + stream = TrackingStream([b"one", b"two"]) + inner = MockTransport([payment_required(), httpx.Response(200, stream=stream)]) + transport = SyncPaymentTransport(methods=[MockMethod()], inner=inner) + try: + response = transport.handle_request(httpx.Request("GET", "https://example.com")) + assert stream.started is False + assert response.read() == b"onetwo" + assert stream.started is True + finally: + transport.close() + + +class TestWrappedSyncClient: + def test_wrap_client_is_idempotent_and_preserves_payment_authorization(self) -> None: + inner = MockTransport([payment_required(), httpx.Response(200, content=b"paid")]) + runtime = PaymentRuntime([MockMethod()]) + client = httpx.Client(transport=inner, auth=("user", "password")) + try: + assert runtime.wrap_client(client) is client + assert runtime.wrap_client(client) is client + response = client.get("https://example.com/paid") + finally: + client.close() + runtime.close() + + assert response.status_code == 200 + assert inner.requests[0].headers["authorization"].startswith("Basic ") + assert inner.requests[1].headers["authorization"].startswith("Payment ") + + def test_redirected_402_uses_challenged_origin_policy(self) -> None: + requests: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + requests.append(request) + if request.url.host == "allowed.example": + return httpx.Response(302, headers={"location": "https://evil.example/paid"}) + return payment_required() + + method = MockMethod() + runtime = PaymentRuntime([method], allowed_origins=["https://allowed.example"]) + client = runtime.wrap_client( + httpx.Client(transport=httpx.MockTransport(handler), follow_redirects=True) + ) + try: + response = client.get("https://allowed.example/start") + finally: + client.close() + runtime.close() + + assert response.status_code == 402 + assert [request.url.host for request in requests] == ["allowed.example", "evil.example"] + method.create_credential.assert_not_called() + + +class TestRuntimeBridge: + def test_concurrent_sync_calls_share_one_method_loop(self) -> None: + method = MockMethod() + runtime = PaymentRuntime([method]) + try: + with ThreadPoolExecutor(max_workers=4) as pool: + list( + pool.map( + lambda _: runtime.create_credential_sync(challenge(), method), + range(4), + ) + ) + finally: + runtime.close() + + assert len(method.loops) == 4 + assert len({id(loop) for loop in method.loops}) == 1 + + @pytest.mark.asyncio + async def test_async_method_stays_on_caller_loop(self) -> None: + caller_loop = asyncio.get_running_loop() + future = caller_loop.create_future() + method = MockMethod() + + async def create(challenge: Challenge): + await future + return make_credential({"hash": "0xabc"}, challenge_id=challenge.id) + + method.create_credential = AsyncMock(side_effect=create) + runtime = PaymentRuntime([method]) + try: + task = asyncio.create_task(runtime.create_credential(challenge(), method)) + await asyncio.sleep(0) + future.set_result(None) + await task + finally: + runtime.close() + + def test_bridge_rejects_same_thread_blocking(self) -> None: + runtime = PaymentRuntime([]) + + async def block_bridge() -> None: + with pytest.raises(RuntimeError, match="Cannot block"): + runtime._bridge.run(asyncio.sleep(0)) + + try: + runtime._bridge.run(block_bridge()) + finally: + runtime.close() + + def test_close_cancels_in_flight_bridge_work(self) -> None: + started = threading.Event() + + class BlockingMethod(MockMethod): + async def _create_credential(self, challenge: Challenge) -> Any: + started.set() + await asyncio.Event().wait() + + method = BlockingMethod() + runtime = PaymentRuntime([method]) + errors: list[BaseException] = [] + + def create() -> None: + try: + runtime.create_credential_sync(challenge(), method) + except BaseException as error: + errors.append(error) + + thread = threading.Thread(target=create) + thread.start() + assert started.wait(1) + runtime.close() + thread.join(timeout=1) + + assert thread.is_alive() is False + assert errors + + def test_close_is_idempotent(self) -> None: + method = MockMethod() + runtime = PaymentRuntime([method]) + runtime.create_credential_sync(challenge(), method) + + runtime.close() + runtime.close() + + with pytest.raises(RuntimeError, match="closed"): + runtime.create_credential_sync(challenge(), method) + + +def test_openai_sync_streaming_retries_without_eager_read() -> None: + openai = pytest.importorskip("openai") + requests: list[httpx.Request] = [] + bodies: list[bytes] = [] + paid_stream = TrackingStream( + [ + b'data: {"id":"chatcmpl-test","object":"chat.completion.chunk",' + b'"created":0,"model":"test","choices":[{"index":0,"delta":' + b'{"content":"hel"},"finish_reason":null}]}\n\n', + b'data: {"id":"chatcmpl-test","object":"chat.completion.chunk",' + b'"created":0,"model":"test","choices":[{"index":0,"delta":' + b'{"content":"lo"},"finish_reason":"stop"}]}\n\n', + b"data: [DONE]\n\n", + ] + ) + + def handler(request: httpx.Request) -> httpx.Response: + requests.append(request) + bodies.append(request.content) + if len(requests) == 1: + return payment_required() + return httpx.Response( + 200, + headers={"content-type": "text/event-stream"}, + stream=paid_stream, + ) + + runtime = PaymentRuntime([MockMethod()]) + http_client = runtime.wrap_client(httpx.Client(transport=httpx.MockTransport(handler))) + client = openai.OpenAI( + api_key="test", + base_url="https://example.com/v1", + http_client=http_client, + max_retries=0, + ) + result: dict[str, Any] = {} + + def run() -> None: + try: + stream = client.chat.completions.create( + model="test", + messages=[{"role": "user", "content": "hello"}], + stream=True, + ) + result["lazy"] = not paid_stream.started + result["text"] = "".join(chunk.choices[0].delta.content or "" for chunk in stream) + except BaseException as error: + result["error"] = error + + thread = threading.Thread(target=run) + thread.start() + thread.join(timeout=5) + try: + assert thread.is_alive() is False + if error := result.get("error"): + raise error + assert result == {"lazy": True, "text": "hello"} + assert len(requests) == 2 + assert bodies[0] == bodies[1] + assert requests[1].headers["authorization"].startswith("Payment ") + finally: + client.close() + runtime.close()