diff --git a/.changelog/quiet-cows-laugh.md b/.changelog/quiet-cows-laugh.md new file mode 100644 index 0000000..6be59af --- /dev/null +++ b/.changelog/quiet-cows-laugh.md @@ -0,0 +1,5 @@ +--- +pympp: minor +--- + +Added Stripe payment method (`mpp.methods.stripe`) supporting the Shared Payment Token (SPT) flow for HTTP 402 authentication. Includes client-side `StripeMethod` and `stripe()` factory, server-side `ChargeIntent` for PaymentIntent verification via Stripe SDK or raw HTTP, Pydantic schemas, and a `stripe` optional dependency group. diff --git a/examples/README.md b/examples/README.md index efed72b..c59a5fa 100644 --- a/examples/README.md +++ b/examples/README.md @@ -8,6 +8,7 @@ Code examples for using the Machine Payments Protocol (pympp). |---------|-------------| | [fetch/](fetch/) | CLI tool for fetching URLs with automatic payment handling | | [mcp-server/](mcp-server/) | MCP server with payment-protected tools | +| [stripe/](stripe/) | Stripe SPT payment flow (server + headless client) | ## Documentation Examples diff --git a/examples/stripe/README.md b/examples/stripe/README.md new file mode 100644 index 0000000..369f1c1 --- /dev/null +++ b/examples/stripe/README.md @@ -0,0 +1,84 @@ +# Stripe Example + +A pay-per-fortune API using Stripe's Shared Payment Token (SPT) flow. + +## What This Demonstrates + +- Server-side payment protection with `Mpp.create()` and the Stripe method +- SPT proxy endpoint (secret key stays server-side) +- Headless client using a test card (`pm_card_visa`) +- Full 402 → challenge → credential → retry flow + +## Prerequisites + +- Python 3.12+ +- [uv](https://docs.astral.sh/uv/) (recommended) or pip +- A Stripe test-mode secret key (`sk_test_...`) + +## Installation + +```bash +cd examples/stripe +uv sync +``` + +## Running + +**Start the server:** + +```bash +export STRIPE_SECRET_KEY=sk_test_... +uv run python server.py +``` + +The server starts at http://localhost:8000. + +**Run the client** (in another terminal): + +```bash +uv run python client.py +# 🥠 A smooth long journey! Great expectations. +# 📝 Receipt: pi_3Q... +``` + +## Testing Manually + +**Without payment** (returns 402): + +```bash +curl -i http://localhost:8000/api/fortune +# HTTP/1.1 402 Payment Required +# WWW-Authenticate: Payment ... +``` + +## How It Works + +``` +Client Server Stripe + │ │ │ + │ GET /api/fortune │ │ + ├──────────────────────────────> │ │ + │ │ │ + │ 402 + WWW-Authenticate │ │ + │<────────────────────────────── │ │ + │ │ │ + │ POST /api/create-spt │ │ + ├──────────────────────────────> │ Create SPT (test helper) │ + │ ├─────────────────────────────> │ + │ spt_... │ │ + │<────────────────────────────── │<───────────────────────────── │ + │ │ │ + │ GET /api/fortune │ │ + │ Authorization: Payment │ │ + ├──────────────────────────────> │ PaymentIntent (SPT + confirm)│ + │ ├─────────────────────────────> │ + │ │ pi_... succeeded │ + │ 200 + fortune + receipt │<───────────────────────────── │ + │<────────────────────────────── │ │ +``` + +1. Client requests the fortune → server returns 402 with a payment challenge +2. pympp client calls `create_token` → POSTs to `/api/create-spt` → server creates SPT via Stripe +3. Client retries with a credential containing the SPT +4. Server creates a PaymentIntent with `shared_payment_granted_token` and `confirm=True` +5. On success, returns the fortune with a receipt diff --git a/examples/stripe/client.py b/examples/stripe/client.py new file mode 100644 index 0000000..c45b5ac --- /dev/null +++ b/examples/stripe/client.py @@ -0,0 +1,78 @@ +#!/usr/bin/env python3 +"""CLI client that pays for a fortune using Stripe SPTs. + +Uses a test card (pm_card_visa) for headless operation — no browser needed. + +Usage: + export STRIPE_SECRET_KEY=sk_test_... + python client.py [--server http://localhost:8000] +""" + +import argparse +import asyncio +import sys + +import httpx + +from mpp.client import Client +from mpp.methods.stripe import stripe + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + prog="stripe-fortune", + description="Fetch a fortune with automatic Stripe payment", + ) + parser.add_argument( + "--server", + default="http://localhost:8000", + help="Server base URL (default: http://localhost:8000)", + ) + return parser.parse_args() + + +async def run(args: argparse.Namespace) -> int: + server_url = args.server.rstrip("/") + + async def create_token(params): + """Proxy SPT creation through the server.""" + async with httpx.AsyncClient() as http: + response = await http.post( + f"{server_url}/api/create-spt", + json={ + "paymentMethod": params.payment_method, + "amount": params.amount, + "currency": params.currency, + "expiresAt": params.expires_at, + "networkId": params.network_id, + "metadata": params.metadata, + }, + ) + response.raise_for_status() + return response.json()["spt"] + + method = stripe( + create_token=create_token, + payment_method="pm_card_visa", + intents={}, + ) + + async with Client(methods=[method]) as client: + response = await client.get(f"{server_url}/api/fortune") + + if response.status_code >= 400: + print(f"Error {response.status_code}: {response.text}", file=sys.stderr) + return 1 + + data = response.json() + print(f"🥠 {data['fortune']}") + print(f"📝 Receipt: {data['receipt']}") + return 0 + + +def main() -> None: + sys.exit(asyncio.run(run(parse_args()))) + + +if __name__ == "__main__": + main() diff --git a/examples/stripe/pyproject.toml b/examples/stripe/pyproject.toml new file mode 100644 index 0000000..9931ce7 --- /dev/null +++ b/examples/stripe/pyproject.toml @@ -0,0 +1,14 @@ +[project] +name = "stripe-example" +version = "0.1.0" +description = "Stripe payment example using the Machine Payments Protocol" +requires-python = ">=3.12" +dependencies = [ + "pympp[stripe,server]", + "fastapi", + "uvicorn", + "httpx", +] + +[tool.uv.sources] +pympp = { path = "../.." } diff --git a/examples/stripe/server.py b/examples/stripe/server.py new file mode 100644 index 0000000..3adb16a --- /dev/null +++ b/examples/stripe/server.py @@ -0,0 +1,108 @@ +"""Stripe payment-protected API server. + +Demonstrates the Machine Payments Protocol with Stripe's Shared Payment +Token (SPT) flow. Two endpoints: + +- POST /api/create-spt — proxy for SPT creation (requires secret key) +- GET /api/fortune — paid endpoint ($1.00 per fortune) +""" + +import base64 +import os +import random + +from fastapi import FastAPI, Request +from fastapi.responses import JSONResponse + +from mpp import Challenge +from mpp.methods.stripe import ChargeIntent, stripe +from mpp.server import Mpp + +app = FastAPI(title="Stripe Fortune Server") + +SECRET_KEY = os.environ["STRIPE_SECRET_KEY"] + +server = Mpp.create( + method=stripe( + network_id=os.environ.get("STRIPE_NETWORK_ID", "internal"), + payment_method_types=["card"], + currency="usd", + decimals=2, + recipient=os.environ.get("STRIPE_ACCOUNT", "acct_default"), + intents={"charge": ChargeIntent(secret_key=SECRET_KEY)}, + ), +) + +FORTUNES = [ + "A beautiful, smart, and loving person will come into your life.", + "A dubious friend may be an enemy in camouflage.", + "A faithful friend is a strong defense.", + "A fresh start will put you on your way.", + "A golden egg of opportunity falls into your lap this month.", + "A good time to finish up old tasks.", + "A light heart carries you through all the hard times ahead.", + "A smooth long journey! Great expectations.", +] + + +@app.post("/api/create-spt") +async def create_spt(request: Request): + """Proxy endpoint for SPT creation. + + The client calls this with a payment method ID and challenge details. + We call Stripe's test SPT endpoint using our secret key. + """ + body = await request.json() + + params = { + "payment_method": body["paymentMethod"], + "usage_limits[currency]": body["currency"], + "usage_limits[max_amount]": body["amount"], + "usage_limits[expires_at]": str(body["expiresAt"]), + } + + import httpx + + auth_value = base64.b64encode(f"{SECRET_KEY}:".encode()).decode() + async with httpx.AsyncClient() as client: + response = await client.post( + "https://api.stripe.com/v1/test_helpers/shared_payment/granted_tokens", + headers={ + "Authorization": f"Basic {auth_value}", + "Content-Type": "application/x-www-form-urlencoded", + }, + data=params, + ) + if not response.is_success: + return JSONResponse(status_code=502, content=response.json()) + result = response.json() + + return {"spt": result["id"]} + + +@app.get("/api/fortune") +async def fortune(request: Request): + """Paid endpoint — returns a fortune for $1.00.""" + result = await server.charge( + authorization=request.headers.get("Authorization"), + amount="1", + ) + + if isinstance(result, Challenge): + return JSONResponse( + status_code=402, + content={"error": "Payment required"}, + headers={"WWW-Authenticate": result.to_www_authenticate(server.realm)}, + ) + + credential, receipt = result + return { + "fortune": random.choice(FORTUNES), + "receipt": receipt.reference, + } + + +if __name__ == "__main__": + import uvicorn + + uvicorn.run(app, host="0.0.0.0", port=8000) diff --git a/pyproject.toml b/pyproject.toml index 275151f..f66e07c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -27,6 +27,9 @@ tempo = [ "pytempo>=0.2.1", "pydantic>=2.0", ] +stripe = [ + "pydantic>=2.0", +] server = ["pydantic>=2.0", "python-dotenv>=1.0"] mcp = ["mcp>=1.1.0"] dev = [ diff --git a/src/mpp/_defaults.py b/src/mpp/_defaults.py new file mode 100644 index 0000000..5960e37 --- /dev/null +++ b/src/mpp/_defaults.py @@ -0,0 +1,4 @@ +"""Shared defaults for all payment methods.""" + +DEFAULT_TIMEOUT = 30.0 +"""HTTP request timeout in seconds.""" diff --git a/src/mpp/methods/stripe/__init__.py b/src/mpp/methods/stripe/__init__.py new file mode 100644 index 0000000..11d5b26 --- /dev/null +++ b/src/mpp/methods/stripe/__init__.py @@ -0,0 +1,41 @@ +"""Stripe payment method for HTTP 402 authentication. + +Uses Stripe's Shared Payment Token (SPT) flow for one-time charges. + +Example: + # Client-side + from mpp.client import get + from mpp.methods.stripe import stripe, ChargeIntent + + async def create_spt(params): + # Proxy to your server endpoint that creates an SPT + ... + return spt_token + + response = await get( + "https://api.example.com/resource", + methods=[stripe( + create_token=create_spt, + payment_method="pm_card_visa", + intents={}, + )], + ) + + # Server-side + from mpp.server import Mpp + from mpp.methods.stripe import stripe, ChargeIntent + + server = Mpp.create( + method=stripe( + network_id="bn_...", + payment_method_types=["card"], + currency="usd", + decimals=2, + intents={"charge": ChargeIntent(secret_key="sk_...")}, + ), + ) +""" + +from mpp.methods.stripe.client import StripeMethod, stripe +from mpp.methods.stripe.intents import ChargeIntent +from mpp.methods.stripe.schemas import ChargeRequest, StripeCredentialPayload diff --git a/src/mpp/methods/stripe/_defaults.py b/src/mpp/methods/stripe/_defaults.py new file mode 100644 index 0000000..588fe39 --- /dev/null +++ b/src/mpp/methods/stripe/_defaults.py @@ -0,0 +1,3 @@ +"""Default constants for Stripe payment method.""" + +STRIPE_API_BASE = "https://api.stripe.com/v1" diff --git a/src/mpp/methods/stripe/client.py b/src/mpp/methods/stripe/client.py new file mode 100644 index 0000000..c3e22eb --- /dev/null +++ b/src/mpp/methods/stripe/client.py @@ -0,0 +1,222 @@ +"""Stripe payment method for client-side credential creation. + +Implements the charge client method using Stripe's Shared Payment Token (SPT) flow. +""" + +from __future__ import annotations + +import math +import time +from collections.abc import Awaitable, Callable +from dataclasses import dataclass, field +from typing import TYPE_CHECKING, Any + +from mpp import Challenge, Credential + +if TYPE_CHECKING: + from mpp import Credential as CredentialType + from mpp.server.intent import Intent + + +@dataclass(frozen=True) +class OnChallengeParameters: + """Parameters passed to the ``create_token`` callback. + + Attributes: + amount: Payment amount in smallest currency unit (e.g. ``"150"`` for $1.50). + challenge: The full payment challenge from the server. + currency: Three-letter ISO currency code (e.g. ``"usd"``). + expires_at: SPT expiration as a Unix timestamp (seconds). + metadata: Optional metadata from the challenge's methodDetails. + network_id: Stripe Business Network profile ID. + payment_method: Stripe payment method ID (e.g. ``"pm_card_visa"``). + """ + + amount: str + challenge: Challenge + currency: str + expires_at: int + metadata: dict[str, str] | None + network_id: str + payment_method: str + + +CreateTokenFn = Callable[[OnChallengeParameters], Awaitable[str]] + + +@dataclass +class StripeMethod: + """Stripe payment method implementation. + + Handles client-side credential creation for Stripe SPT payments. + """ + + name: str = "stripe" + create_token: CreateTokenFn | None = None + payment_method: str | None = None + external_id: str | None = None + currency: str | None = None + decimals: int = 2 + recipient: str | None = None + network_id: str | None = None + payment_method_types: list[str] = field(default_factory=lambda: ["card"]) + _intents: dict[str, Intent] = field(default_factory=dict) + + @property + def intents(self) -> dict[str, Intent]: + """Available intents for this method.""" + return self._intents + + def transform_request( + self, request: dict[str, Any], credential: CredentialType | None + ) -> dict[str, Any]: + """Inject Stripe-specific methodDetails into the challenge request. + + Called by ``Mpp`` before challenge creation to add ``networkId`` + and ``paymentMethodTypes`` to the request's ``methodDetails``. + """ + method_details = dict(request.get("methodDetails", {})) + if self.network_id and "networkId" not in method_details: + method_details["networkId"] = self.network_id + if self.payment_method_types and "paymentMethodTypes" not in method_details: + method_details["paymentMethodTypes"] = self.payment_method_types + request = {**request, "methodDetails": method_details} + return request + + async def create_credential(self, challenge: Challenge) -> Credential: + """Create a credential to satisfy the given challenge. + + Calls the user-supplied ``create_token`` callback to obtain an SPT, + then wraps it in a credential for the Authorization header. + + Args: + challenge: The payment challenge from the server. + + Returns: + A credential with the SPT payload. + + Raises: + ValueError: If no ``create_token`` callback or ``payment_method`` is configured. + """ + if self.create_token is None: + raise ValueError("No create_token callback configured") + + request = challenge.request + method_details = request.get("methodDetails", {}) + + payment_method = self.payment_method + if not payment_method: + raise ValueError("payment_method is required (pass to stripe() or via context)") + + amount = str(request.get("amount", "")) + currency = str(request.get("currency", "")) + network_id = method_details.get("networkId") if isinstance(method_details, dict) else None + if not network_id: + raise ValueError("networkId is required in challenge.methodDetails") + metadata = method_details.get("metadata") if isinstance(method_details, dict) else None + if isinstance(metadata, dict) and "externalId" in metadata: + raise ValueError( + "methodDetails.metadata.externalId is reserved; use credential externalId instead" + ) + + if challenge.expires: + expires_at = math.floor(_parse_iso_timestamp(challenge.expires)) + else: + expires_at = math.floor(time.time()) + 3600 + + spt = await self.create_token( + OnChallengeParameters( + amount=amount, + challenge=challenge, + currency=currency, + expires_at=expires_at, + metadata=metadata, + network_id=network_id, + payment_method=payment_method, + ) + ) + + payload: dict[str, Any] = {"spt": spt} + if self.external_id: + payload["externalId"] = self.external_id + + return Credential( + challenge=challenge.to_echo(), + payload=payload, + ) + + +def _parse_iso_timestamp(iso_str: str) -> float: + """Parse an ISO 8601 timestamp to a Unix timestamp (seconds).""" + from datetime import datetime + + dt = datetime.fromisoformat(iso_str.replace("Z", "+00:00")) + return dt.timestamp() + + +# ────────────────────────────────────────────────────────────────── +# Factory +# ────────────────────────────────────────────────────────────────── + + +def stripe( + intents: dict[str, Intent], + create_token: CreateTokenFn | None = None, + payment_method: str | None = None, + external_id: str | None = None, + currency: str | None = None, + decimals: int = 2, + recipient: str | None = None, + network_id: str | None = None, + payment_method_types: list[str] | None = None, +) -> StripeMethod: + """Create a Stripe payment method. + + Args: + intents: Intents to register (e.g. ``{"charge": ChargeIntent(...)}``) + create_token: Callback to create an SPT (client-side). + Receives :class:`OnChallengeParameters` and returns the SPT string. + payment_method: Default Stripe payment method ID (e.g. ``"pm_card_visa"``). + external_id: Optional client-side external reference ID. + currency: Default ISO currency code (e.g. ``"usd"``). + decimals: Decimal places for the currency (default: 2 for USD cents). + recipient: Optional default recipient. + network_id: Stripe Business Network profile ID. Included in + challenge ``methodDetails.networkId``. + payment_method_types: Stripe payment method types (default: ``["card"]``). + Included in challenge ``methodDetails.paymentMethodTypes``. + + Returns: + A configured :class:`StripeMethod` instance. + + Example: + from mpp.methods.stripe import stripe, ChargeIntent + + # Server + method = stripe( + network_id="bn_...", + payment_method_types=["card"], + currency="usd", + decimals=2, + intents={"charge": ChargeIntent(secret_key="sk_...")}, + ) + + # Client + method = stripe( + create_token=my_spt_proxy, + payment_method="pm_card_visa", + intents={"charge": ChargeIntent(secret_key="sk_...")}, + ) + """ + method = StripeMethod( + create_token=create_token, + payment_method=payment_method, + external_id=external_id, + currency=currency, + decimals=decimals, + recipient=recipient, + network_id=network_id, + payment_method_types=payment_method_types or ["card"], + ) + method._intents = dict(intents) + return method diff --git a/src/mpp/methods/stripe/intents.py b/src/mpp/methods/stripe/intents.py new file mode 100644 index 0000000..ceec2c4 --- /dev/null +++ b/src/mpp/methods/stripe/intents.py @@ -0,0 +1,278 @@ +"""Stripe payment intents (server-side verification). + +Implements the charge intent using Stripe's Shared Payment Token (SPT) flow. +""" + +from __future__ import annotations + +import base64 +from datetime import UTC, datetime +from typing import Any + +from mpp import Credential, Receipt +from mpp._defaults import DEFAULT_TIMEOUT +from mpp.errors import ( + PaymentActionRequiredError, + PaymentExpiredError, + VerificationFailedError, +) +from mpp.methods.stripe._defaults import STRIPE_API_BASE +from mpp.methods.stripe.schemas import StripeCredentialPayload + + +def _build_analytics(credential: Credential) -> dict[str, str]: + """Build MPP analytics metadata for the Stripe PaymentIntent.""" + challenge = credential.challenge + analytics: dict[str, str] = { + "mpp_challenge_id": challenge.id, + "mpp_intent": challenge.intent, + "mpp_is_mpp": "true", + "mpp_server_id": challenge.realm, + "mpp_version": "1", + } + if credential.source: + analytics["mpp_client_id"] = credential.source + return analytics + + +def _resolve_payment_intents(client: Any) -> Any: + """Resolve the payment_intents accessor from a Stripe client. + + Supports both the modern ``StripeClient`` (``client.v1.payment_intents``) + and legacy/custom clients (``client.payment_intents``). + """ + v1 = getattr(client, "v1", None) + if v1 is not None: + pi = getattr(v1, "payment_intents", None) + if pi is not None: + return pi + pi = getattr(client, "payment_intents", None) + if pi is not None: + return pi + raise TypeError("Unsupported Stripe client: expected .v1.payment_intents or .payment_intents") + + +class ChargeIntent: + """Stripe charge intent for one-time payments via SPTs. + + Verifies payment by creating a Stripe PaymentIntent with the + client-supplied Shared Payment Token (SPT). + + Accepts either a ``client`` (a pre-configured Stripe SDK instance) + or a raw ``secret_key``. Using ``client`` is recommended. + + Example: + import stripe as stripe_sdk + from mpp.methods.stripe import ChargeIntent + + client = stripe_sdk.StripeClient("sk_...") + intent = ChargeIntent(client=client) + + # Or with a raw secret key (no Stripe SDK needed): + intent = ChargeIntent(secret_key="sk_...") + """ + + name = "charge" + + def __init__( + self, + client: Any | None = None, + secret_key: str | None = None, + http_client: Any | None = None, + timeout: float = DEFAULT_TIMEOUT, + ) -> None: + """Initialize the charge intent. + + Args: + client: Pre-configured Stripe SDK instance (duck-typed). + Supports both ``StripeClient`` (v8+, ``client.v1.payment_intents``) + and legacy clients (``client.payment_intents``). + secret_key: Stripe secret API key for raw HTTP verification. + Used only when ``client`` is not provided. + http_client: Optional httpx client for raw HTTP calls. + If provided, the caller is responsible for closing it. + timeout: Request timeout in seconds (default: 30). + + Raises: + ValueError: If neither ``client`` nor ``secret_key`` is provided. + """ + if client is None and secret_key is None: + raise ValueError("Either client or secret_key is required") + self._client = client + self._secret_key = secret_key + self._http_client = http_client + self._owns_client = http_client is None + self._timeout = timeout + + async def __aenter__(self) -> ChargeIntent: + """Enter async context, creating HTTP client if needed.""" + if self._client is None: + await self._get_http_client() + return self + + async def __aexit__(self, *args: Any) -> None: + """Exit async context, closing owned HTTP client.""" + await self.aclose() + + async def aclose(self) -> None: + """Close the HTTP client if we own it.""" + if self._owns_client and self._http_client is not None: + await self._http_client.aclose() + self._http_client = None + + async def _get_http_client(self) -> Any: + """Get or create an httpx async client.""" + if self._http_client is None: + import httpx + + self._http_client = httpx.AsyncClient(timeout=self._timeout) + return self._http_client + + async def verify( + self, + credential: Credential, + request: dict[str, Any], + ) -> Receipt: + """Verify a Stripe charge credential. + + Creates a Stripe PaymentIntent using the SPT from the credential + payload, then checks that payment succeeded. + + Args: + credential: The payment credential from the client. + request: The original payment request parameters. + + Returns: + A receipt indicating success. + + Raises: + VerificationFailedError: If the SPT is missing or PaymentIntent fails. + PaymentExpiredError: If the challenge has expired. + PaymentActionRequiredError: If 3DS or other action is needed. + """ + challenge = credential.challenge + + if challenge.expires: + expires = datetime.fromisoformat(challenge.expires.replace("Z", "+00:00")) + if expires < datetime.now(UTC): + raise PaymentExpiredError(challenge.expires) + + try: + parsed = StripeCredentialPayload.model_validate(credential.payload) + except Exception as err: + raise VerificationFailedError( + "Invalid credential payload: missing or malformed spt" + ) from err + + spt = parsed.spt + credential_external_id = parsed.externalId + + user_metadata = request.get("methodDetails", {}).get("metadata") + resolved_metadata = {**_build_analytics(credential), **(user_metadata or {})} + + if self._client is not None: + pi = await self._create_with_client( + client=self._client, + challenge_id=challenge.id, + request=request, + spt=spt, + metadata=resolved_metadata, + ) + else: + pi = await self._create_with_secret_key( + secret_key=self._secret_key, # type: ignore[arg-type] + challenge_id=challenge.id, + request=request, + spt=spt, + metadata=resolved_metadata, + ) + + if pi["status"] == "requires_action": + raise PaymentActionRequiredError("Stripe PaymentIntent requires action") + if pi["status"] != "succeeded": + raise VerificationFailedError(f"Stripe PaymentIntent status: {pi['status']}") + + return Receipt.success( + reference=pi["id"], + method="stripe", + external_id=credential_external_id, + ) + + async def _create_with_client( + self, + client: Any, + challenge_id: str, + request: dict[str, Any], + spt: str, + metadata: dict[str, str], + ) -> dict[str, str]: + """Create a PaymentIntent using the Stripe SDK client.""" + try: + payment_intents = _resolve_payment_intents(client) + result = payment_intents.create( + { + "amount": int(request["amount"]), + "automatic_payment_methods": { + "allow_redirects": "never", + "enabled": True, + }, + "confirm": True, + "currency": request["currency"], + "metadata": metadata, + "shared_payment_granted_token": spt, + }, + options={"idempotency_key": f"mppx_{challenge_id}_{spt}"}, + ) + return {"id": result.id, "status": result.status} + except (VerificationFailedError, TypeError): + raise + except Exception as err: + raise VerificationFailedError("Stripe PaymentIntent failed") from err + + async def _create_with_secret_key( + self, + secret_key: str, + challenge_id: str, + request: dict[str, Any], + spt: str, + metadata: dict[str, str], + ) -> dict[str, str]: + """Create a PaymentIntent using raw HTTP with a secret key.""" + http_client = await self._get_http_client() + + auth_value = base64.b64encode(f"{secret_key}:".encode()).decode() + + body: dict[str, str] = { + "amount": str(request["amount"]), + "automatic_payment_methods[allow_redirects]": "never", + "automatic_payment_methods[enabled]": "true", + "confirm": "true", + "currency": str(request["currency"]), + "shared_payment_granted_token": spt, + } + for key, value in metadata.items(): + body[f"metadata[{key}]"] = value + + response = await http_client.post( + f"{STRIPE_API_BASE}/payment_intents", + headers={ + "Authorization": f"Basic {auth_value}", + "Content-Type": "application/x-www-form-urlencoded", + "Idempotency-Key": f"mppx_{challenge_id}_{spt}", + }, + data=body, + ) + + if not response.is_success: + detail = None + try: + err = response.json().get("error", {}) + detail = err.get("message") or err.get("code") + except Exception: + detail = response.text[:200] if response.text else None + raise VerificationFailedError( + detail or f"Stripe PaymentIntent failed (HTTP {response.status_code})" + ) + + result = response.json() + return {"id": result["id"], "status": result["status"]} diff --git a/src/mpp/methods/stripe/schemas.py b/src/mpp/methods/stripe/schemas.py new file mode 100644 index 0000000..4502fcb --- /dev/null +++ b/src/mpp/methods/stripe/schemas.py @@ -0,0 +1,35 @@ +"""Pydantic schemas for Stripe payment requests and credentials.""" + +from __future__ import annotations + +from pydantic import BaseModel, Field + + +class StripeMethodDetails(BaseModel): + """Method-specific details for Stripe charge requests.""" + + metadata: dict[str, str] | None = None + networkId: str + paymentMethodTypes: list[str] = Field(min_length=1) + + +class ChargeRequest(BaseModel): + """Request schema for the Stripe charge intent. + + After the transform in ``stripe()``, ``amount`` is in the smallest + currency unit (e.g. cents for USD) and ``decimals`` is removed. + """ + + amount: str + currency: str + description: str | None = None + externalId: str | None = None + methodDetails: StripeMethodDetails + recipient: str | None = None + + +class StripeCredentialPayload(BaseModel): + """Credential payload for Stripe SPT-based payments.""" + + externalId: str | None = None + spt: str diff --git a/src/mpp/methods/tempo/_rpc.py b/src/mpp/methods/tempo/_rpc.py index 5b51836..f48119a 100644 --- a/src/mpp/methods/tempo/_rpc.py +++ b/src/mpp/methods/tempo/_rpc.py @@ -5,7 +5,7 @@ import asyncio from typing import Any -DEFAULT_TIMEOUT = 30.0 +from mpp._defaults import DEFAULT_TIMEOUT async def _rpc_call( diff --git a/src/mpp/methods/tempo/intents.py b/src/mpp/methods/tempo/intents.py index 8d582ab..cafd433 100644 --- a/src/mpp/methods/tempo/intents.py +++ b/src/mpp/methods/tempo/intents.py @@ -13,6 +13,7 @@ import attrs from mpp import Credential, Receipt +from mpp._defaults import DEFAULT_TIMEOUT from mpp.errors import VerificationError from mpp.methods.tempo._defaults import DEFAULT_FEE_PAYER_URL, PATH_USD, rpc_url_for_chain from mpp.methods.tempo.schemas import ( @@ -29,8 +30,6 @@ from mpp.methods.tempo.account import TempoAccount -DEFAULT_TIMEOUT = 30.0 - # Receipt polling: 20 * 0.5s = ~10s, enough for testnet block times (~2-4s). MAX_RECEIPT_RETRY_ATTEMPTS = 20 RECEIPT_RETRY_DELAY_SECONDS = 0.5 diff --git a/tests/test_stripe.py b/tests/test_stripe.py new file mode 100644 index 0000000..2f03725 --- /dev/null +++ b/tests/test_stripe.py @@ -0,0 +1,774 @@ +"""Tests for the Stripe payment method.""" + +from __future__ import annotations + +import base64 +import math +import time +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +from typing import Any +from unittest.mock import AsyncMock + +import httpx +import pytest +from pydantic import ValidationError + +from mpp import Challenge, ChallengeEcho, Credential +from mpp.errors import ( + PaymentActionRequiredError, + PaymentExpiredError, + VerificationFailedError, +) +from mpp.methods.stripe import ChargeIntent, stripe +from mpp.methods.stripe.client import OnChallengeParameters +from mpp.methods.stripe.intents import _resolve_payment_intents +from mpp.methods.stripe.schemas import ChargeRequest, StripeCredentialPayload + +# ────────────────────────────────────────────────────────────────── +# Schema tests +# ────────────────────────────────────────────────────────────────── + + +class TestStripeCredentialPayload: + def test_valid_payload(self): + payload = StripeCredentialPayload.model_validate({"spt": "spt_test_123"}) + assert payload.spt == "spt_test_123" + assert payload.externalId is None + + def test_with_external_id(self): + payload = StripeCredentialPayload.model_validate( + {"spt": "spt_test_123", "externalId": "order-42"} + ) + assert payload.spt == "spt_test_123" + assert payload.externalId == "order-42" + + def test_missing_spt(self): + with pytest.raises(ValidationError): + StripeCredentialPayload.model_validate({"externalId": "order-42"}) + + +class TestChargeRequest: + def test_valid_request(self): + req = ChargeRequest.model_validate( + { + "amount": "150", + "currency": "usd", + "methodDetails": { + "networkId": "bn_test", + "paymentMethodTypes": ["card"], + }, + } + ) + assert req.amount == "150" + assert req.currency == "usd" + assert req.methodDetails.networkId == "bn_test" + assert req.methodDetails.paymentMethodTypes == ["card"] + + def test_missing_method_details(self): + with pytest.raises(ValidationError): + ChargeRequest.model_validate( + { + "amount": "150", + "currency": "usd", + } + ) + + def test_empty_payment_method_types(self): + with pytest.raises(ValidationError): + ChargeRequest.model_validate( + { + "amount": "150", + "currency": "usd", + "methodDetails": { + "networkId": "bn_test", + "paymentMethodTypes": [], + }, + } + ) + + +# ────────────────────────────────────────────────────────────────── +# Client tests +# ────────────────────────────────────────────────────────────────── + + +def _make_challenge(**overrides: Any) -> Challenge: + defaults = { + "id": "test-challenge-id", + "method": "stripe", + "intent": "charge", + "request": { + "amount": "150", + "currency": "usd", + "methodDetails": { + "networkId": "bn_test", + "paymentMethodTypes": ["card"], + }, + }, + "realm": "api.example.com", + "request_b64": "eyJ0ZXN0IjoidHJ1ZSJ9", + "expires": (datetime.now(UTC) + timedelta(hours=1)).isoformat(), + } + defaults.update(overrides) + return Challenge(**defaults) + + +class TestStripeMethod: + @pytest.mark.asyncio + async def test_create_credential(self): + async def fake_create_token(params: OnChallengeParameters) -> str: + assert params.amount == "150" + assert params.currency == "usd" + assert params.network_id == "bn_test" + assert params.payment_method == "pm_card_visa" + return "spt_test_abc" + + method = stripe( + create_token=fake_create_token, + payment_method="pm_card_visa", + intents={"charge": ChargeIntent(secret_key="sk_test_123")}, + ) + + challenge = _make_challenge() + cred = await method.create_credential(challenge) + + assert cred.payload["spt"] == "spt_test_abc" + assert cred.challenge.method == "stripe" + assert cred.challenge.intent == "charge" + + @pytest.mark.asyncio + async def test_create_credential_with_external_id(self): + async def fake_create_token(params: OnChallengeParameters) -> str: + return "spt_test_abc" + + method = stripe( + create_token=fake_create_token, + payment_method="pm_card_visa", + external_id="order-42", + intents={"charge": ChargeIntent(secret_key="sk_test_123")}, + ) + + challenge = _make_challenge() + cred = await method.create_credential(challenge) + + assert cred.payload["spt"] == "spt_test_abc" + assert cred.payload["externalId"] == "order-42" + + @pytest.mark.asyncio + async def test_create_credential_no_create_token_raises(self): + method = stripe( + payment_method="pm_card_visa", + intents={"charge": ChargeIntent(secret_key="sk_test_123")}, + ) + + challenge = _make_challenge() + with pytest.raises(ValueError, match="create_token"): + await method.create_credential(challenge) + + @pytest.mark.asyncio + async def test_create_credential_no_payment_method_raises(self): + async def fake_create_token(params: OnChallengeParameters) -> str: + return "spt_test_abc" + + method = stripe( + create_token=fake_create_token, + intents={"charge": ChargeIntent(secret_key="sk_test_123")}, + ) + + challenge = _make_challenge() + with pytest.raises(ValueError, match="payment_method"): + await method.create_credential(challenge) + + @pytest.mark.asyncio + async def test_create_credential_missing_network_id_raises(self): + """networkId is required in challenge.methodDetails (mppx parity).""" + + async def fake_create_token(params: OnChallengeParameters) -> str: + return "spt_test_abc" + + method = stripe( + create_token=fake_create_token, + payment_method="pm_card_visa", + intents={"charge": ChargeIntent(secret_key="sk_test_123")}, + ) + + challenge = _make_challenge( + request={ + "amount": "150", + "currency": "usd", + "methodDetails": { + "paymentMethodTypes": ["card"], + }, + } + ) + with pytest.raises(ValueError, match="networkId is required"): + await method.create_credential(challenge) + + @pytest.mark.asyncio + async def test_create_credential_rejects_metadata_external_id(self): + """metadata.externalId is reserved (mppx parity).""" + + async def fake_create_token(params: OnChallengeParameters) -> str: + return "spt_test_abc" + + method = stripe( + create_token=fake_create_token, + payment_method="pm_card_visa", + intents={"charge": ChargeIntent(secret_key="sk_test_123")}, + ) + + challenge = _make_challenge( + request={ + "amount": "150", + "currency": "usd", + "methodDetails": { + "networkId": "bn_test", + "paymentMethodTypes": ["card"], + "metadata": {"externalId": "should-fail"}, + }, + } + ) + with pytest.raises(ValueError, match="externalId is reserved"): + await method.create_credential(challenge) + + def test_transform_request(self): + method = stripe( + network_id="bn_test", + payment_method_types=["card"], + intents={"charge": ChargeIntent(secret_key="sk_test_123")}, + ) + + request = {"amount": "150", "currency": "usd"} + result = method.transform_request(request, None) + + assert result["methodDetails"]["networkId"] == "bn_test" + assert result["methodDetails"]["paymentMethodTypes"] == ["card"] + + def test_transform_request_does_not_overwrite(self): + method = stripe( + network_id="bn_default", + payment_method_types=["card"], + intents={"charge": ChargeIntent(secret_key="sk_test_123")}, + ) + + request = { + "amount": "150", + "currency": "usd", + "methodDetails": {"networkId": "bn_override"}, + } + result = method.transform_request(request, None) + + assert result["methodDetails"]["networkId"] == "bn_override" + assert result["methodDetails"]["paymentMethodTypes"] == ["card"] + + def test_method_name(self): + method = stripe( + intents={"charge": ChargeIntent(secret_key="sk_test_123")}, + ) + assert method.name == "stripe" + + def test_intents(self): + intent = ChargeIntent(secret_key="sk_test_123") + method = stripe(intents={"charge": intent}) + assert method.intents["charge"] is intent + + @pytest.mark.asyncio + async def test_expires_at_from_challenge(self): + """Verify expires_at is computed from the challenge's expires field.""" + recorded_params: list[OnChallengeParameters] = [] + + async def fake_create_token(params: OnChallengeParameters) -> str: + recorded_params.append(params) + return "spt_test" + + method = stripe( + create_token=fake_create_token, + payment_method="pm_card_visa", + intents={"charge": ChargeIntent(secret_key="sk_test_123")}, + ) + + expires = (datetime.now(UTC) + timedelta(hours=2)).isoformat() + challenge = _make_challenge(expires=expires) + await method.create_credential(challenge) + + expected = math.floor(datetime.fromisoformat(expires.replace("Z", "+00:00")).timestamp()) + assert recorded_params[0].expires_at == expected + + @pytest.mark.asyncio + async def test_expires_at_fallback_when_no_expires(self): + """When challenge has no expires, default to now + 1 hour.""" + recorded_params: list[OnChallengeParameters] = [] + + async def fake_create_token(params: OnChallengeParameters) -> str: + recorded_params.append(params) + return "spt_test" + + method = stripe( + create_token=fake_create_token, + payment_method="pm_card_visa", + intents={"charge": ChargeIntent(secret_key="sk_test_123")}, + ) + + challenge = _make_challenge(expires=None) + before = math.floor(time.time()) + 3600 + await method.create_credential(challenge) + after = math.floor(time.time()) + 3600 + + assert before <= recorded_params[0].expires_at <= after + + +# ────────────────────────────────────────────────────────────────── +# Server intent tests +# ────────────────────────────────────────────────────────────────── + + +def _make_credential( + spt: str = "spt_test_abc", + external_id: str | None = None, + expires: str | None = None, +) -> Credential: + payload: dict[str, Any] = {"spt": spt} + if external_id: + payload["externalId"] = external_id + if expires is None: + expires = (datetime.now(UTC) + timedelta(hours=1)).isoformat() + return Credential( + challenge=ChallengeEcho( + id="test-challenge-id", + realm="api.example.com", + method="stripe", + intent="charge", + request="eyJ0ZXN0IjoidHJ1ZSJ9", + expires=expires, + ), + payload=payload, + source="stripe:test", + ) + + +SAMPLE_REQUEST: dict[str, Any] = { + "amount": "150", + "currency": "usd", + "methodDetails": { + "networkId": "bn_test", + "paymentMethodTypes": ["card"], + }, +} + + +@dataclass +class FakePaymentIntent: + id: str = "pi_test_123" + status: str = "succeeded" + + +class FakePaymentIntents: + def __init__(self, result: FakePaymentIntent | None = None): + self._result = result or FakePaymentIntent() + + def create(self, *args: Any, **kwargs: Any) -> FakePaymentIntent: + return self._result + + +class FakeStripeClient: + """Legacy-style client with client.payment_intents.""" + + def __init__(self, result: FakePaymentIntent | None = None): + self.payment_intents = FakePaymentIntents(result) + + +class FakeV1: + def __init__(self, result: FakePaymentIntent | None = None): + self.payment_intents = FakePaymentIntents(result) + + +class FakeModernStripeClient: + """Modern-style client with client.v1.payment_intents (stripe-python v8+).""" + + def __init__(self, result: FakePaymentIntent | None = None): + self.v1 = FakeV1(result) + + +class TestResolvePaymentIntents: + def test_modern_client_v1(self): + client = FakeModernStripeClient() + pi = _resolve_payment_intents(client) + assert pi is client.v1.payment_intents + + def test_legacy_client(self): + client = FakeStripeClient() + pi = _resolve_payment_intents(client) + assert pi is client.payment_intents + + def test_unsupported_client_raises(self): + with pytest.raises(TypeError, match="Unsupported Stripe client"): + _resolve_payment_intents(object()) + + +class TestChargeIntent: + @pytest.mark.asyncio + async def test_verify_with_client_success(self): + intent = ChargeIntent(client=FakeStripeClient()) + credential = _make_credential() + receipt = await intent.verify(credential, SAMPLE_REQUEST) + + assert receipt.status == "success" + assert receipt.reference == "pi_test_123" + assert receipt.method == "stripe" + + @pytest.mark.asyncio + async def test_verify_with_modern_client_success(self): + """Verify works with client.v1.payment_intents (stripe-python v8+).""" + intent = ChargeIntent(client=FakeModernStripeClient()) + credential = _make_credential() + receipt = await intent.verify(credential, SAMPLE_REQUEST) + + assert receipt.status == "success" + assert receipt.reference == "pi_test_123" + assert receipt.method == "stripe" + + @pytest.mark.asyncio + async def test_verify_with_external_id(self): + intent = ChargeIntent(client=FakeStripeClient()) + credential = _make_credential(external_id="order-42") + receipt = await intent.verify(credential, SAMPLE_REQUEST) + + assert receipt.external_id == "order-42" + + @pytest.mark.asyncio + async def test_verify_expired_challenge(self): + intent = ChargeIntent(client=FakeStripeClient()) + expired = (datetime.now(UTC) - timedelta(hours=1)).isoformat() + credential = _make_credential(expires=expired) + + with pytest.raises(PaymentExpiredError): + await intent.verify(credential, SAMPLE_REQUEST) + + @pytest.mark.asyncio + async def test_verify_requires_action(self): + pi = FakePaymentIntent(status="requires_action") + intent = ChargeIntent(client=FakeStripeClient(result=pi)) + credential = _make_credential() + + with pytest.raises(PaymentActionRequiredError): + await intent.verify(credential, SAMPLE_REQUEST) + + @pytest.mark.asyncio + async def test_verify_failed_status(self): + pi = FakePaymentIntent(status="requires_payment_method") + intent = ChargeIntent(client=FakeStripeClient(result=pi)) + credential = _make_credential() + + with pytest.raises(VerificationFailedError, match="requires_payment_method"): + await intent.verify(credential, SAMPLE_REQUEST) + + @pytest.mark.asyncio + async def test_verify_missing_spt(self): + intent = ChargeIntent(client=FakeStripeClient()) + credential = Credential( + challenge=ChallengeEcho( + id="test-challenge-id", + realm="api.example.com", + method="stripe", + intent="charge", + request="eyJ0ZXN0IjoidHJ1ZSJ9", + expires=(datetime.now(UTC) + timedelta(hours=1)).isoformat(), + ), + payload={"not_spt": "bad"}, + ) + + with pytest.raises(VerificationFailedError, match="spt"): + await intent.verify(credential, SAMPLE_REQUEST) + + @pytest.mark.asyncio + async def test_verify_client_exception(self): + class FailingIntents: + def create(self, *args: Any, **kwargs: Any) -> Any: + raise RuntimeError("Stripe API error") + + class FailingClient: + payment_intents = FailingIntents() + + intent = ChargeIntent(client=FailingClient()) + credential = _make_credential() + + with pytest.raises(VerificationFailedError, match="PaymentIntent failed"): + await intent.verify(credential, SAMPLE_REQUEST) + + def test_no_client_or_secret_key_raises(self): + with pytest.raises(ValueError, match="Either client or secret_key"): + ChargeIntent() + + @pytest.mark.asyncio + async def test_analytics_metadata(self): + """Verify analytics metadata is passed to PaymentIntent creation.""" + captured: list[tuple[tuple, dict]] = [] + + class CapturingIntents: + def create(self, *args: Any, **kwargs: Any) -> FakePaymentIntent: + captured.append((args, kwargs)) + return FakePaymentIntent() + + class CapturingClient: + payment_intents = CapturingIntents() + + intent = ChargeIntent(client=CapturingClient()) + credential = _make_credential() + await intent.verify(credential, SAMPLE_REQUEST) + + params = captured[0][0][0] + metadata = params["metadata"] + assert metadata["mpp_version"] == "1" + assert metadata["mpp_is_mpp"] == "true" + assert metadata["mpp_intent"] == "charge" + assert metadata["mpp_challenge_id"] == "test-challenge-id" + assert metadata["mpp_server_id"] == "api.example.com" + assert metadata["mpp_client_id"] == "stripe:test" + + @pytest.mark.asyncio + async def test_idempotency_key(self): + """Verify idempotency key format matches mppx.""" + captured: list[tuple[tuple, dict]] = [] + + class CapturingIntents: + def create(self, *args: Any, **kwargs: Any) -> FakePaymentIntent: + captured.append((args, kwargs)) + return FakePaymentIntent() + + class CapturingClient: + payment_intents = CapturingIntents() + + intent = ChargeIntent(client=CapturingClient()) + credential = _make_credential(spt="spt_test_xyz") + await intent.verify(credential, SAMPLE_REQUEST) + + options = captured[0][1]["options"] + assert options["idempotency_key"] == "mppx_test-challenge-id_spt_test_xyz" + + @pytest.mark.asyncio + async def test_client_request_body_is_first_positional_arg(self): + """Verify request body is passed as first positional arg (Stripe SDK convention).""" + captured: list[tuple[tuple, dict]] = [] + + class CapturingIntents: + def create(self, *args: Any, **kwargs: Any) -> FakePaymentIntent: + captured.append((args, kwargs)) + return FakePaymentIntent() + + class CapturingClient: + payment_intents = CapturingIntents() + + intent = ChargeIntent(client=CapturingClient()) + credential = _make_credential() + await intent.verify(credential, SAMPLE_REQUEST) + + assert len(captured[0][0]) == 1 + body = captured[0][0][0] + assert body["amount"] == 150 + assert body["currency"] == "usd" + assert body["confirm"] is True + assert body["shared_payment_granted_token"] == "spt_test_abc" + assert "options" in captured[0][1] + + @pytest.mark.asyncio + async def test_user_metadata_overrides_analytics(self): + """User-supplied metadata should override analytics keys.""" + captured: list[tuple[tuple, dict]] = [] + + class CapturingIntents: + def create(self, *args: Any, **kwargs: Any) -> FakePaymentIntent: + captured.append((args, kwargs)) + return FakePaymentIntent() + + class CapturingClient: + payment_intents = CapturingIntents() + + intent = ChargeIntent(client=CapturingClient()) + credential = _make_credential() + request_with_metadata = { + **SAMPLE_REQUEST, + "methodDetails": { + **SAMPLE_REQUEST["methodDetails"], + "metadata": {"mpp_version": "custom", "user_key": "user_val"}, + }, + } + await intent.verify(credential, request_with_metadata) + + metadata = captured[0][0][0]["metadata"] + assert metadata["mpp_version"] == "custom" + assert metadata["user_key"] == "user_val" + + +# ────────────────────────────────────────────────────────────────── +# Raw HTTP path tests (_create_with_secret_key) +# ────────────────────────────────────────────────────────────────── + + +class TestChargeIntentRawHttp: + @pytest.mark.asyncio + async def test_verify_with_secret_key_success(self): + """Verify the raw HTTP path creates a PaymentIntent successfully.""" + mock_response = httpx.Response( + 200, + json={"id": "pi_http_123", "status": "succeeded"}, + request=httpx.Request("POST", "https://api.stripe.com/v1/payment_intents"), + ) + mock_client = AsyncMock(spec=httpx.AsyncClient) + mock_client.post.return_value = mock_response + + intent = ChargeIntent(secret_key="sk_test_raw", http_client=mock_client) + credential = _make_credential() + receipt = await intent.verify(credential, SAMPLE_REQUEST) + + assert receipt.status == "success" + assert receipt.reference == "pi_http_123" + assert receipt.method == "stripe" + + call_kwargs = mock_client.post.call_args + assert "api.stripe.com/v1/payment_intents" in call_kwargs.args[0] + + headers = call_kwargs.kwargs["headers"] + expected_auth = base64.b64encode(b"sk_test_raw:").decode() + assert headers["Authorization"] == f"Basic {expected_auth}" + assert headers["Idempotency-Key"] == "mppx_test-challenge-id_spt_test_abc" + + data = call_kwargs.kwargs["data"] + assert data["amount"] == "150" + assert data["currency"] == "usd" + assert data["shared_payment_granted_token"] == "spt_test_abc" + + @pytest.mark.asyncio + async def test_verify_with_secret_key_stripe_error(self): + """Verify Stripe error details are surfaced in the exception.""" + mock_response = httpx.Response( + 400, + json={"error": {"message": "Invalid card number", "code": "card_declined"}}, + request=httpx.Request("POST", "https://api.stripe.com/v1/payment_intents"), + ) + mock_client = AsyncMock(spec=httpx.AsyncClient) + mock_client.post.return_value = mock_response + + intent = ChargeIntent(secret_key="sk_test_raw", http_client=mock_client) + credential = _make_credential() + + with pytest.raises(VerificationFailedError, match="Invalid card number"): + await intent.verify(credential, SAMPLE_REQUEST) + + @pytest.mark.asyncio + async def test_verify_with_secret_key_non_json_error(self): + """Verify non-JSON error responses are handled gracefully.""" + mock_response = httpx.Response( + 500, + text="Internal Server Error", + request=httpx.Request("POST", "https://api.stripe.com/v1/payment_intents"), + ) + mock_client = AsyncMock(spec=httpx.AsyncClient) + mock_client.post.return_value = mock_response + + intent = ChargeIntent(secret_key="sk_test_raw", http_client=mock_client) + credential = _make_credential() + + with pytest.raises(VerificationFailedError, match="Internal Server Error"): + await intent.verify(credential, SAMPLE_REQUEST) + + @pytest.mark.asyncio + async def test_verify_with_secret_key_metadata_in_form(self): + """Verify metadata is encoded as form fields.""" + mock_response = httpx.Response( + 200, + json={"id": "pi_meta", "status": "succeeded"}, + request=httpx.Request("POST", "https://api.stripe.com/v1/payment_intents"), + ) + mock_client = AsyncMock(spec=httpx.AsyncClient) + mock_client.post.return_value = mock_response + + intent = ChargeIntent(secret_key="sk_test_raw", http_client=mock_client) + credential = _make_credential() + await intent.verify(credential, SAMPLE_REQUEST) + + data = mock_client.post.call_args.kwargs["data"] + assert data["metadata[mpp_is_mpp]"] == "true" + assert data["metadata[mpp_version]"] == "1" + + +# ────────────────────────────────────────────────────────────────── +# Async context manager tests +# ────────────────────────────────────────────────────────────────── + + +class TestChargeIntentLifecycle: + @pytest.mark.asyncio + async def test_aclose_closes_owned_client(self): + """Owned HTTP client is closed on aclose().""" + intent = ChargeIntent(secret_key="sk_test") + mock_client = AsyncMock(spec=httpx.AsyncClient) + intent._http_client = mock_client + intent._owns_client = True + + await intent.aclose() + + mock_client.aclose.assert_awaited_once() + assert intent._http_client is None + + @pytest.mark.asyncio + async def test_aclose_does_not_close_injected_client(self): + """Injected HTTP client is NOT closed on aclose().""" + mock_client = AsyncMock(spec=httpx.AsyncClient) + intent = ChargeIntent(secret_key="sk_test", http_client=mock_client) + + await intent.aclose() + + mock_client.aclose.assert_not_awaited() + + @pytest.mark.asyncio + async def test_context_manager_closes_owned_client(self): + """__aexit__ closes the owned HTTP client.""" + intent = ChargeIntent(secret_key="sk_test") + mock_client = AsyncMock(spec=httpx.AsyncClient) + intent._http_client = mock_client + intent._owns_client = True + + async with intent: + pass + + mock_client.aclose.assert_awaited_once() + + +# ────────────────────────────────────────────────────────────────── +# Integration: stripe() factory +# ────────────────────────────────────────────────────────────────── + + +class TestStripeFactory: + def test_defaults(self): + method = stripe( + intents={"charge": ChargeIntent(secret_key="sk_test")}, + ) + assert method.name == "stripe" + assert method.decimals == 2 + assert method.payment_method_types == ["card"] + assert method.currency is None + + def test_custom_params(self): + method = stripe( + intents={"charge": ChargeIntent(secret_key="sk_test")}, + currency="eur", + decimals=0, + network_id="bn_custom", + payment_method_types=["card", "sepa_debit"], + recipient="acct_123", + ) + assert method.currency == "eur" + assert method.decimals == 0 + assert method.network_id == "bn_custom" + assert method.payment_method_types == ["card", "sepa_debit"] + assert method.recipient == "acct_123" + + def test_no_secret_key_param(self): + """Factory no longer accepts secret_key (removed per review).""" + with pytest.raises(TypeError): + stripe( + intents={"charge": ChargeIntent(secret_key="sk_test")}, + secret_key="sk_test", # type: ignore[call-arg] + )