diff --git a/agentscore_commerce/stripe_multichain/__init__.py b/agentscore_commerce/stripe_multichain/__init__.py index d97bbae..8385784 100644 --- a/agentscore_commerce/stripe_multichain/__init__.py +++ b/agentscore_commerce/stripe_multichain/__init__.py @@ -5,23 +5,16 @@ create_mppx_stripe, ) from agentscore_commerce.stripe_multichain.payment_intent import ( - CreateMultichainPaymentIntentInput, MultichainPaymentIntentResult, StripeClientLike, create_multichain_payment_intent, get_deposit_address, ) -from agentscore_commerce.stripe_multichain.pi_cache import ( - PiCache, - PiCacheOptions, - create_pi_cache, -) +from agentscore_commerce.stripe_multichain.pi_cache import PiCache, create_pi_cache from agentscore_commerce.stripe_multichain.simulate_deposit import ( DEFAULT_BUYER_WALLET, STRIPE_TEST_TX_HASH_FAILED, STRIPE_TEST_TX_HASH_SUCCESS, - SimulateCryptoDepositInput, - SimulateDepositIfTestModeInput, simulate_crypto_deposit, simulate_deposit_if_test_mode, ) @@ -31,12 +24,8 @@ "DEFAULT_PAYMENT_METHOD_TYPES", "STRIPE_TEST_TX_HASH_FAILED", "STRIPE_TEST_TX_HASH_SUCCESS", - "CreateMultichainPaymentIntentInput", "MultichainPaymentIntentResult", "PiCache", - "PiCacheOptions", - "SimulateCryptoDepositInput", - "SimulateDepositIfTestModeInput", "StripeClientLike", "create_mppx_stripe", "create_multichain_payment_intent", diff --git a/agentscore_commerce/stripe_multichain/payment_intent.py b/agentscore_commerce/stripe_multichain/payment_intent.py index 3e4368e..d4b2aed 100644 --- a/agentscore_commerce/stripe_multichain/payment_intent.py +++ b/agentscore_commerce/stripe_multichain/payment_intent.py @@ -4,7 +4,7 @@ chains, returning the PI id + deposit addresses per network. Distinct from the Stripe SPT flow. """ -from dataclasses import dataclass, field +from dataclasses import dataclass from typing import Any, Protocol @@ -16,44 +16,46 @@ class StripeClientLike(Protocol): payment_intents: StripePaymentIntentsAPI -@dataclass -class CreateMultichainPaymentIntentInput: - stripe: Any # StripeClientLike, but kept loose so vendors can pass their actual `stripe.StripeClient` - amount: int # in cents (Stripe convention) - currency: str = "usd" - networks: list[str] = field(default_factory=lambda: ["tempo", "base", "solana"]) - metadata: dict[str, str] | None = None - idempotency_key: str | None = None - - @dataclass class MultichainPaymentIntentResult: payment_intent_id: str deposit_addresses: dict[str, str] -def create_multichain_payment_intent(input: CreateMultichainPaymentIntentInput) -> MultichainPaymentIntentResult: +_DEFAULT_NETWORKS: tuple[str, ...] = ("tempo", "base", "solana") + + +def create_multichain_payment_intent( + *, + stripe: Any, # StripeClientLike, kept loose so vendors can pass their actual `stripe.StripeClient` + amount: int, # in cents (Stripe convention) + currency: str = "usd", + networks: list[str] | None = None, + metadata: dict[str, str] | None = None, + idempotency_key: str | None = None, +) -> MultichainPaymentIntentResult: """Create a Stripe PaymentIntent with multichain crypto deposit_options. Returns the PI id + per-network deposit addresses. Raises if Stripe doesn't return any addresses. """ + resolved_networks = list(networks) if networks else list(_DEFAULT_NETWORKS) params: dict[str, Any] = { - "amount": input.amount, - "currency": input.currency, + "amount": amount, + "currency": currency, "payment_method_types": ["crypto"], "payment_method_data": {"type": "crypto"}, "payment_method_options": { - "crypto": {"mode": "deposit", "deposit_options": {"networks": input.networks}}, + "crypto": {"mode": "deposit", "deposit_options": {"networks": resolved_networks}}, }, "confirm": True, } - if input.metadata: - params["metadata"] = input.metadata + if metadata: + params["metadata"] = metadata pi = ( - input.stripe.payment_intents.create(params, idempotency_key=input.idempotency_key) - if input.idempotency_key - else input.stripe.payment_intents.create(params) + stripe.payment_intents.create(params, idempotency_key=idempotency_key) + if idempotency_key + else stripe.payment_intents.create(params) ) deposit_addresses: dict[str, str] = {} next_action = getattr(pi, "next_action", None) or (pi.get("next_action") if isinstance(pi, dict) else None) diff --git a/agentscore_commerce/stripe_multichain/pi_cache.py b/agentscore_commerce/stripe_multichain/pi_cache.py index 251ca9b..6380a9e 100644 --- a/agentscore_commerce/stripe_multichain/pi_cache.py +++ b/agentscore_commerce/stripe_multichain/pi_cache.py @@ -52,19 +52,6 @@ class _Entry(Generic[T]): expires_at: float -@dataclass -class PiCacheOptions: - """Optional configuration for :func:`create_pi_cache`.""" - - #: Redis connection URL (e.g. ``rediss://…cache.amazonaws.com:6379``). When omitted, - #: the cache falls back to in-process dicts with the same API. - redis_url: str | None = None - #: TTL for cached entries in seconds. Default 300. - ttl_seconds: int = 300 - #: Prefix for Redis keys. Default ``'payto:'``. - key_prefix: str = "payto:" - - @dataclass class PiCache: """Stripe PI + deposit-address cache produced by :func:`create_pi_cache`.""" @@ -78,17 +65,25 @@ class PiCache: stop: Callable[[], None] -def create_pi_cache(opts: PiCacheOptions | None = None) -> PiCache: +def create_pi_cache( + *, + redis_url: str | None = None, + ttl_seconds: int = 300, + key_prefix: str = "payto:", +) -> PiCache: """Construct a Stripe PI + deposit-address cache instance. Returns a ``PiCache`` with async ``cache_address`` / ``has_address`` (Redis-backed when ``redis_url`` is set) and sync helpers for PI-id and network-address lookup. A background task evicts expired in-memory entries every 60 seconds; call ``stop()`` from server shutdown handlers to cancel it. + + ``redis_url`` — connection URL (e.g. ``rediss://…cache.amazonaws.com:6379``); when + omitted, the cache falls back to in-process dicts with the same API. + ``ttl_seconds`` — entry TTL (default 300). + ``key_prefix`` — Redis key prefix (default ``'payto:'``). """ - options = opts or PiCacheOptions() - ttl = options.ttl_seconds - key_prefix = options.key_prefix + ttl = ttl_seconds redis_client: _RedisLike | None = None addr_mem_cache: dict[str, float] = {} @@ -97,7 +92,7 @@ def create_pi_cache(opts: PiCacheOptions | None = None) -> PiCache: async def _get_redis() -> _RedisLike | None: nonlocal redis_client - if not options.redis_url: + if not redis_url: return None if redis_client is not None: return redis_client @@ -112,7 +107,7 @@ async def _get_redis() -> _RedisLike | None: "[pi-cache] redis_url set but `redis` is not installed. Run `pip install redis` or unset redis_url." ) return None - redis_client = redis_asyncio.from_url(options.redis_url) + redis_client = redis_asyncio.from_url(redis_url) return redis_client async def cache_address(address: str) -> None: diff --git a/agentscore_commerce/stripe_multichain/simulate_deposit.py b/agentscore_commerce/stripe_multichain/simulate_deposit.py index 3e79b41..74b557e 100644 --- a/agentscore_commerce/stripe_multichain/simulate_deposit.py +++ b/agentscore_commerce/stripe_multichain/simulate_deposit.py @@ -2,7 +2,6 @@ import logging from collections.abc import Callable -from dataclasses import dataclass, field from typing import Literal import httpx @@ -28,57 +27,52 @@ STRIPE_TEST_TX_HASH_FAILED = "0x000000000000000000000000000000000000000000000000000000testfailed" -@dataclass -class SimulateCryptoDepositInput: - payment_intent_id: str - network: Literal["tempo", "base", "solana"] - stripe_secret_key: str - buyer_wallet: str | None = None - token_currency: str | None = None - transaction_hash: str | None = None - stripe_version: str | None = None - stripe_api_base: str = "https://api.stripe.com" - extra: dict[str, str] = field(default_factory=dict) - - -async def simulate_crypto_deposit(input: SimulateCryptoDepositInput) -> None: +async def simulate_crypto_deposit( + *, + payment_intent_id: str, + network: Literal["tempo", "base", "solana"], + stripe_secret_key: str, + buyer_wallet: str | None = None, + token_currency: str | None = None, + transaction_hash: str | None = None, + stripe_version: str | None = None, + stripe_api_base: str = "https://api.stripe.com", + extra: dict[str, str] | None = None, +) -> None: """Call Stripe's `test_helpers/payment_intents/{id}/simulate_crypto_deposit` endpoint.""" - url = f"{input.stripe_api_base}/v1/test_helpers/payment_intents/{input.payment_intent_id}/simulate_crypto_deposit" + url = f"{stripe_api_base}/v1/test_helpers/payment_intents/{payment_intent_id}/simulate_crypto_deposit" params: dict[str, str] = { - "network": input.network, - "buyer_wallet": input.buyer_wallet or DEFAULT_BUYER_WALLET.get(input.network, ""), + "network": network, + "buyer_wallet": buyer_wallet or DEFAULT_BUYER_WALLET.get(network, ""), } - if input.token_currency: - params["token_currency"] = input.token_currency - if input.transaction_hash: - params["transaction_hash"] = input.transaction_hash - params.update(input.extra) + if token_currency: + params["token_currency"] = token_currency + if transaction_hash: + params["transaction_hash"] = transaction_hash + if extra: + params.update(extra) headers: dict[str, str] = { - "Authorization": f"Bearer {input.stripe_secret_key}", + "Authorization": f"Bearer {stripe_secret_key}", "Content-Type": "application/x-www-form-urlencoded", } - if input.stripe_version: - headers["Stripe-Version"] = input.stripe_version + if stripe_version: + headers["Stripe-Version"] = stripe_version async with httpx.AsyncClient() as client: res = await client.post(url, headers=headers, content="&".join(f"{k}={v}" for k, v in params.items())) if res.status_code >= 300: raise RuntimeError(f"Stripe simulate_crypto_deposit failed: {res.status_code} {res.text}") -@dataclass -class SimulateDepositIfTestModeInput: - """Input for :func:`simulate_deposit_if_test_mode`.""" - - get_payment_intent_id: Callable[[str], str | None] - deposit_address: str - network: Literal["tempo", "base", "solana"] - stripe_secret_key: str - buyer_wallet: str | None = None - token_currency: str = "usdc" - stripe_version: str | None = None - - -async def simulate_deposit_if_test_mode(input: SimulateDepositIfTestModeInput) -> None: +async def simulate_deposit_if_test_mode( + *, + get_payment_intent_id: Callable[[str], str | None], + deposit_address: str, + network: Literal["tempo", "base", "solana"], + stripe_secret_key: str, + buyer_wallet: str | None = None, + token_currency: str = "usdc", # noqa: S107 — literal default, not a secret + stripe_version: str | None = None, +) -> None: """Higher-level wrapper around :func:`simulate_crypto_deposit` for the testnet/dev path. Bundles the three steps every Stripe-multichain merchant repeats: @@ -94,34 +88,32 @@ async def simulate_deposit_if_test_mode(input: SimulateDepositIfTestModeInput) - Use case is exclusively dev/testnet end-to-end — production servers (sk_live_) no-op. """ - if not input.stripe_secret_key.startswith("sk_test_"): + if not stripe_secret_key.startswith("sk_test_"): return - pi_id = input.get_payment_intent_id(input.deposit_address) + pi_id = get_payment_intent_id(deposit_address) if not pi_id: logger.warning( "[stripe] Skipping deposit simulation — no PI cached for deposit address %s… (network=%s). " "The PI cache TTL may have expired between 402 emission and settlement.", - input.deposit_address[:10], - input.network, + deposit_address[:10], + network, ) return try: await simulate_crypto_deposit( - SimulateCryptoDepositInput( - payment_intent_id=pi_id, - network=input.network, - stripe_secret_key=input.stripe_secret_key, - buyer_wallet=input.buyer_wallet, - token_currency=input.token_currency, - transaction_hash=STRIPE_TEST_TX_HASH_SUCCESS, - stripe_version=input.stripe_version, - ) + payment_intent_id=pi_id, + network=network, + stripe_secret_key=stripe_secret_key, + buyer_wallet=buyer_wallet, + token_currency=token_currency, + transaction_hash=STRIPE_TEST_TX_HASH_SUCCESS, + stripe_version=stripe_version, ) - logger.warning("[stripe] ✓ Simulated %s deposit for PI %s", input.network, pi_id) + logger.warning("[stripe] ✓ Simulated %s deposit for PI %s", network, pi_id) except Exception as err: logger.error( "[stripe] ✗ Failed to simulate %s deposit for PI %s: %s", - input.network, + network, pi_id, err, ) diff --git a/examples/multi_rail_merchant.py b/examples/multi_rail_merchant.py index 8a51903..74deaf6 100644 --- a/examples/multi_rail_merchant.py +++ b/examples/multi_rail_merchant.py @@ -58,8 +58,6 @@ verify_x402_request, ) from agentscore_commerce.stripe_multichain import ( - PiCacheOptions, - SimulateDepositIfTestModeInput, create_pi_cache, simulate_deposit_if_test_mode, ) @@ -75,7 +73,7 @@ # Singleton Stripe PI / deposit-address cache. Backed by Redis when REDIS_URL is set # (multi-instance deployments need this so a deposit lands on whichever instance # settles it); falls back to in-process dict for single-instance dev. -pi_cache = create_pi_cache(PiCacheOptions(redis_url=os.environ.get("REDIS_URL"))) +pi_cache = create_pi_cache(redis_url=os.environ.get("REDIS_URL")) app = FastAPI() _gate = AgentScoreGate( @@ -168,12 +166,10 @@ async def purchase(request: Request, assess: dict = Depends(get_agentscore_data) # Fire Stripe testnet sim; no-ops on live keys. x402 settle only ever # lands on base in 1.4+ (Solana moved to MPP `solana/charge`). await simulate_deposit_if_test_mode( - SimulateDepositIfTestModeInput( - get_payment_intent_id=pi_cache.get_payment_intent_id, - deposit_address=verified.signed_pay_to, - network="base", - stripe_secret_key=os.environ["STRIPE_SECRET_KEY"], - ) + get_payment_intent_id=pi_cache.get_payment_intent_id, + deposit_address=verified.signed_pay_to, + network="base", + stripe_secret_key=os.environ["STRIPE_SECRET_KEY"], ) headers: dict[str, str] = {} diff --git a/examples/stripe_multichain_merchant.py b/examples/stripe_multichain_merchant.py index 5a36641..8607c43 100644 --- a/examples/stripe_multichain_merchant.py +++ b/examples/stripe_multichain_merchant.py @@ -24,8 +24,6 @@ from agentscore_commerce.stripe_multichain import ( STRIPE_TEST_TX_HASH_SUCCESS, - CreateMultichainPaymentIntentInput, - SimulateCryptoDepositInput, create_multichain_payment_intent, get_deposit_address, simulate_crypto_deposit, @@ -40,13 +38,11 @@ async def buy(body: dict): # Create a multichain PaymentIntent — Stripe issues deposit addresses for each requested chain. result = create_multichain_payment_intent( - CreateMultichainPaymentIntentInput( - stripe=stripe_client, - amount=body.get("amount_cents", 25000), - networks=["tempo", "base", "solana"], - metadata={"order_id": body.get("order_id", "ord_demo"), "merchant": "example"}, - idempotency_key=body.get("order_id"), - ) + stripe=stripe_client, + amount=body.get("amount_cents", 25000), + networks=["tempo", "base", "solana"], + metadata={"order_id": body.get("order_id", "ord_demo"), "merchant": "example"}, + idempotency_key=body.get("order_id"), ) base_addr = get_deposit_address(result, "base") @@ -65,13 +61,11 @@ async def buy(body: dict): @app.post("/testnet/simulate-deposit") async def simulate_deposit(body: dict): await simulate_crypto_deposit( - SimulateCryptoDepositInput( - payment_intent_id=body["payment_intent_id"], - network=body["network"], - stripe_secret_key=os.environ["STRIPE_SECRET_KEY"], - stripe_version="2026-03-04.preview", - token_currency="usdc", - transaction_hash=STRIPE_TEST_TX_HASH_SUCCESS, - ) + payment_intent_id=body["payment_intent_id"], + network=body["network"], + stripe_secret_key=os.environ["STRIPE_SECRET_KEY"], + stripe_version="2026-03-04.preview", + token_currency="usdc", + transaction_hash=STRIPE_TEST_TX_HASH_SUCCESS, ) return {"ok": True, "simulated": True} diff --git a/tests/test_lifted_helpers.py b/tests/test_lifted_helpers.py index 763fc41..f9078af 100644 --- a/tests/test_lifted_helpers.py +++ b/tests/test_lifted_helpers.py @@ -26,8 +26,6 @@ from agentscore_commerce.stripe_multichain import ( STRIPE_TEST_TX_HASH_FAILED, STRIPE_TEST_TX_HASH_SUCCESS, - PiCacheOptions, - SimulateDepositIfTestModeInput, create_pi_cache, simulate_deposit_if_test_mode, ) @@ -39,7 +37,7 @@ @pytest.mark.asyncio async def test_pi_cache_address_round_trip(): - cache = create_pi_cache(PiCacheOptions(ttl_seconds=10)) + cache = create_pi_cache(ttl_seconds=10) assert await cache.has_address("0xdeadbeef") is False await cache.cache_address("0xdeadbeef") assert await cache.has_address("0xdeadbeef") is True @@ -47,7 +45,7 @@ async def test_pi_cache_address_round_trip(): def test_pi_cache_payment_intent_round_trip(): - cache = create_pi_cache(PiCacheOptions(ttl_seconds=10)) + cache = create_pi_cache(ttl_seconds=10) assert cache.get_payment_intent_id("0xaddr") is None cache.cache_payment_intent("0xaddr", "pi_test_123") assert cache.get_payment_intent_id("0xaddr") == "pi_test_123" @@ -55,7 +53,7 @@ def test_pi_cache_payment_intent_round_trip(): def test_pi_cache_network_addresses_round_trip(): - cache = create_pi_cache(PiCacheOptions(ttl_seconds=10)) + cache = create_pi_cache(ttl_seconds=10) cache.cache_network_addresses("pi_test", {"base": "0xbase", "solana": "Gso1ana"}) assert cache.get_network_deposit_address("pi_test", "base") == "0xbase" assert cache.get_network_deposit_address("pi_test", "solana") == "Gso1ana" @@ -65,7 +63,7 @@ def test_pi_cache_network_addresses_round_trip(): def test_pi_cache_ttl_eviction_via_expired_entries(): - cache = create_pi_cache(PiCacheOptions(ttl_seconds=0)) + cache = create_pi_cache(ttl_seconds=0) cache.cache_payment_intent("0xaddr", "pi_short") # ttl=0 means expires_at == now; subsequent get returns None time.sleep(0.01) @@ -94,12 +92,10 @@ async def test_pi_cache_no_redis_url_falls_back_to_memory_only(): async def test_simulate_deposit_skips_on_live_key(): called: list[str] = [] await simulate_deposit_if_test_mode( - SimulateDepositIfTestModeInput( - get_payment_intent_id=lambda addr: called.append(addr) or "pi_x", - deposit_address="0xaddr", - network="base", - stripe_secret_key="sk_live_real_one", - ) + get_payment_intent_id=lambda addr: called.append(addr) or "pi_x", + deposit_address="0xaddr", + network="base", + stripe_secret_key="sk_live_real_one", ) # Should never even look up the PI on a live key assert called == [] @@ -108,12 +104,10 @@ async def test_simulate_deposit_skips_on_live_key(): @pytest.mark.asyncio async def test_simulate_deposit_no_pi_warns_and_returns(): await simulate_deposit_if_test_mode( - SimulateDepositIfTestModeInput( - get_payment_intent_id=lambda _addr: None, - deposit_address="0xaddr", - network="base", - stripe_secret_key="sk_test_xyz", - ) + get_payment_intent_id=lambda _addr: None, + deposit_address="0xaddr", + network="base", + stripe_secret_key="sk_test_xyz", ) # No exception; warning logged (not asserted here, would need caplog) diff --git a/tests/test_stripe_multichain.py b/tests/test_stripe_multichain.py index 13234a5..e0e5f3a 100644 --- a/tests/test_stripe_multichain.py +++ b/tests/test_stripe_multichain.py @@ -3,8 +3,6 @@ import respx from agentscore_commerce.stripe_multichain import ( - CreateMultichainPaymentIntentInput, - SimulateCryptoDepositInput, create_multichain_payment_intent, get_deposit_address, simulate_crypto_deposit, @@ -42,9 +40,7 @@ def test_create_multichain_payment_intent_extracts_addresses(): }, } api = _FakeAPI(response) - result = create_multichain_payment_intent( - CreateMultichainPaymentIntentInput(stripe=_FakeClient(api), amount=10000, idempotency_key="k1") - ) + result = create_multichain_payment_intent(stripe=_FakeClient(api), amount=10000, idempotency_key="k1") assert result.payment_intent_id == "pi_123" assert result.deposit_addresses == {"tempo": "0xtempo", "base": "0xbase", "solana": "solanaaddr"} assert api.last_idem == "k1" @@ -54,9 +50,7 @@ def test_create_multichain_payment_intent_extracts_addresses(): def test_create_multichain_payment_intent_raises_when_no_addresses(): response = {"id": "pi_x", "next_action": None} with pytest.raises(RuntimeError, match="No deposit addresses"): - create_multichain_payment_intent( - CreateMultichainPaymentIntentInput(stripe=_FakeClient(_FakeAPI(response)), amount=100) - ) + create_multichain_payment_intent(stripe=_FakeClient(_FakeAPI(response)), amount=100) def test_get_deposit_address_returns_per_network(): @@ -73,9 +67,7 @@ async def test_simulate_crypto_deposit_calls_test_helpers_endpoint(): return_value=httpx.Response(200, text="{}") ) await simulate_crypto_deposit( - SimulateCryptoDepositInput( - payment_intent_id="pi_1", network="base", stripe_secret_key="sk_test_x", token_currency="usdc" - ) + payment_intent_id="pi_1", network="base", stripe_secret_key="sk_test_x", token_currency="usdc" ) assert route.called @@ -86,9 +78,55 @@ async def test_simulate_crypto_deposit_raises_on_non_2xx(): return_value=httpx.Response(400, text='{"error":"bad"}') ) with pytest.raises(RuntimeError, match="failed: 400"): - await simulate_crypto_deposit( - SimulateCryptoDepositInput(payment_intent_id="pi_2", network="base", stripe_secret_key="sk_test_x") - ) + await simulate_crypto_deposit(payment_intent_id="pi_2", network="base", stripe_secret_key="sk_test_x") + + +@respx.mock +async def test_simulate_crypto_deposit_includes_transaction_hash_stripe_version_and_extra(): + """Optional kwargs (`transaction_hash`, `stripe_version`, `extra`) reach the wire.""" + route = respx.post("https://api.stripe.com/v1/test_helpers/payment_intents/pi_3/simulate_crypto_deposit").mock( + return_value=httpx.Response(200, text="{}") + ) + await simulate_crypto_deposit( + payment_intent_id="pi_3", + network="base", + stripe_secret_key="sk_test_x", + token_currency="usdc", + transaction_hash="0xabc", + stripe_version="2024-04-10", + extra={"description": "smoke"}, + ) + assert route.called + body = route.calls.last.request.content.decode() + assert "transaction_hash=0xabc" in body + assert "description=smoke" in body + assert route.calls.last.request.headers.get("Stripe-Version") == "2024-04-10" + + +async def test_simulate_deposit_if_test_mode_logs_and_swallows_errors(caplog): + """If `simulate_crypto_deposit` raises, the wrapper logs the failure and returns.""" + import logging + + from agentscore_commerce.stripe_multichain import simulate_deposit_if_test_mode + + async def _raises(**_kwargs): + raise RuntimeError("simulated boom") + + import agentscore_commerce.stripe_multichain.simulate_deposit as mod + + original = mod.simulate_crypto_deposit + mod.simulate_crypto_deposit = _raises # type: ignore[assignment] + try: + with caplog.at_level(logging.ERROR, logger="agentscore_commerce.stripe_multichain"): + await simulate_deposit_if_test_mode( + get_payment_intent_id=lambda _addr: "pi_xerr", + deposit_address="0xaddr", + network="base", + stripe_secret_key="sk_test_x", + ) + assert any("Failed to simulate base deposit for PI pi_xerr" in r.message for r in caplog.records) + finally: + mod.simulate_crypto_deposit = original # type: ignore[assignment] # ── create_mppx_stripe: the pympp wrapper ───────────────────────────────────