"""Provider-neutral payments.checkout: Stripe and PayPal adapters. Return/cancel URLs are not payment success. orders.order and invoice_record stay separate. Capture/refund are verbs on the checkout order. Dashboard is not required. """ from __future__ import annotations import json import re from pathlib import Path from unittest.mock import AsyncMock, MagicMock from uuid import uuid4 import pytest import yaml from src.domain.app_blocks.canonical import ( CANONICAL_BLOCK_IDS, DUAL_NAMESPACE_IDS, feature_pack_only_aliases, ) from src.domain.app_blocks.catalogs import FEATURE_PACK_ALIASES from src.domain.app_blocks.compiler import compile_install_plan from src.domain.app_blocks.generated_product_conformance_12h import ENTITY_BY_BLOCK from src.domain.app_blocks.hosts import merge_host_map from src.domain.app_blocks.planner import ProductIntent, select_capabilities from src.domain.app_blocks.public import _BLOCK_COPY from src.domain.app_blocks.registry import load_default_registry from src.domain.cloud.service import _REQUEST_TYPE_POLICY from src.domain.feature_packs.catalog_v2 import overlay_for from src.domain.feature_packs.contract import CATALOG_PRIMITIVES from src.domain.feature_packs.normalize import load_pack_contract, stamp_composition from src.domain.feature_packs.service import PackInstallService from src.integrations.platform_ports import QUEUE_REQUEST_TYPES ROOT = Path(__file__).resolve().parents[1] REGISTRY = ROOT / "src/domain/app_blocks/platform_coverage_registry.v1.yaml" CHECKOUT = "payments.checkout" JOB_PATH = "/api/jobs/payment.checkout.create" FORBIDDEN_HOST = ( "cloudgate", "beapipie", "P2HS_BACKEND_URL", "sk_live", "sk_test", "createCheckout", "/api/checkout", "order_record", "invoice_record", ) def _selected(purpose: str, goal: str = "") -> set[str]: intent = ProductIntent(purpose=purpose, first_build_goal=goal) return {block for _, block, _ in select_capabilities(intent)} def _host_fragment(root: Path, implementation: str) -> str: rel, _, symbol = implementation.partition("#") text = (root / rel).read_text(encoding="utf-8") if not symbol: return text match = re.search( rf"(?:export\s+)?(?:async\s+)?function\s+{re.escape(symbol)}\b", text, ) if not match: return text rest = text[match.end() :] nxt = re.search(r"\nexport\s+(?:async\s+)?function\s+", rest) end = match.end() + nxt.start() if nxt else len(text) return text[match.start() : end] def test_checkout_is_dual_namespace_not_pack_only(): assert FEATURE_PACK_ALIASES[CHECKOUT] == "payment-checkout" assert CHECKOUT in CANONICAL_BLOCK_IDS assert CHECKOUT in DUAL_NAMESPACE_IDS assert CHECKOUT not in feature_pack_only_aliases() payment_jobs = { "payments.checkout.create", "payments.order.capture", "payments.refund.create", "payments.catalog.upsert", } assert payment_jobs.isdisjoint(CATALOG_PRIMITIVES) block = load_default_registry().require(CHECKOUT) canonical = load_default_registry().canonical.require(CHECKOUT) assert canonical.installation.feature_packs == ("payment-checkout",) primitives = {dep.id for dep in canonical.requires if dep.kind == "primitive"} assert primitives == {"payments.checkout.create"} optional = {dep.id for dep in canonical.optional_requires if dep.kind == "primitive"} assert optional == {"payments.order.capture", "payments.refund.create", "payments.catalog.upsert"} assert "data.entity" not in primitives assert {slot.id for slot in block.slots if slot.type == "ui_slot"} == { "create", "return", "cancel", "catalog", "capture", "refund", } assert _BLOCK_COPY[CHECKOUT] == "Checkout" def test_ontology_split_vs_orders_and_invoices(): assert ENTITY_BY_BLOCK[CHECKOUT] == "order" assert ENTITY_BY_BLOCK["orders.order"] == "order_record" assert ENTITY_BY_BLOCK["billing.invoice"] == "invoice_record" checkout = overlay_for("payment-checkout") keys = {entity.entity_key for entity in checkout["entities"]} assert keys == {"product", "price", "order"} order = next(entity for entity in checkout["entities"] if entity.entity_key == "order") assert order.schema_mode == "protected" application = overlay_for("order-management") assert application["entities"][0].entity_key == "order_record" invoices = overlay_for("billing-management") assert any(entity.entity_key == "invoice_record" for entity in invoices["entities"]) bindings = overlay_for("payment-checkout")["webhook_bindings"] assert {item.provider for item in bindings} == {"stripe", "paypal"} assert {item.target for item in bindings} == {"pack:payment-checkout.reconcile"} stripe_events = next(item.events for item in bindings if item.provider == "stripe") paypal_events = next(item.events for item in bindings if item.provider == "paypal") assert "setup_intent.succeeded" in stripe_events assert "charge.refunded" in stripe_events assert "payment_intent.payment_failed" in stripe_events assert "PAYMENT.AUTHORIZATION.CREATED" in paypal_events assert "PAYMENT.CAPTURE.REFUNDED" in paypal_events policy = overlay_for("payment-checkout")["field_policy"]["order"] assert "provider_checkout_id" in policy assert "provider_setup_id" in policy assert "capture_mode" in policy assert "stripe_checkout_session_id" in policy def test_wave3_checkout_app_block_gap_stays_on_orders(): registry = yaml.safe_load(REGISTRY.read_text(encoding="utf-8")) by_id = {row["id"]: row for row in registry["capabilities"]} assert "checkout_app_block" in by_id["orders"]["gaps"] assert by_id["orders.order"]["gaps"] == ["checkout_app_block"] assert by_id[CHECKOUT]["status"] == "frozen" assert by_id[CHECKOUT]["gaps"] == [] assert by_id[CHECKOUT]["engine"] == "installed" assert by_id[CHECKOUT]["generated_host"] == "proven" assert by_id[CHECKOUT]["dashboard"] == "not_required_wave_5" assert by_id["payment-checkout"]["status"] == "provider_specific" assert by_id["payment-checkout"]["dependencies"] == ["stripe", "paypal"] assert "must_not_become_catalog_or_orders_ontology" in by_id["payment-checkout"]["gaps"] def test_planner_selects_checkout_without_collapsing_orders(): assert CHECKOUT in _selected("Staff manage a product catalog and add Stripe checkout") assert "catalog.product" in _selected("Staff manage a product catalog and add Stripe checkout") assert CHECKOUT in _selected("Staff manage product catalog and accept payments") assert CHECKOUT in _selected("Build a Stripe checkout") assert CHECKOUT in _selected("Build a PayPal checkout") assert CHECKOUT in _selected("Accept payments") assert CHECKOUT in _selected("shopping checkout") assert "catalog.product" not in _selected("Build a Stripe checkout") assert "orders.order" not in _selected("shopping checkout") history = _selected("Show customer order history after Stripe checkout") assert "orders.order" in history assert CHECKOUT not in history later = _selected("Add Stripe checkout products and a Shopify product catalog later") assert CHECKOUT not in later assert "catalog.product" not in later catalog_only = _selected("Staff manage a product catalog. No checkout, orders, or billing.") assert CHECKOUT not in catalog_only billing = _selected("Stripe billing portal for workspace invoices") assert CHECKOUT not in billing def test_install_plan_uses_existing_jobs_not_provider_sdk_handlers(): plan = compile_install_plan( None, source_prompt="Build a Stripe checkout", ) assert CHECKOUT in {block.id for block in plan.blocks} assert "payment-checkout" in plan.required_feature_packs assert "payments.checkout.create" in plan.required_primitives payment_jobs = { "payment.checkout.create", "payment.capture", "payment.refund", "payment.catalog.upsert", } assert payment_jobs <= set(QUEUE_REQUEST_TYPES) assert payment_jobs <= set(_REQUEST_TYPE_POLICY) invented = { "checkout.create", "stripe.checkout.create", "paypal.checkout.create", } assert not (invented & set(QUEUE_REQUEST_TYPES)) assert not (invented & set(_REQUEST_TYPE_POLICY)) def test_checkout_adapters_are_stripe_or_paypal_not_both_required(): contract = load_pack_contract("payment-checkout", {"requires": []}) assert contract.configuration_providers() == ["stripe", "paypal"] binding = contract.provider_bindings[0] assert binding.providers == ["stripe", "paypal"] assert binding.configuration_required is True notes = contract.compatibility.implementation_notes or "" assert "stripe, paypal" in notes stamped = stamp_composition( "payment-checkout", { "requires": [], "entities": [ { "entity_key": "price", "schema": { "fields": [ {"key": "stripe_price_id", "name": "Stripe Price ID", "type": "text"}, {"key": "unit_amount", "name": "Unit Amount", "type": "number"}, {"key": "currency", "name": "Currency", "type": "text"}, ] }, }, { "entity_key": "order", "schema": { "fields": [ {"key": "source_job_id", "name": "Source Job ID", "type": "text"}, {"key": "status", "name": "Status", "type": "text"}, {"key": "provider", "name": "Provider", "type": "text"}, { "key": "provider_checkout_id", "name": "Provider checkout ID", "type": "text", }, ] }, }, ], }, ) price_fields = {field["key"]: field for field in stamped["entities"][0]["schema"]["fields"]} assert price_fields["stripe_price_id"].get("provider") is True assert "system_managed" not in price_fields["unit_amount"] order_fields = {field["key"]: field for field in stamped["entities"][1]["schema"]["fields"]} assert order_fields["provider_checkout_id"].get("provider") is True @pytest.mark.asyncio async def test_missing_prerequisites_or_within_checkout_binding(monkeypatch): connected: set[str] = set() async def fake_has(_repo, _project_id, key): return key in connected monkeypatch.setattr( "src.domain.feature_packs.service.has_active_connector_connection", fake_has, ) svc = PackInstallService(AsyncMock(), MagicMock()) project_id = uuid4() connected.clear() _hard, missing = await svc._missing_prerequisites( project_id, "payment-checkout", {"requires": []} ) assert set(missing) == {"stripe", "paypal"} connected.add("paypal") _hard, missing = await svc._missing_prerequisites( project_id, "payment-checkout", {"requires": []} ) assert missing == [] connected.clear() connected.add("stripe") _hard, missing = await svc._missing_prerequisites( project_id, "payment-checkout", {"requires": []} ) assert missing == [] def test_generated_hosts_use_relative_job_bff_and_do_not_treat_return_as_paid(): for framework in ("html", "nuxt_app", "react_app"): root = ROOT / "boilerplates" / framework hosts = merge_host_map(json.loads((root / "p2hs/app-block-hosts.json").read_text())) rows = {item["slot"]: item for item in hosts["hosts"] if item["block"] == CHECKOUT} assert {"create", "return", "cancel", "catalog", "capture", "refund"} <= set(rows) rel = rows["create"]["implementation"].partition("#")[0] create = (root / rel).read_text(encoding="utf-8") catalog = _host_fragment(root, rows["catalog"]["implementation"]) capture_host = _host_fragment(root, rows["capture"]["implementation"]) refund_host = _host_fragment(root, rows["refund"]["implementation"]) returned = _host_fragment(root, rows["return"]["implementation"]) cancelled = _host_fragment(root, rows["cancel"]["implementation"]) assert JOB_PATH in create assert "Idempotency-Key" in create assert "/api/jobs/" in create assert "success_url" in create assert "cancel_url" in create assert "checkout_url" in create assert "capture_mode" in create assert "client_secret" in create assert "publishable_key" in create assert "/api/jobs/payment.catalog.upsert" in catalog assert "/api/jobs/payment.capture" in capture_host assert "/api/jobs/payment.refund" in refund_host blob = f"{create}\n{returned}\n{cancelled}\n{catalog}\n{capture_host}\n{refund_host}" for marker in FORBIDDEN_HOST: assert marker not in blob, (framework, marker) assert "Awaiting payment confirmation" in returned assert "not proof of payment" in returned.lower() assert "not completed" in cancelled.lower() assert not re.search(r"\bpaid\b", returned, re.I) assert not re.search(r"\bpaid\b", cancelled, re.I) assert "status=paid" not in returned assert "PATCH" not in returned assert "PATCH" not in cancelled catalog_job = root / "functions/api/jobs/payment.catalog.upsert.js" capture = root / "functions/api/jobs/payment.capture.js" refund = root / "functions/api/jobs/payment.refund.js" assert catalog_job.exists(), (framework, catalog_job) assert capture.exists(), (framework, capture) assert refund.exists(), (framework, refund) assert "order_record" not in capture.read_text(encoding="utf-8") assert "invoice_record" not in refund.read_text(encoding="utf-8") assert "order_record" not in catalog_job.read_text(encoding="utf-8")