From 1a926ccc92e73c768fe368b6f4df252c6fa75295 Mon Sep 17 00:00:00 2001 From: Parv Ahuja <17094219+parvahuja@users.noreply.github.com> Date: Wed, 29 Jul 2026 12:14:39 -0700 Subject: [PATCH] feat: add owned runtime and synchronous payment transport --- .changelog/runtime-sync-http.md | 5 + README.md | 16 + pyproject.toml | 1 + src/mpp/client/__init__.py | 1 + src/mpp/client/_http.py | 74 +++- src/mpp/client/sync_transport.py | 221 ++++++++++ src/mpp/client/transport.py | 65 ++- src/mpp/runtime.py | 470 +++++++++++++++++++- tests/test_owned_runtime.py | 710 +++++++++++++++++++++++++++++++ tests/test_sync_client.py | 695 ++++++++++++++++++++++++++++++ 10 files changed, 2202 insertions(+), 56 deletions(-) create mode 100644 .changelog/runtime-sync-http.md create mode 100644 src/mpp/client/sync_transport.py create mode 100644 tests/test_owned_runtime.py create mode 100644 tests/test_sync_client.py diff --git a/.changelog/runtime-sync-http.md b/.changelog/runtime-sync-http.md new file mode 100644 index 00000000..ba506521 --- /dev/null +++ b/.changelog/runtime-sync-http.md @@ -0,0 +1,5 @@ +--- +pympp: minor +--- + +Added an owned-loop payment runtime and synchronous HTTPX payment transport. diff --git a/README.md b/README.md index 35e43626..a7d2ac62 100644 --- a/README.md +++ b/README.md @@ -78,6 +78,22 @@ If a credential is sent but its outcome cannot be confirmed, matching attempts raise `mpp.errors.PaymentOutcomeUnknownError`. Reconcile them externally before calling `runtime.reset_unknown_outcomes(reconciled=True)`. +Synchronous and mixed sync/async integrations use an explicit owned asyncio loop +runtime: + +```python +import httpx +from mpp.client import SyncPaymentTransport +from mpp.runtime import OwnedPaymentRuntime + +with OwnedPaymentRuntime([method]) as runtime: + with httpx.Client(transport=SyncPaymentTransport(runtime=runtime)) as client: + response = client.get("https://api.example.com/paid") +``` + +For loop-bound resources, pass async-context-manager factories with +`method_factories=` so they are created and closed on the owned loop. + ## Examples | Example | Description | diff --git a/pyproject.toml b/pyproject.toml index 7596fbc5..edb03e84 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,6 +5,7 @@ description = "Python SDK for the Machine Payments Protocol (MPP)" readme = "README.md" requires-python = ">=3.11" dependencies = [ + "anyio>=4,<5", "httpx>=0.27", ] authors = [ 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/_http.py b/src/mpp/client/_http.py index df61231e..5ffbf086 100644 --- a/src/mpp/client/_http.py +++ b/src/mpp/client/_http.py @@ -4,6 +4,7 @@ import hashlib import re +import threading from dataclasses import dataclass from datetime import UTC, datetime from http.cookies import CookieError, SimpleCookie @@ -20,7 +21,7 @@ if TYPE_CHECKING: from collections.abc import Sequence - from mpp.runtime import Method, PaymentRuntime + from mpp.runtime import Method, OwnedPaymentRuntime, PaymentRuntime _COOKIE_ESCAPE = re.compile(r"%[0-9a-fA-F]{2}") _PAYMENT_MARKER = "mpp.payment_attempt" @@ -80,8 +81,13 @@ def __init__(self) -> None: self._unreconciled_count = 0 self._circuit: _UnknownOutcome | None = None self._reconciliation = _Reconciliation() + self._lock = threading.RLock() def begin(self, challenge: Challenge, request: httpx.Request) -> _HttpPaymentAttempt: + with self._lock: + return self._begin(challenge, request) + + def _begin(self, challenge: Challenge, request: httpx.Request) -> _HttpPaymentAttempt: marker = request.extensions.get(_PAYMENT_MARKER) if isinstance(marker, _HttpPaymentAttempt): raise _outcome_error(marker) @@ -112,6 +118,10 @@ def begin(self, challenge: Challenge, request: httpx.Request) -> _HttpPaymentAtt return attempt def mark_sent(self, attempt: _HttpPaymentAttempt, request: httpx.Request) -> None: + with self._lock: + self._mark_sent(attempt, request) + + def _mark_sent(self, attempt: _HttpPaymentAttempt, request: httpx.Request) -> None: if self._circuit is not None: self.discard(attempt) raise _outcome_error(self._circuit) @@ -128,6 +138,14 @@ def mark_unknown( self, attempt: _HttpPaymentAttempt, cause: BaseException, + ) -> _UnknownOutcome: + with self._lock: + return self._mark_unknown(attempt, cause) + + def _mark_unknown( + self, + attempt: _HttpPaymentAttempt, + cause: BaseException, ) -> _UnknownOutcome: if attempt.unknown_outcome is not None: return attempt.unknown_outcome @@ -155,6 +173,10 @@ def mark_unknown( return outcome def complete(self, attempt: _HttpPaymentAttempt) -> None: + with self._lock: + self._complete(attempt) + + def _complete(self, attempt: _HttpPaymentAttempt) -> None: if attempt.completed or attempt.unknown_outcome is not None: return attempt.completed = True @@ -164,21 +186,25 @@ def complete(self, attempt: _HttpPaymentAttempt) -> None: request.extensions.pop(_PAYMENT_MARKER, None) def discard(self, attempt: _HttpPaymentAttempt) -> None: - if not attempt.sent: - self.complete(attempt) + with self._lock: + if not attempt.sent: + self._complete(attempt) def reset(self, *, reconciled: bool) -> None: - if not reconciled: - raise ValueError("Unknown payment outcomes must be externally reconciled before reset") - self._reconciliation.reconciled = True - self._reconciliation = _Reconciliation() - self._entries = { - key: entry - for key, entry in self._entries.items() - if isinstance(entry, _HttpPaymentAttempt) - } - self._unreconciled_count = 0 - self._circuit = None + with self._lock: + if not reconciled: + raise ValueError( + "Unknown payment outcomes must be externally reconciled before reset" + ) + self._reconciliation.reconciled = True + self._reconciliation = _Reconciliation() + self._entries = { + key: entry + for key, entry in self._entries.items() + if isinstance(entry, _HttpPaymentAttempt) + } + self._unreconciled_count = 0 + self._circuit = None def _remove(self, attempt: _HttpPaymentAttempt) -> None: for key in attempt.keys: @@ -242,6 +268,24 @@ def retry_request(self, authorization: str) -> httpx.Request: return retry +def _settle_http_payment( + attempt: _HttpPaymentAttempt, + payment: _HttpPayment, + response: httpx.Response, +) -> PaymentOutcomeUnknownError | None: + if response.status_code < 400: + attempt.complete() + return None + detail = ( + "Server returned another payment challenge after receiving a credential" + if response.status_code == 402 + else f"Credentialed request returned HTTP {response.status_code}" + ) + cause = RuntimeError(detail) + attempt.unknown(cause) + return payment.unknown(cause) + + class _AllowedOrigins: def __init__(self, allowed: Sequence[str] | None) -> None: self._allow_all = allowed is None @@ -326,7 +370,7 @@ def _challenge_is_expired(challenge: Challenge) -> bool: def _match_http_challenge( - runtime: PaymentRuntime, + runtime: PaymentRuntime | OwnedPaymentRuntime, challenges: list[Challenge], ) -> tuple[Challenge | None, Method | None]: try: diff --git a/src/mpp/client/sync_transport.py b/src/mpp/client/sync_transport.py new file mode 100644 index 00000000..54efbb77 --- /dev/null +++ b/src/mpp/client/sync_transport.py @@ -0,0 +1,221 @@ +"""Synchronous payment-aware HTTPX transport.""" + +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING, Any + +import httpx + +from mpp.client._http import ( + _PAYMENT_SENT, + _challenge_is_expired, + _failed_payload, + _HttpPayment, + _match_http_challenge, + _payment_challenges, + _propagate_response_cookies, + _response_request, + _settle_http_payment, +) +from mpp.client.transport import _EventHandlers +from mpp.errors import PaymentError, PaymentOutcomeUnknownError +from mpp.events import PAYMENT_FAILED, PAYMENT_RESPONSE, EventDispatcher +from mpp.runtime import Method, OwnedPaymentRuntime + +if TYPE_CHECKING: + from collections.abc import Sequence + +logger = logging.getLogger(__name__) + + +def _close_response(response: httpx.Response) -> None: + try: + response.close() + except BaseException: + pass + + +class SyncPaymentTransport(_EventHandlers, httpx.BaseTransport): + """HTTPX transport that synchronously handles one payment challenge.""" + + def __init__( + self, + methods: Sequence[Method] | None = None, + inner: httpx.BaseTransport | None = None, + events: EventDispatcher | None = None, + *, + runtime: OwnedPaymentRuntime | 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") + if not isinstance(runtime, OwnedPaymentRuntime): + raise TypeError("SyncPaymentTransport requires OwnedPaymentRuntime") + self._runtime = runtime + else: + if methods is None: + raise ValueError("Pass methods or runtime") + self._runtime = OwnedPaymentRuntime(methods, events=events) + self._inner = inner or httpx.HTTPTransport() + self._events = self._runtime.events + + def _fail( + self, + payment: _HttpPayment, + error: Exception, + *, + continuation: bool = False, + **details: Any, + ) -> None: + emit = self._runtime._emit_event_sync if continuation else self._runtime.emit_event_sync + emit(PAYMENT_FAILED, payment.failed(error, **details)) + + def handle_request(self, request: httpx.Request) -> httpx.Response: + """Send a request and retry one payable 402 response.""" + if request.headers.get("content-type", "").lower().startswith("multipart/form-data"): + request.read() + response = self._inner.handle_request(request) + if response.status_code != 402: + return response + request = _response_request(response, request) + if not self._runtime.allows_http_payment(request.url): + return response + payment_source = request.extensions.get(_PAYMENT_SENT) + if isinstance(payment_source, int) and payment_source != id(request): + return response + + challenges, parse_error = _payment_challenges(response) + challenge = method = None + if challenges: + try: + challenge, method = _match_http_challenge(self._runtime, challenges) + except BaseException: + _close_response(response) + raise + if challenge is None or method is None: + if parse_error is not None or challenges: + try: + self._runtime.emit_event_sync( + PAYMENT_FAILED, + _failed_payload( + challenge=None, + challenges=challenges, + credential=None, + error=parse_error + or ValueError("No compatible payment method for challenges"), + method=None, + request=request, + response=response, + ), + ) + except BaseException: + _close_response(response) + raise + return response + + payment = _HttpPayment(challenges, challenge, method, request, response) + if _challenge_is_expired(challenge): + logger.warning("Challenge expired at %s, not paying", challenge.expires) + try: + self._fail(payment, ValueError(f"Challenge expired at {challenge.expires}")) + except BaseException: + _close_response(response) + raise + return response + + try: + request.read() + except httpx.StreamConsumed as cause: + error = PaymentError( + "Streaming request bodies cannot be replayed after a payment challenge. " + "Use a buffered body for paid requests." + ) + try: + self._fail(payment, error) + finally: + _close_response(response) + raise error from cause + except BaseException: + _close_response(response) + raise + + try: + response.read() + response.close() + except BaseException: + _close_response(response) + raise + + with self._runtime._paid_operation(): + try: + attempt = self._runtime._begin_http_payment(challenge, request) + except PaymentOutcomeUnknownError as error: + self._fail(payment, error, continuation=True) + raise + + try: + credential = self._runtime._create_credential_sync( + challenge, + method, + event_payload=payment.event_payload(), + ) + authorization = credential.to_authorization() + payment.credential = credential + attempt.credential = credential + retry_request = payment.retry_request(authorization) + except BaseException as error: + attempt.discard() + if isinstance(error, Exception): + self._fail(payment, error, continuation=True) + raise + + try: + attempt.mark_sent(retry_request) + payment_response = self._inner.handle_request(retry_request) + except BaseException as cause: + if not attempt.sent: + attempt.discard() + if isinstance(cause, Exception): + self._fail(payment, cause, continuation=True) + raise + outcome = attempt.unknown(cause) + if not isinstance(cause, Exception): + raise + error = payment.unknown(outcome.cause) + self._fail(payment, error, continuation=True) + raise error from cause + + try: + if error := _settle_http_payment(attempt, payment, payment_response): + self._fail( + payment, + error, + continuation=True, + response=payment_response, + ) + if payment_response.status_code == 402: + raise error from error.cause + + _response_request(payment_response, retry_request) + _propagate_response_cookies(response, payment_response) + if payment_response.is_success: + self._runtime._emit_event_sync( + PAYMENT_RESPONSE, + payment.event_payload(payment_response), + ) + return payment_response + except BaseException as error: + if attempt.sent and not attempt.completed and attempt.unknown_outcome is None: + attempt.unknown(error) + _close_response(payment_response) + raise + + def close(self) -> None: + """Close the inner transport and an implicitly created runtime.""" + 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 a56fe683..73bed70f 100644 --- a/src/mpp/client/transport.py +++ b/src/mpp/client/transport.py @@ -24,6 +24,7 @@ _payment_challenges, _propagate_response_cookies, _response_request, + _settle_http_payment, ) from mpp.errors import PaymentError, PaymentOutcomeUnknownError from mpp.events import ( @@ -35,7 +36,7 @@ EventHandler, Unsubscribe, ) -from mpp.runtime import Method, PaymentRuntime +from mpp.runtime import Method, OwnedPaymentRuntime, PaymentRuntime logger = logging.getLogger(__name__) @@ -88,7 +89,7 @@ def __init__( inner: httpx.AsyncBaseTransport | None = None, events: EventDispatcher | None = None, *, - runtime: PaymentRuntime | None = None, + runtime: PaymentRuntime | OwnedPaymentRuntime | None = None, ) -> None: self._owns_runtime = runtime is None if runtime is not None: @@ -102,18 +103,16 @@ def __init__( self._inner = inner or httpx.AsyncHTTPTransport() self._events = self._runtime.events - async def _fail(self, payment: _HttpPayment, error: Exception, **details: Any) -> None: - await self._runtime._emit_event(PAYMENT_FAILED, payment.failed(error, **details)) - - async def _unknown( + async def _fail( self, payment: _HttpPayment, - cause: BaseException, - response: httpx.Response | None = None, - ) -> PaymentOutcomeUnknownError: - error = payment.unknown(cause) - await self._fail(payment, error, response=response) - return error + error: Exception, + *, + continuation: bool = False, + **details: Any, + ) -> None: + emit = self._runtime._emit_event if continuation else self._runtime.emit_event + await emit(PAYMENT_FAILED, payment.failed(error, **details)) async def handle_async_request(self, request: httpx.Request) -> httpx.Response: """Handle request, automatically retrying on 402 with credentials.""" @@ -136,7 +135,7 @@ async def handle_async_request(self, request: httpx.Request) -> httpx.Response: challenge = method = None if challenges: try: - self._runtime.start() + await self._runtime.astart() challenge, method = _match_http_challenge(self._runtime, challenges) except BaseException: await _close_response(response) @@ -199,7 +198,7 @@ async def handle_async_request(self, request: httpx.Request) -> httpx.Response: try: attempt = self._runtime._begin_http_payment(challenge, request) except PaymentOutcomeUnknownError as error: - await self._fail(payment, error) + await self._fail(payment, error, continuation=True) raise try: @@ -215,7 +214,7 @@ async def handle_async_request(self, request: httpx.Request) -> httpx.Response: except BaseException as error: attempt.discard() if isinstance(error, Exception): - await self._fail(payment, error) + await self._fail(payment, error, continuation=True) raise try: @@ -225,31 +224,25 @@ async def handle_async_request(self, request: httpx.Request) -> httpx.Response: if not attempt.sent: attempt.discard() if isinstance(cause, Exception): - await self._fail(payment, cause) + await self._fail(payment, cause, continuation=True) raise outcome = attempt.unknown(cause) if not isinstance(cause, Exception): raise - error = await self._unknown(payment, outcome.cause) + error = payment.unknown(outcome.cause) + await self._fail(payment, error, continuation=True) raise error from cause try: - if payment_response.status_code == 402: - cause = RuntimeError( - "Server returned another payment challenge after receiving a credential" - ) - attempt.unknown(cause) - error = await self._unknown(payment, cause, response=payment_response) - raise error from cause - - if payment_response.status_code >= 400: - cause = RuntimeError( - f"Credentialed request returned HTTP {payment_response.status_code}" + if error := _settle_http_payment(attempt, payment, payment_response): + await self._fail( + payment, + error, + continuation=True, + response=payment_response, ) - attempt.unknown(cause) - await self._unknown(payment, cause, response=payment_response) - else: - attempt.complete() + if payment_response.status_code == 402: + raise error from error.cause _response_request(payment_response, retry_request) _propagate_response_cookies(response, payment_response) @@ -286,7 +279,7 @@ def __init__( self, methods: Sequence[Method] | None = None, *, - runtime: PaymentRuntime | None = None, + runtime: PaymentRuntime | OwnedPaymentRuntime | None = None, ) -> None: self._transport = PaymentTransport(methods=methods, runtime=runtime) self._client = httpx.AsyncClient(transport=self._transport) @@ -330,7 +323,7 @@ async def request( url: str, *, methods: Sequence[Method] | None = None, - runtime: PaymentRuntime | None = None, + runtime: PaymentRuntime | OwnedPaymentRuntime | None = None, **kwargs: Any, ) -> httpx.Response: """Send an HTTP request with automatic payment handling. @@ -353,7 +346,7 @@ async def get( url: str, *, methods: Sequence[Method] | None = None, - runtime: PaymentRuntime | None = None, + runtime: PaymentRuntime | OwnedPaymentRuntime | None = None, **kwargs: Any, ) -> httpx.Response: """Send a GET request with automatic payment handling.""" @@ -364,7 +357,7 @@ async def post( url: str, *, methods: Sequence[Method] | None = None, - runtime: PaymentRuntime | None = None, + runtime: PaymentRuntime | OwnedPaymentRuntime | None = None, **kwargs: Any, ) -> httpx.Response: """Send a POST request with automatic payment handling.""" diff --git a/src/mpp/runtime.py b/src/mpp/runtime.py index b3b02d50..192141b6 100644 --- a/src/mpp/runtime.py +++ b/src/mpp/runtime.py @@ -2,11 +2,21 @@ from __future__ import annotations -from collections.abc import Mapping, Sequence -from contextlib import contextmanager -from typing import TYPE_CHECKING, Any, Protocol, Self, runtime_checkable +import asyncio +import threading +from collections.abc import Awaitable, Callable, Mapping, Sequence +from concurrent.futures import Future +from contextlib import ( + AbstractAsyncContextManager, + ExitStack, + contextmanager, +) +from contextvars import ContextVar +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any, Protocol, Self, TypeVar, runtime_checkable import httpx +from anyio.from_thread import BlockingPortal, start_blocking_portal from mpp import Challenge, Credential from mpp.events import ( @@ -18,7 +28,20 @@ if TYPE_CHECKING: from mpp.client import PaymentTransport - from mpp.client._http import _HttpPaymentAttempt + from mpp.client._http import _HttpPaymentAttempt as _Attempt + + +@dataclass(slots=True) +class _OwnedRuntimeScope: + key: object + active: bool = True + + +_OWNED_RUNTIME_SCOPES: ContextVar[tuple[_OwnedRuntimeScope, ...]] = ContextVar( + "mpp_owned_runtime_scopes", + default=(), +) +_ResultT = TypeVar("_ResultT") @runtime_checkable @@ -32,6 +55,50 @@ async def create_credential(self, challenge: Challenge) -> Credential: ... +MethodFactory = Callable[[], AbstractAsyncContextManager[Method]] + + +class _PortalError(Exception): + def __init__(self, error: BaseException) -> None: + self.error = error + + +async def _portal_call(factory: Callable[[], Awaitable[_ResultT]]) -> _ResultT: + try: + return await factory() + except BaseException as error: + if isinstance(error, (Exception, asyncio.CancelledError)): + raise + raise _PortalError(error) from None + + +async def _settle_cancelled_task(task: asyncio.Task[Any]) -> None: + while not task.done(): + try: + await asyncio.shield(task) + except asyncio.CancelledError: + continue + except BaseException: + break + if not task.cancelled(): + task.exception() + + +@contextmanager +def _owned_scope(key: object): + scope = _OwnedRuntimeScope(key) + token = _OWNED_RUNTIME_SCOPES.set((*_OWNED_RUNTIME_SCOPES.get(), scope)) + try: + yield + finally: + scope.active = False + _OWNED_RUNTIME_SCOPES.reset(token) + + +def _owned_in_scope(key: object) -> bool: + return any(scope.active and scope.key is key for scope in _OWNED_RUNTIME_SCOPES.get()) + + class PaymentRuntime: """Match methods and create credentials on the caller's event loop. @@ -191,7 +258,7 @@ def _begin_http_payment( self, challenge: Challenge, request: httpx.Request, - ) -> _HttpPaymentAttempt: + ) -> _Attempt: return self._http.begin(challenge, request) @contextmanager @@ -217,6 +284,399 @@ async def aclose(self) -> None: self.close() +class OwnedPaymentRuntime: + """Run borrowed and managed payment methods on one owned asyncio loop.""" + + def __init__( + self, + methods: Sequence[Method] = (), + *, + method_factories: Sequence[MethodFactory] = (), + events: EventDispatcher | None = None, + allowed_origins: Sequence[str] | None = None, + ) -> None: + from mpp.client._http import _AllowedOrigins + + self._borrowed = tuple(methods) + self._factories = tuple(method_factories) + self.events = events or EventDispatcher() + self._runtime_allowed_origins = allowed_origins + self._allowed_origins = _AllowedOrigins(allowed_origins) + self._changed = threading.Condition() + self._scope_key = object() + self._state = "new" + self._start_claimed = False + self._async_starters = 0 + self._leases = 0 + self._stack: ExitStack | None = None + self._portal: BlockingPortal | None = None + self._owner_thread_id: int | None = None + self._runtime: PaymentRuntime | None = None + + def _start(self, *, claim: bool) -> None: + with self._changed: + if self._state == "starting" and threading.get_ident() == self._owner_thread_id: + raise RuntimeError("Cannot start OwnedPaymentRuntime from its owned event loop") + self._start_claimed |= claim + while self._state == "starting": + self._changed.wait() + if self._state == "open": + return + if self._state != "new": + raise RuntimeError(f"OwnedPaymentRuntime is {self._state}") + self._state = "starting" + + stack = ExitStack() + try: + portal = stack.enter_context(start_blocking_portal(backend="asyncio")) + owner_thread_id = portal.call(threading.get_ident) + with self._changed: + self._owner_thread_id = owner_thread_id + managed = [] + for factory in self._factories: + context = portal.call(factory) + if not isinstance(context, AbstractAsyncContextManager): + raise TypeError("Expected an asynchronous context manager") + managed.append(stack.enter_context(portal.wrap_async_context_manager(context))) + runtime = PaymentRuntime( + (*self._borrowed, *managed), + events=self.events, + allowed_origins=self._runtime_allowed_origins, + ) + stack.callback(runtime.close) + except BaseException: + try: + stack.close() + finally: + with self._changed: + self._owner_thread_id = None + self._state = "closed" + self._changed.notify_all() + raise + + with self._changed: + self._stack = stack.pop_all() + self._portal = portal + self._owner_thread_id = owner_thread_id + self._runtime = runtime + self._state = "open" + self._changed.notify_all() + + def start(self) -> Self: + """Start the owned event loop and initialize managed methods.""" + self._start(claim=True) + return self + + __enter__ = start + + def __exit__(self, *_args: Any) -> None: + self.close() + + async def astart(self) -> Self: + """Start the runtime without blocking the caller's event loop.""" + with self._changed: + if self._state == "starting" and threading.get_ident() == self._owner_thread_id: + raise RuntimeError("Cannot start OwnedPaymentRuntime from its owned event loop") + if self._state == "open": + self._start_claimed = True + return self + self._async_starters += 1 + task = asyncio.create_task(asyncio.to_thread(self._start, claim=False)) + try: + await asyncio.shield(task) + except asyncio.CancelledError: + await _settle_cancelled_task(task) + with self._changed: + self._async_starters -= 1 + cleanup = asyncio.create_task(asyncio.to_thread(self._close, unclaimed_only=True)) + await _settle_cancelled_task(cleanup) + raise + except BaseException: + with self._changed: + self._async_starters -= 1 + raise + with self._changed: + self._async_starters -= 1 + self._start_claimed = True + return self + + __aenter__ = astart + + async def aclose(self) -> None: + """Close the runtime without blocking the caller's event loop.""" + if _owned_in_scope(self._scope_key) or threading.get_ident() == self._owner_thread_id: + self.close() + await asyncio.to_thread(self.close) + + async def __aexit__(self, *_args: Any) -> None: + await self.aclose() + + def _acquire(self, *, continuation: bool = False) -> None: + if not continuation: + self.start() + with self._changed: + if continuation: + if not _owned_in_scope(self._scope_key) or self._state not in { + "open", + "closing", + }: + raise RuntimeError("No active OwnedPaymentRuntime operation") + elif self._state != "open": + raise RuntimeError(f"OwnedPaymentRuntime is {self._state}") + self._leases += 1 + + def _release(self) -> None: + with self._changed: + self._leases -= 1 + self._changed.notify_all() + + @contextmanager + def _lease(self, *, continuation: bool = False): + self._acquire(continuation=continuation) + try: + with _owned_scope(self._scope_key): + yield + finally: + self._release() + + @contextmanager + def _paid_operation(self): + """Keep the runtime open for one complete payment flow.""" + with self._lease(): + yield + + def _submit( + self, + factory: Callable[[], Awaitable[_ResultT]], + *, + continuation: bool = False, + ) -> Future[_ResultT]: + self._acquire(continuation=continuation) + + async def run() -> _ResultT: + try: + with _owned_scope(self._scope_key): + return await _portal_call(factory) + finally: + self._release() + + try: + assert self._portal is not None + return self._portal.start_task_soon(run) + except BaseException: + self._release() + raise + + def _call( + self, + factory: Callable[[], Awaitable[_ResultT]], + *, + continuation: bool = False, + ) -> _ResultT: + if threading.get_ident() == self._owner_thread_id: + raise RuntimeError("Cannot call a sync runtime API from its owned event loop") + future = self._submit(factory, continuation=continuation) + try: + return future.result() + except _PortalError as error: + raise error.error from None + except BaseException: + future.cancel() + raise + + async def _acall( + self, + factory: Callable[[], Awaitable[_ResultT]], + *, + continuation: bool = False, + ) -> _ResultT: + if threading.get_ident() == self._owner_thread_id: + with self._lease(continuation=continuation): + return await factory() + if not continuation: + await self.astart() + future = self._submit(factory, continuation=continuation) + try: + return await asyncio.wrap_future(future) + except _PortalError as error: + raise error.error from None + + def _leased_core(self) -> PaymentRuntime: + if not _owned_in_scope(self._scope_key): + raise RuntimeError("No active OwnedPaymentRuntime operation") + assert self._runtime is not None + return self._runtime + + @property + def methods(self) -> tuple[Method, ...]: + with self._lease(): + return self._leased_core().methods + + def match_challenge( + self, + challenges: Sequence[Challenge], + *, + prefer_method_order: bool = True, + allow_name_only: bool = False, + ) -> tuple[Challenge, Method]: + with self._lease(): + return self._leased_core().match_challenge( + challenges, + prefer_method_order=prefer_method_order, + allow_name_only=allow_name_only, + ) + + async def create_credential( + self, + challenge: Challenge, + method: Method, + *, + allow_name_only: bool = False, + event_payload: dict[str, Any] | None = None, + ) -> Credential: + return await self._create_credential( + challenge, + method, + allow_name_only=allow_name_only, + event_payload=event_payload, + _continuation=False, + ) + + async def _create_credential( + self, + challenge: Challenge, + method: Method, + *, + allow_name_only: bool = False, + event_payload: dict[str, Any] | None = None, + _continuation: bool = True, + ) -> Credential: + return await self._acall( + lambda: self._leased_core()._create_credential( + challenge, + method, + allow_name_only=allow_name_only, + event_payload=event_payload, + ), + continuation=_continuation, + ) + + def create_credential_sync( + self, + challenge: Challenge, + method: Method, + *, + allow_name_only: bool = False, + event_payload: dict[str, Any] | None = None, + ) -> Credential: + return self._create_credential_sync( + challenge, + method, + allow_name_only=allow_name_only, + event_payload=event_payload, + _continuation=False, + ) + + def _create_credential_sync( + self, + challenge: Challenge, + method: Method, + *, + allow_name_only: bool = False, + event_payload: dict[str, Any] | None = None, + _continuation: bool = True, + ) -> Credential: + return self._call( + lambda: self._leased_core()._create_credential( + challenge, + method, + allow_name_only=allow_name_only, + event_payload=event_payload, + ), + continuation=_continuation, + ) + + async def emit_event(self, name: str, payload: EventPayload) -> Any: + return await self._emit_event(name, payload, _continuation=False) + + async def _emit_event( + self, + name: str, + payload: EventPayload, + *, + _continuation: bool = True, + ) -> Any: + return await self._acall( + lambda: self._leased_core()._emit_event(name, payload), + continuation=_continuation, + ) + + def emit_event_sync(self, name: str, payload: EventPayload) -> Any: + return self._emit_event_sync(name, payload, _continuation=False) + + def _emit_event_sync( + self, + name: str, + payload: EventPayload, + *, + _continuation: bool = True, + ) -> Any: + return self._call( + lambda: self._leased_core()._emit_event(name, payload), + continuation=_continuation, + ) + + def allows_http_payment(self, url: httpx.URL) -> bool: + return self._allowed_origins.allows(url) + + def reset_unknown_outcomes(self, *, reconciled: bool) -> None: + with self._lease(): + self._leased_core().reset_unknown_outcomes(reconciled=reconciled) + + def _begin_http_payment(self, challenge: Challenge, request: httpx.Request) -> _Attempt: + return self._leased_core()._begin_http_payment(challenge, request) + + def _close(self, *, unclaimed_only: bool = False) -> None: + if _owned_in_scope(self._scope_key): + raise RuntimeError("Cannot close OwnedPaymentRuntime from an active operation") + if threading.get_ident() == self._owner_thread_id: + raise RuntimeError("Cannot close OwnedPaymentRuntime from its owned event loop") + + with self._changed: + while self._state == "starting": + self._changed.wait() + if unclaimed_only and ( + self._state != "open" or self._async_starters or self._start_claimed + ): + return + while self._state == "closing": + self._changed.wait() + if self._state != "open": + self._state = "closed" + self._changed.notify_all() + return + self._state = "closing" + while self._leases: + self._changed.wait() + stack = self._stack + + assert stack is not None + try: + stack.close() + finally: + with self._changed: + self._stack = None + self._portal = None + self._owner_thread_id = None + self._runtime = None + self._state = "closed" + self._changed.notify_all() + + def close(self) -> None: + """Wait for active operations, close methods, and stop the owned loop.""" + self._close() + + def _is_method(value: Any) -> bool: return isinstance(getattr(value, "name", None), str) and callable( getattr(value, "create_credential", None) diff --git a/tests/test_owned_runtime.py b/tests/test_owned_runtime.py new file mode 100644 index 00000000..13d34bac --- /dev/null +++ b/tests/test_owned_runtime.py @@ -0,0 +1,710 @@ +"""Tests for the explicit owned-loop payment runtime.""" + +from __future__ import annotations + +import asyncio +import threading +from contextlib import asynccontextmanager +from types import MappingProxyType +from typing import Any, cast + +import httpx +import pytest + +import mpp.runtime as runtime_module +from mpp import Challenge, Credential +from mpp.client import PaymentTransport +from mpp.runtime import OwnedPaymentRuntime + + +def challenge(identifier: str = "test") -> Challenge: + return Challenge(id=identifier, method="tempo", intent="charge", request={}) + + +class MockMethod: + name = "tempo" + intents = MappingProxyType({"charge": object()}) + + def __init__(self) -> None: + self.loops: list[asyncio.AbstractEventLoop] = [] + + async def create_credential(self, challenge: Challenge) -> Credential: + self.loops.append(asyncio.get_running_loop()) + return Credential(challenge=challenge.to_echo(), payload={"ok": True}) + + +async def test_factory_sync_async_events_and_exit_share_owned_loop() -> None: + caller_loop = asyncio.get_running_loop() + loops: list[asyncio.AbstractEventLoop] = [] + events: list[str] = [] + + @asynccontextmanager + async def factory(): + events.append("enter") + loops.append(asyncio.get_running_loop()) + method = MockMethod() + try: + yield method + finally: + loops.append(asyncio.get_running_loop()) + events.append("exit") + + async with OwnedPaymentRuntime(method_factories=[factory]) as runtime: + method = cast(MockMethod, runtime.methods[0]) + runtime.events.on("*", lambda _event: loops.append(asyncio.get_running_loop())) + await runtime.create_credential(challenge("async"), method) + await asyncio.to_thread(runtime.create_credential_sync, challenge("sync"), method) + + assert events == ["enter", "exit"] + assert len(method.loops) == 2 + assert len({*loops, *method.loops}) == 1 + assert loops[0] is not caller_loop + + +async def test_borrowed_method_is_not_entered_or_closed() -> None: + events: list[str] = [] + + class BorrowedMethod(MockMethod): + async def __aenter__(self): + events.append("enter") + return self + + async def __aexit__(self, *_args: Any) -> None: + events.append("exit") + + method = BorrowedMethod() + async with OwnedPaymentRuntime([method]) as runtime: + await runtime.create_credential(challenge(), method) + + assert events == [] + + +async def test_name_only_credential_creation() -> None: + method = MockMethod() + value = Challenge(id="legacy", method="tempo", intent="subscription", request={}) + async with OwnedPaymentRuntime([method]) as runtime: + matched = runtime.match_challenge([value], allow_name_only=True) + created = await runtime.create_credential(*matched, allow_name_only=True) + created_sync = await asyncio.to_thread( + runtime.create_credential_sync, *matched, allow_name_only=True + ) + assert created.payload == created_sync.payload == {"ok": True} + + +def test_factory_failure_unwinds_and_closes_runtime() -> None: + events: list[str] = [] + + @asynccontextmanager + async def entered(): + events.append("enter") + try: + yield MockMethod() + finally: + events.append("exit") + + @asynccontextmanager + async def failed(): + raise ValueError("factory failed") + yield MockMethod() # pragma: no cover + + runtime = OwnedPaymentRuntime(method_factories=[entered, failed]) + with pytest.raises(ValueError, match="factory failed"): + runtime.start() + + assert events == ["enter", "exit"] + with pytest.raises(RuntimeError, match="closed"): + runtime.start() + + +def test_failed_start_is_closed_after_portal_stops( + monkeypatch: pytest.MonkeyPatch, +) -> None: + original = runtime_module.start_blocking_portal + exiting = threading.Event() + release = threading.Event() + + def delayed_portal(**kwargs: Any): + inner = original(**kwargs) + + class Context: + def __enter__(self): + return inner.__enter__() + + def __exit__(self, *args: Any): + exiting.set() + assert release.wait(1) + return inner.__exit__(*args) + + return Context() + + monkeypatch.setattr(runtime_module, "start_blocking_portal", delayed_portal) + + @asynccontextmanager + async def failed(): + raise ValueError("factory failed") + yield MockMethod() # pragma: no cover + + runtime = OwnedPaymentRuntime(method_factories=[failed]) + errors: list[BaseException] = [] + + def start() -> None: + try: + runtime.start() + except BaseException as error: + errors.append(error) + + starter = threading.Thread(target=start) + starter.start() + assert exiting.wait(1) + closed = threading.Event() + closer = threading.Thread(target=lambda: (runtime.close(), closed.set())) + closer.start() + assert not closed.wait(0.05) + release.set() + starter.join(timeout=1) + closer.join(timeout=1) + + assert not starter.is_alive() and not closer.is_alive() + assert len(errors) == 1 and isinstance(errors[0], ValueError) + assert runtime._state == "closed" + + +def test_factory_contract_is_strict() -> None: + runtime = OwnedPaymentRuntime( + method_factories=[lambda: MockMethod()], # type: ignore[list-item] + ) + with pytest.raises(TypeError, match="asynchronous context manager"): + runtime.start() + + +def test_factory_runtime_reentry_fails_fast() -> None: + runtimes: list[OwnedPaymentRuntime] = [] + + @asynccontextmanager + async def factory(): + runtime = runtimes[0] + assert runtime._owner_thread_id == threading.get_ident() + with pytest.raises(RuntimeError, match="owned event loop"): + runtime.start() + with pytest.raises(RuntimeError, match="owned event loop"): + await runtime.astart() + with pytest.raises(RuntimeError, match="owned event loop"): + runtime.close() + with pytest.raises(RuntimeError, match="owned event loop"): + await runtime.emit_event("nested", {}) + yield MockMethod() + + runtime = OwnedPaymentRuntime(method_factories=[factory]) + runtimes.append(runtime) + with runtime: + assert runtime.methods + + +def test_concurrent_first_start_is_shared() -> None: + entered = threading.Event() + release = threading.Event() + started: list[OwnedPaymentRuntime] = [] + + @asynccontextmanager + async def factory(): + entered.set() + await asyncio.to_thread(release.wait) + yield MockMethod() + + runtime = OwnedPaymentRuntime(method_factories=[factory]) + thread = threading.Thread(target=lambda: started.append(runtime.start())) + thread.start() + assert entered.wait(1) + waiter = threading.Thread(target=lambda: started.append(runtime.start())) + waiter.start() + try: + assert waiter.is_alive() + finally: + release.set() + thread.join(timeout=1) + waiter.join(timeout=1) + runtime.close() + assert not thread.is_alive() and not waiter.is_alive() + assert started == [runtime, runtime] + + +def test_close_during_start_and_concurrent_close_are_idempotent() -> None: + entered = threading.Event() + release = threading.Event() + + @asynccontextmanager + async def factory(): + entered.set() + await asyncio.to_thread(release.wait) + yield MockMethod() + + runtime = OwnedPaymentRuntime(method_factories=[factory]) + starter = threading.Thread(target=runtime.start) + closers = [threading.Thread(target=runtime.close) for _ in range(2)] + starter.start() + assert entered.wait(1) + for closer in closers: + closer.start() + release.set() + starter.join(timeout=1) + for closer in closers: + closer.join(timeout=1) + + assert not starter.is_alive() + assert all(not closer.is_alive() for closer in closers) + assert runtime._state == "closed" + + +def test_close_waits_for_operation_and_active_close_fails_fast() -> None: + runtime = OwnedPaymentRuntime().start() + entered = threading.Event() + release = threading.Event() + closed = threading.Event() + + def operation() -> None: + with runtime._paid_operation(): + entered.set() + assert release.wait(1) + + worker = threading.Thread(target=operation) + worker.start() + assert entered.wait(1) + closer = threading.Thread(target=lambda: (runtime.close(), closed.set())) + closer.start() + assert not closed.wait(0.05) + + release.set() + worker.join(timeout=1) + closer.join(timeout=1) + assert not worker.is_alive() and not closer.is_alive() + assert closed.is_set() + + active = OwnedPaymentRuntime().start() + with active._paid_operation(): + with pytest.raises(RuntimeError, match="active operation"): + active.close() + active.close() + + +async def test_child_task_inherits_scope_but_acquires_its_own_lease() -> None: + runtime = OwnedPaymentRuntime().start() + child_started = asyncio.Event() + child_release = asyncio.Event() + closed = threading.Event() + + async def child() -> None: + with runtime._paid_operation(): + child_started.set() + await child_release.wait() + + with runtime._paid_operation(): + task = asyncio.create_task(child()) + await child_started.wait() + + closer = threading.Thread(target=lambda: (runtime.close(), closed.set())) + closer.start() + assert not await asyncio.to_thread(closed.wait, 0.05) + child_release.set() + await task + closer.join(timeout=1) + + assert not closer.is_alive() + assert closed.is_set() + + +async def test_inherited_scope_expires_with_parent_operation() -> None: + runtime = OwnedPaymentRuntime().start() + release = asyncio.Event() + + async def close_after_parent() -> None: + await release.wait() + await asyncio.to_thread(runtime.close) + + with runtime._paid_operation(): + task = asyncio.create_task(close_after_parent()) + + release.set() + await asyncio.wait_for(task, 1) + + +def test_sync_call_from_owner_loop_fails_fast() -> None: + runtime: OwnedPaymentRuntime + + class ReentrantMethod(MockMethod): + async def create_credential(self, challenge: Challenge) -> Credential: + runtime.emit_event_sync("nested", {}) + return await super().create_credential(challenge) + + method = ReentrantMethod() + runtime = OwnedPaymentRuntime([method]).start() + try: + with pytest.raises(RuntimeError, match="owned event loop"): + runtime.create_credential_sync(challenge(), method) + finally: + runtime.close() + + +async def test_async_close_from_owner_loop_fails_fast() -> None: + runtime: OwnedPaymentRuntime + + class ReentrantMethod(MockMethod): + async def create_credential(self, challenge: Challenge) -> Credential: + await runtime.aclose() + return await super().create_credential(challenge) + + method = ReentrantMethod() + runtime = OwnedPaymentRuntime([method]).start() + try: + with pytest.raises(RuntimeError, match="active operation"): + await runtime.create_credential(challenge(), method) + finally: + runtime.close() + + +async def test_close_from_inherited_to_thread_scope_fails_fast() -> None: + runtime: OwnedPaymentRuntime + + class ReentrantMethod(MockMethod): + async def create_credential(self, challenge: Challenge) -> Credential: + await asyncio.to_thread(runtime.close) + return await super().create_credential(challenge) + + method = ReentrantMethod() + runtime = OwnedPaymentRuntime([method]).start() + try: + with pytest.raises(RuntimeError, match="active operation"): + await asyncio.wait_for(runtime.create_credential(challenge(), method), 1) + finally: + runtime.close() + + +async def test_lazy_async_http_start_does_not_block_caller_loop() -> None: + release = threading.Event() + caller_ran = threading.Event() + remained_responsive: list[bool] = [] + + @asynccontextmanager + async def factory(): + await asyncio.to_thread(release.wait) + yield MockMethod() + + def handler(request: httpx.Request) -> httpx.Response: + if "authorization" not in request.headers: + return httpx.Response( + 402, + headers={ + "www-authenticate": challenge().to_www_authenticate("example.com"), + }, + ) + return httpx.Response(200, content=b"paid") + + def watchdog() -> None: + remained_responsive.append(caller_ran.wait(0.5)) + release.set() + + runtime = OwnedPaymentRuntime(method_factories=[factory]) + transport = PaymentTransport(runtime=runtime, inner=httpx.MockTransport(handler)) + thread = threading.Thread(target=watchdog) + thread.start() + caller_loop_turn = asyncio.create_task(asyncio.sleep(0, result=caller_ran.set())) + try: + response = await transport.handle_async_request( + httpx.Request("GET", "https://example.com"), + ) + await caller_loop_turn + finally: + release.set() + thread.join() + await transport.aclose() + await runtime.aclose() + + assert response.status_code == 200 + assert remained_responsive == [True] + + +@pytest.mark.parametrize("operation", ["event", "credential"]) +async def test_lazy_async_runtime_start_does_not_block_caller_loop(operation: str) -> None: + release = threading.Event() + caller_ran = threading.Event() + remained_responsive: list[bool] = [] + method = MockMethod() + + @asynccontextmanager + async def factory(): + await asyncio.to_thread(release.wait) + yield method + + def watchdog() -> None: + remained_responsive.append(caller_ran.wait(0.5)) + release.set() + + runtime = OwnedPaymentRuntime(method_factories=[factory]) + thread = threading.Thread(target=watchdog) + thread.start() + caller_loop_turn = asyncio.create_task(asyncio.sleep(0, result=caller_ran.set())) + try: + if operation == "event": + await runtime.emit_event("test", {}) + else: + await runtime.create_credential(challenge(), method) + await caller_loop_turn + finally: + release.set() + thread.join() + await runtime.aclose() + + assert remained_responsive == [True] + + +async def test_repeatedly_cancelled_start_closes_initialized_resources() -> None: + entered = threading.Event() + release = threading.Event() + exited = threading.Event() + + @asynccontextmanager + async def factory(): + entered.set() + await asyncio.to_thread(release.wait) + try: + yield MockMethod() + finally: + exited.set() + + runtime = OwnedPaymentRuntime(method_factories=[factory]) + task = asyncio.create_task(runtime.astart()) + assert await asyncio.to_thread(entered.wait, 1) + task.cancel() + await asyncio.sleep(0) + task.cancel() + release.set() + + with pytest.raises(asyncio.CancelledError): + await task + assert runtime._state == "closed" + assert exited.is_set() + + +async def test_cancelled_shared_start_does_not_close_other_caller() -> None: + entered = threading.Event() + release = threading.Event() + + @asynccontextmanager + async def factory(): + entered.set() + await asyncio.to_thread(release.wait) + yield MockMethod() + + runtime = OwnedPaymentRuntime(method_factories=[factory]) + cancelled = asyncio.create_task(runtime.astart()) + waiting = asyncio.create_task(runtime.astart()) + assert await asyncio.to_thread(entered.wait, 1) + cancelled.cancel() + release.set() + + with pytest.raises(asyncio.CancelledError): + await cancelled + assert await waiting is runtime + assert runtime._state == "open" + runtime.close() + + +async def test_cancelled_async_start_preserves_waiting_sync_start() -> None: + entered = threading.Event() + release = threading.Event() + started: list[OwnedPaymentRuntime] = [] + + @asynccontextmanager + async def factory(): + entered.set() + await asyncio.to_thread(release.wait) + yield MockMethod() + + runtime = OwnedPaymentRuntime(method_factories=[factory]) + cancelled = asyncio.create_task(runtime.astart()) + assert await asyncio.to_thread(entered.wait, 1) + waiter = threading.Thread(target=lambda: started.append(runtime.start())) + waiter.start() + while not runtime._start_claimed: + await asyncio.sleep(0) + cancelled.cancel() + release.set() + + with pytest.raises(asyncio.CancelledError): + await cancelled + waiter.join(timeout=1) + assert not waiter.is_alive() + assert started == [runtime] + assert runtime._state == "open" + runtime.close() + + +async def test_all_cancelled_starters_close_initialized_resources() -> None: + entered = threading.Event() + release = threading.Event() + exited = threading.Event() + + @asynccontextmanager + async def factory(): + entered.set() + await asyncio.to_thread(release.wait) + try: + yield MockMethod() + finally: + exited.set() + + runtime = OwnedPaymentRuntime(method_factories=[factory]) + tasks = [asyncio.create_task(runtime.astart()) for _ in range(2)] + assert await asyncio.to_thread(entered.wait, 1) + for task in tasks: + task.cancel() + release.set() + + for task in tasks: + with pytest.raises(asyncio.CancelledError): + await task + assert runtime._state == "closed" + assert exited.is_set() + + +async def test_close_waits_while_async_payment_finishes() -> None: + started = threading.Event() + release = threading.Event() + events: list[str] = [] + + class BlockingMethod(MockMethod): + async def create_credential(self, challenge: Challenge) -> Credential: + started.set() + await asyncio.to_thread(release.wait) + return await super().create_credential(challenge) + + method = BlockingMethod() + + def handler(request: httpx.Request) -> httpx.Response: + if "authorization" not in request.headers: + return httpx.Response( + 402, + headers={ + "www-authenticate": challenge().to_www_authenticate("example.com"), + }, + ) + return httpx.Response(200, content=b"paid") + + runtime = OwnedPaymentRuntime([method]) + runtime.events.on("payment.response", lambda _payload: events.append("response")) + + async def reject_nested_operation(_payload: dict[str, Any]) -> None: + with pytest.raises(RuntimeError, match="closing"): + await runtime.emit_event("nested", {}) + + runtime.events.on("credential.created", reject_nested_operation) + transport = PaymentTransport(runtime=runtime, inner=httpx.MockTransport(handler)) + request = asyncio.create_task( + transport.handle_async_request(httpx.Request("GET", "https://example.com")) + ) + assert await asyncio.to_thread(started.wait, 1) + close = asyncio.create_task(asyncio.to_thread(runtime.close)) + while runtime._state != "closing": + await asyncio.sleep(0) + release.set() + + assert (await request).status_code == 200 + assert events == ["response"] + await close + await transport.aclose() + + +async def test_operation_started_during_close_fails_fast() -> None: + runtime = OwnedPaymentRuntime().start() + closer = threading.Thread(target=runtime.close) + + with runtime._paid_operation(): + closer.start() + while runtime._state != "closing": + await asyncio.sleep(0) + with pytest.raises(RuntimeError, match="closing"): + await asyncio.wait_for(asyncio.create_task(runtime.emit_event("late", {})), 1) + with pytest.raises(RuntimeError, match="closing"), runtime._paid_operation(): + pass + + closer.join(timeout=1) + assert not closer.is_alive() + + +def test_same_thread_operation_started_during_close_fails_fast() -> None: + runtime = OwnedPaymentRuntime().start() + closer = threading.Thread(target=runtime.close) + + with runtime._paid_operation(): + closer.start() + while runtime._state != "closing": + threading.Event().wait(0.001) + with pytest.raises(RuntimeError, match="closing"): + runtime.emit_event_sync("late", {}) + + closer.join(timeout=1) + assert not closer.is_alive() + + +def test_non_exception_base_exception_does_not_kill_portal() -> None: + class Abort(BaseException): + pass + + class AbortOnceMethod(MockMethod): + calls = 0 + + async def create_credential(self, challenge: Challenge) -> Credential: + self.calls += 1 + if self.calls == 1: + raise Abort + return await super().create_credential(challenge) + + method = AbortOnceMethod() + runtime = OwnedPaymentRuntime([method]).start() + try: + with pytest.raises(Abort): + runtime.create_credential_sync(challenge("abort"), method) + assert runtime.create_credential_sync(challenge("ok"), method).payload == {"ok": True} + finally: + runtime.close() + + +async def test_cancelled_async_call_finishes_method_cleanup_before_close() -> None: + started = threading.Event() + cleaned = threading.Event() + + class BlockingMethod(MockMethod): + async def create_credential(self, challenge: Challenge) -> Credential: + started.set() + try: + await asyncio.Event().wait() + finally: + cleaned.set() + raise AssertionError(challenge.id) + + method = BlockingMethod() + runtime = OwnedPaymentRuntime([method]).start() + task = asyncio.create_task(runtime.create_credential(challenge(), method)) + assert await asyncio.to_thread(started.wait, 1) + task.cancel() + + with pytest.raises(asyncio.CancelledError): + await task + await asyncio.to_thread(runtime.close) + assert cleaned.is_set() + + +def test_method_exit_base_exception_still_stops_runtime() -> None: + class Abort(BaseException): + pass + + @asynccontextmanager + async def factory(): + try: + yield MockMethod() + finally: + raise Abort + + runtime = OwnedPaymentRuntime(method_factories=[factory]).start() + with pytest.raises(Abort): + runtime.close() + runtime.close() diff --git a/tests/test_sync_client.py b/tests/test_sync_client.py new file mode 100644 index 00000000..1655c363 --- /dev/null +++ b/tests/test_sync_client.py @@ -0,0 +1,695 @@ +"""Tests for synchronous payment-aware HTTP clients.""" + +from __future__ import annotations + +import asyncio +import threading +from contextlib import asynccontextmanager +from http.cookies import SimpleCookie +from typing import Any, cast +from unittest.mock import AsyncMock + +import httpx +import pytest + +from mpp import Challenge, Credential +from mpp.client import SyncPaymentTransport +from mpp.errors import PaymentError, PaymentOutcomeUnknownError +from mpp.runtime import OwnedPaymentRuntime, 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) -> Credential: + 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 ConsumingTransport(MockTransport): + def __init__(self, responses: list[httpx.Response]) -> None: + super().__init__(responses) + self.bodies: list[bytes] = [] + + def handle_request(self, request: httpx.Request) -> httpx.Response: + self.requests.append(request) + self.bodies.append(b"".join(cast(httpx.SyncByteStream, request.stream))) + return self.responses.pop(0) + + +class TrackingStream(httpx.SyncByteStream): + def __init__(self, chunks: list[bytes], *, broken: bool = False) -> None: + self.chunks = chunks + self.broken = broken + self.started = False + self.closed = False + + def __iter__(self): + self.started = True + if self.broken: + raise httpx.ReadError("body lost") + yield from self.chunks + + def close(self) -> None: + self.closed = True + + +class NonSeekableFile: + def __init__(self, content: bytes) -> None: + self.content = content + + def read(self, size: int = -1) -> bytes: + if size < 0: + size = len(self.content) + chunk, self.content = self.content[:size], self.content[size:] + return chunk + + +def challenge(**overrides: Any) -> Challenge: + values = {"id": "test-id", "method": "tempo", "intent": "charge", "request": {}} + values.update(overrides) + return Challenge(**values) + + +def payment_required(**overrides: Any) -> httpx.Response: + return httpx.Response( + 402, + headers={"www-authenticate": challenge(**overrides).to_www_authenticate("example.com")}, + ) + + +def test_passes_through_free_response_without_starting_runtime() -> None: + inner = MockTransport([httpx.Response(200, content=b"ok")]) + transport = SyncPaymentTransport(methods=[], inner=inner) + runtime = transport._runtime + + response = transport.handle_request(httpx.Request("GET", "https://example.com")) + transport.close() + + assert response.content == b"ok" + assert len(inner.requests) == 1 + with pytest.raises(RuntimeError, match="closed"): + runtime.start() + + +def test_paid_retry_applies_and_propagates_challenge_cookies() -> None: + requests: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + requests.append(request) + if "authorization" not in request.headers: + return httpx.Response( + 402, + headers=[ + ("www-authenticate", challenge().to_www_authenticate("example.com")), + ("set-cookie", "session=new; Path=/"), + ("set-cookie", "payment_nonce=nonce-1; Path=/"), + ], + ) + return httpx.Response( + 200, + headers={"set-cookie": "final_cookie=ok; Path=/"}, + content=b"paid", + ) + + transport = SyncPaymentTransport(methods=[MockMethod()], inner=httpx.MockTransport(handler)) + with httpx.Client(transport=transport) as client: + client.cookies.set("session", "old", domain="example.com", path="/") + response = client.get("https://example.com/paid") + + retry_cookies = SimpleCookie(requests[1].headers["cookie"]) + assert response.status_code == 200 + assert retry_cookies["session"].value == "new" + assert retry_cookies["payment_nonce"].value == "nonce-1" + assert dict(client.cookies) == { + "session": "new", + "payment_nonce": "nonce-1", + "final_cookie": "ok", + } + assert response.headers.get_list("set-cookie") == [ + "session=new; Path=/", + "payment_nonce=nonce-1; Path=/", + "final_cookie=ok; Path=/", + ] + + +@pytest.mark.parametrize( + "request_value", + [ + pytest.param( + httpx.Request("POST", "https://example.com", content=b'{"hello":"world"}'), + id="bytes", + ), + pytest.param( + httpx.Request( + "POST", + "https://example.com", + files={"file": ("hello.txt", b"hello", "text/plain")}, + ), + id="multipart", + ), + ], +) +def test_replays_buffered_request_bodies(request_value: httpx.Request) -> None: + inner = ConsumingTransport([payment_required(), httpx.Response(200, content=b"paid")]) + transport = SyncPaymentTransport(methods=[MockMethod()], inner=inner) + try: + response = transport.handle_request(request_value) + finally: + transport.close() + + assert response.status_code == 200 + assert inner.bodies[1] == inner.bodies[0] + assert inner.requests[1].headers["authorization"].startswith("Payment ") + + +def test_replays_non_seekable_multipart_file() -> None: + inner = ConsumingTransport([payment_required(), httpx.Response(200, content=b"paid")]) + transport = SyncPaymentTransport(methods=[MockMethod()], inner=inner) + request = httpx.Request( + "POST", + "https://example.com", + files={ + "file": ( + "hello.txt", + cast(Any, NonSeekableFile(b"FILE-CONTENT")), + "text/plain", + ) + }, + ) + try: + response = transport.handle_request(request) + finally: + transport.close() + + assert response.status_code == 200 + assert inner.bodies[0] == inner.bodies[1] + assert b"FILE-CONTENT" in inner.bodies[1] + + +def test_free_generator_body_passes_through() -> None: + received: list[bytes] = [] + + def handler(request: httpx.Request) -> httpx.Response: + received.append(b"".join(cast(httpx.SyncByteStream, request.stream))) + return httpx.Response(200, content=b"ok") + + transport = SyncPaymentTransport(methods=[MockMethod()], inner=httpx.MockTransport(handler)) + try: + response = transport.handle_request( + httpx.Request("POST", "https://example.com", content=iter([b"one-shot"])) + ) + finally: + transport.close() + + assert response.status_code == 200 + assert received == [b"one-shot"] + + +def test_paid_generator_body_fails_and_closes_challenge_response() -> None: + response_stream = TrackingStream([b"payment explanation"]) + + def body(): + yield b"one-shot" + + class OneShotTransport(httpx.BaseTransport): + def handle_request(self, request: httpx.Request) -> httpx.Response: + _ = b"".join(cast(httpx.SyncByteStream, request.stream)) + return httpx.Response( + 402, + headers={"www-authenticate": challenge().to_www_authenticate("example.com")}, + stream=response_stream, + ) + + transport = SyncPaymentTransport(methods=[MockMethod()], inner=OneShotTransport()) + try: + with pytest.raises(PaymentError, match="cannot be replayed"): + transport.handle_request(httpx.Request("POST", "https://example.com", content=body())) + finally: + transport.close() + + assert response_stream.closed + + +@pytest.mark.parametrize("terminal", ["complete", "error", "close"]) +def test_success_status_completes_payment_before_body(terminal: str) -> None: + stream = TrackingStream([b"paid"], broken=terminal == "error") + method = MockMethod() + runtime = OwnedPaymentRuntime([method]) + transport = SyncPaymentTransport( + runtime=runtime, + inner=MockTransport( + [ + payment_required(), + httpx.Response(200, stream=stream), + payment_required(), + httpx.Response(200, content=b"again"), + ] + ), + ) + response = transport.handle_request(httpx.Request("GET", "https://example.com")) + + assert not stream.started + try: + if terminal == "complete": + assert response.read() == b"paid" + elif terminal == "error": + with pytest.raises(httpx.ReadError, match="body lost"): + response.read() + else: + response.close() + again = transport.handle_request(httpx.Request("GET", "https://example.com")) + assert again.content == b"again" + assert method.create_credential.await_count == 2 + finally: + response.close() + transport.close() + runtime.close() + + +def test_redirect_cannot_trigger_a_second_payment() -> None: + requests: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + requests.append(request) + if request.url.host == "start.test": + if "authorization" not in request.headers: + return payment_required(id="first") + return httpx.Response(302, headers={"location": "https://next.test/resource"}) + return payment_required(id="second") + + method = MockMethod() + transport = SyncPaymentTransport(methods=[method], inner=httpx.MockTransport(handler)) + with httpx.Client(transport=transport, follow_redirects=True) as client: + response = client.get("https://start.test/resource") + + assert response.status_code == 402 + assert [request.url.host for request in requests] == ["start.test", "start.test", "next.test"] + assert "authorization" not in requests[-1].headers + method.create_credential.assert_awaited_once() + + +@pytest.mark.parametrize( + ("header", "failed_events"), + [ + pytest.param("Bearer realm=test", 0, id="non-payment"), + pytest.param("Payment invalid-base64!!", 1, id="malformed"), + pytest.param( + challenge(method="stripe").to_www_authenticate("example.com"), + 1, + id="unsupported", + ), + pytest.param( + challenge(expires="2020-01-01T00:00:00Z").to_www_authenticate("example.com"), + 1, + id="expired", + ), + ], +) +def test_nonpayable_402_remains_lazy(header: str, failed_events: int) -> None: + stream = TrackingStream([b"explanation"]) + response = httpx.Response(402, headers={"www-authenticate": header}, stream=stream) + transport = SyncPaymentTransport( + methods=[MockMethod()], + inner=MockTransport([response]), + ) + failed: list[dict[str, Any]] = [] + transport.on_payment_failed(failed.append) + try: + returned = transport.handle_request(httpx.Request("GET", "https://example.com")) + assert returned is response + assert not stream.started and not stream.closed + if header.startswith("Bearer "): + assert transport._runtime._state == "new" + assert returned.read() == b"explanation" + finally: + transport.close() + + assert len(failed) == failed_events + + +def test_disallowed_402_remains_lazy() -> None: + stream = TrackingStream([b"explanation"]) + response = httpx.Response( + 402, + headers={"www-authenticate": challenge().to_www_authenticate("example.com")}, + stream=stream, + ) + runtime = OwnedPaymentRuntime( + [MockMethod()], + allowed_origins=["https://allowed.example"], + ) + transport = SyncPaymentTransport(runtime=runtime, inner=MockTransport([response])) + try: + returned = transport.handle_request(httpx.Request("GET", "https://disallowed.example")) + assert returned is response + assert not stream.started and not stream.closed + assert runtime._state == "new" + finally: + response.close() + transport.close() + runtime.close() + + +def test_runtime_start_failure_closes_challenge_response() -> None: + stream = TrackingStream([b"payment explanation"]) + + @asynccontextmanager + async def failed_factory(): + raise RuntimeError("factory failed") + yield MockMethod() # pragma: no cover + + response = httpx.Response( + 402, + headers={ + "www-authenticate": challenge().to_www_authenticate("example.com"), + }, + stream=stream, + ) + runtime = OwnedPaymentRuntime(method_factories=[failed_factory]) + transport = SyncPaymentTransport(runtime=runtime, inner=MockTransport([response])) + try: + with pytest.raises(RuntimeError, match="factory failed"): + transport.handle_request(httpx.Request("GET", "https://example.com")) + finally: + transport.close() + runtime.close() + + assert stream.closed + + +@pytest.mark.parametrize("stage", ["credential", "serialization", "retry"]) +def test_payment_failures_emit_and_propagate( + stage: str, + monkeypatch: pytest.MonkeyPatch, +) -> None: + error = RuntimeError(f"{stage} failed") + requests: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + requests.append(request) + if len(requests) == 1: + return payment_required() + raise error + + method = MockMethod() + if stage == "credential": + method.create_credential.side_effect = error + elif stage == "serialization": + monkeypatch.setattr( + Credential, "to_authorization", lambda _self: (_ for _ in ()).throw(error) + ) + + transport = SyncPaymentTransport(methods=[method], inner=httpx.MockTransport(handler)) + failed: list[dict[str, Any]] = [] + transport.on_payment_failed(failed.append) + try: + expected = PaymentOutcomeUnknownError if stage == "retry" else RuntimeError + with pytest.raises(expected): + transport.handle_request(httpx.Request("GET", "https://example.com")) + finally: + transport.close() + + assert len(failed) == 1 + assert isinstance(failed[0]["error"], expected) + if stage == "retry": + assert failed[0]["credential"] is not None + else: + assert failed[0]["credential"] is None + + +def test_repeated_402_after_credential_is_unknown() -> None: + method = MockMethod() + transport = SyncPaymentTransport( + methods=[method], + inner=MockTransport([payment_required(), payment_required()]), + ) + try: + with pytest.raises(PaymentOutcomeUnknownError, match="Do not blindly retry"): + transport.handle_request(httpx.Request("GET", "https://example.com")) + finally: + transport.close() + + method.create_credential.assert_awaited_once() + + +def test_send_boundary_failure_reports_retained_credential() -> None: + method = MockMethod() + runtime = OwnedPaymentRuntime([method]) + retained = make_credential({"hash": "0xretained"}, challenge_id="blocker") + request = httpx.Request( + "POST", + "https://example.com", + content=b"same operation", + ) + + def retain_unknown(_payload: dict[str, Any]) -> None: + blocker_request = httpx.Request( + "POST", + request.url, + content=b"same operation", + ) + blocker = runtime._begin_http_payment( + challenge(id="blocker"), + blocker_request, + ) + blocker.credential = retained + blocker.mark_sent(blocker_request) + blocker.unknown(TimeoutError("response lost")) + + failed: list[dict[str, Any]] = [] + runtime.events.on("credential.created", retain_unknown) + runtime.events.on("payment.failed", failed.append) + inner = MockTransport([payment_required()]) + transport = SyncPaymentTransport(runtime=runtime, inner=inner) + try: + with pytest.raises(PaymentOutcomeUnknownError) as raised: + transport.handle_request(request) + finally: + transport.close() + runtime.close() + + assert raised.value.credential is retained + assert failed[0]["credential"] is retained + assert len(inner.requests) == 1 + + +@pytest.mark.parametrize("status", [200, 402, 503]) +def test_event_abort_closes_unreturned_response_and_preserves_portal(status: int) -> None: + class Abort(BaseException): + pass + + stream = TrackingStream([b"body"]) + runtime = OwnedPaymentRuntime([MockMethod()]) + transport = SyncPaymentTransport( + runtime=runtime, + inner=MockTransport([payment_required(), httpx.Response(status, stream=stream)]), + ) + event = "payment.response" if status == 200 else "payment.failed" + unsubscribe = runtime.events.on(event, lambda _payload: (_ for _ in ()).throw(Abort())) + try: + with pytest.raises(Abort): + transport.handle_request(httpx.Request("GET", "https://example.com")) + unsubscribe() + assert runtime.emit_event_sync("still.alive", {}) is None + finally: + transport.close() + runtime.close() + + assert stream.closed + + +def test_event_abort_closes_unpayable_challenge_response() -> None: + class Abort(BaseException): + pass + + stream = TrackingStream([b"explanation"]) + transport = SyncPaymentTransport( + methods=[MockMethod()], + inner=MockTransport( + [ + httpx.Response( + 402, + headers={"www-authenticate": "Payment invalid-base64!!"}, + stream=stream, + ) + ] + ), + ) + transport.on_payment_failed(lambda _payload: (_ for _ in ()).throw(Abort())) + try: + with pytest.raises(Abort): + transport.handle_request(httpx.Request("GET", "https://example.com")) + finally: + transport.close() + + assert stream.closed + + +def test_handler_can_supply_credential() -> None: + method = MockMethod() + credential = make_credential({"hash": "0xevent"}, challenge_id="test-id") + inner = MockTransport([payment_required(), httpx.Response(200, content=b"paid")]) + transport = SyncPaymentTransport(methods=[method], inner=inner) + events: list[str] = [] + transport.on_challenge_received(lambda _payload: credential) + transport.on_credential_created(lambda _payload: events.append("credential")) + transport.on_payment_response(lambda _payload: events.append("response")) + try: + response = transport.handle_request(httpx.Request("GET", "https://example.com")) + finally: + transport.close() + + assert response.status_code == 200 + assert events == ["credential", "response"] + method.create_credential.assert_not_called() + + +def test_transport_ownership_and_validation() -> None: + with pytest.raises(ValueError, match="methods or runtime"): + SyncPaymentTransport() + + plain = PaymentRuntime() + with pytest.raises(TypeError, match="OwnedPaymentRuntime"): + SyncPaymentTransport(runtime=plain) # type: ignore[arg-type] + + runtime = OwnedPaymentRuntime() + with pytest.raises(ValueError, match="either methods/events or runtime"): + SyncPaymentTransport(methods=[], runtime=runtime) + + inner = MockTransport([]) + borrowed = SyncPaymentTransport(runtime=runtime, inner=inner) + runtime.start() + borrowed.close() + assert runtime.emit_event_sync("still.open", {}) is None + runtime.close() + + +def test_owned_runtime_closes_when_inner_close_fails() -> None: + error = RuntimeError("inner close failed") + + class FailingCloseTransport(MockTransport): + def close(self) -> None: + super().close() + raise error + + transport = SyncPaymentTransport(methods=[], inner=FailingCloseTransport([])) + runtime = transport._runtime + with pytest.raises(RuntimeError) as raised: + transport.close() + + assert raised.value is error + with pytest.raises(RuntimeError, match="closed"): + runtime.start() + + +def test_close_waits_for_committed_retry() -> None: + retry_started = threading.Event() + retry_release = threading.Event() + closed = threading.Event() + events: list[str] = [] + calls = 0 + + def handler(_request: httpx.Request) -> httpx.Response: + nonlocal calls + calls += 1 + if calls == 1: + return payment_required() + retry_started.set() + assert retry_release.wait(1) + return httpx.Response(200, content=b"paid") + + runtime = OwnedPaymentRuntime([MockMethod()]) + runtime.events.on("payment.response", lambda _payload: events.append("response")) + transport = SyncPaymentTransport(runtime=runtime, inner=httpx.MockTransport(handler)) + responses: list[httpx.Response] = [] + request_thread = threading.Thread( + target=lambda: responses.append( + transport.handle_request(httpx.Request("GET", "https://example.com")) + ) + ) + request_thread.start() + assert retry_started.wait(1) + close_thread = threading.Thread(target=lambda: (runtime.close(), closed.set())) + close_thread.start() + assert not closed.wait(0.05) + + retry_release.set() + request_thread.join(timeout=1) + close_thread.join(timeout=1) + transport.close() + + assert responses[0].status_code == 200 + assert closed.is_set() + assert events == ["response"] + + +def test_concurrent_sync_clients_share_atomic_ledger() -> None: + method = MockMethod() + runtime = OwnedPaymentRuntime([method]).start() + barrier = threading.Barrier(2) + duplicate_seen = threading.Event() + begin = runtime._begin_http_payment + + def synchronized_begin(value: Challenge, request: httpx.Request): + barrier.wait(timeout=1) + try: + attempt = begin(value, request) + except PaymentOutcomeUnknownError: + duplicate_seen.set() + raise + assert duplicate_seen.wait(1) + return attempt + + runtime._begin_http_payment = synchronized_begin # type: ignore[method-assign] + transports = [ + SyncPaymentTransport( + runtime=runtime, + inner=MockTransport([payment_required(), httpx.Response(200, content=b"paid")]), + ) + for _ in range(2) + ] + responses: list[httpx.Response] = [] + errors: list[BaseException] = [] + + def send(transport: SyncPaymentTransport) -> None: + try: + responses.append(transport.handle_request(httpx.Request("GET", "https://example.com"))) + except BaseException as error: + errors.append(error) + + threads = [threading.Thread(target=send, args=(transport,)) for transport in transports] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=1) + assert not thread.is_alive() + for transport in transports: + transport.close() + runtime.close() + + assert len(responses) == 1 + assert responses[0].status_code == 200 + assert len(errors) == 1 + assert isinstance(errors[0], PaymentOutcomeUnknownError) + method.create_credential.assert_awaited_once()