padelnomics/scripts/stripe_e2e_test.py

"""
Comprehensive Stripe E2E Tests — real webhooks via ngrok.

Tests every product type, subscription lifecycle, payment failures,
and edge cases against a running dev server with real Stripe webhooks.

Prerequisites:
    1. ngrok http 5000
    2. uv run python scripts/stripe_e2e_setup.py
    3. make dev  (or restart after setup)
    4. uv run python scripts/stripe_e2e_test.py
"""

import os
import sqlite3
import subprocess
import sys
import time

from dotenv import load_dotenv

load_dotenv(override=True)

import stripe

STRIPE_SECRET_KEY = os.getenv("STRIPE_SECRET_KEY", "") or os.getenv("STRIPE_API_PRIVATE_KEY", "")
assert STRIPE_SECRET_KEY, "Set STRIPE_SECRET_KEY or STRIPE_API_PRIVATE_KEY in .env"

stripe.api_key = STRIPE_SECRET_KEY
stripe.max_network_retries = 2

DATABASE_PATH = os.getenv("DATABASE_PATH", "data/app.db")
MAX_WAIT_SECONDS = 20
POLL_SECONDS = 0.5

passed = 0
failed = 0
errors = []
cleanup_sub_ids = []


# ─── Helpers ──────────────────────────────────────────────

def ok(msg):
    global passed
    passed += 1
    print(f"  \u2713 {msg}")


def fail(msg):
    global failed
    failed += 1
    errors.append(msg)
    print(f"  \u2717 {msg}")


def section(title):
    print(f"\n{'─' * 60}")
    print(f"  {title}")
    print(f"{'─' * 60}")


def query_db(sql, params=()):
    conn = sqlite3.connect(f"file:{DATABASE_PATH}?mode=ro", uri=True)
    conn.row_factory = sqlite3.Row
    try:
        return [dict(r) for r in conn.execute(sql, params).fetchall()]
    finally:
        conn.close()


def wait_for_row(sql, params=(), timeout_seconds=MAX_WAIT_SECONDS):
    """Poll until query returns at least one row."""
    deadline = time.time() + timeout_seconds
    while time.time() < deadline:
        rows = query_db(sql, params)
        if rows:
            return rows
        time.sleep(POLL_SECONDS)
    return []


def wait_for_value(sql, params, column, expected, timeout_seconds=MAX_WAIT_SECONDS):
    """Poll until column == expected."""
    deadline = time.time() + timeout_seconds
    last = None
    while time.time() < deadline:
        rows = query_db(sql, params)
        if rows:
            last = rows[0]
            if last[column] == expected:
                return last
        time.sleep(POLL_SECONDS)
    return last


def get_or_create_customer(email, name):
    existing = stripe.Customer.list(email=email, limit=1)
    if existing.data:
        return existing.data[0]
    return stripe.Customer.create(email=email, name=name, metadata={"e2e": "true"})


_pm_cache = {}

def attach_pm(customer_id):
    """Create a fresh test Visa and attach it."""
    if customer_id in _pm_cache:
        return _pm_cache[customer_id]
    pm = stripe.PaymentMethod.create(type="card", card={"token": "tok_visa"})
    stripe.PaymentMethod.attach(pm.id, customer=customer_id)
    stripe.Customer.modify(customer_id, invoice_settings={"default_payment_method": pm.id})
    _pm_cache[customer_id] = pm.id
    return pm.id


def create_sub(customer_id, price_id, metadata, pm_id):
    """Create subscription and track for cleanup."""
    sub = stripe.Subscription.create(
        customer=customer_id,
        items=[{"price": price_id}],
        metadata=metadata,
        default_payment_method=pm_id,
    )
    cleanup_sub_ids.append(sub.id)
    return sub


def cancel_sub(sub_id):
    try:
        stripe.Subscription.cancel(sub_id)
    except stripe.InvalidRequestError:
        pass


# ─── Preflight ────────────────────────────────────────────

section("Preflight")

# Dev server
result = subprocess.run(
    ["curl", "-s", "-o", "/dev/null", "-w", "%{http_code}", "http://localhost:5000/"],
    capture_output=True, text=True, timeout=5,
)
assert result.stdout.strip() in ("200", "301", "302"), f"Dev server down (HTTP {result.stdout.strip()})"
ok("Dev server running")

# Webhook endpoint
endpoint_id = os.getenv("STRIPE_WEBHOOK_ENDPOINT_ID", "")
assert endpoint_id, "STRIPE_WEBHOOK_ENDPOINT_ID not set — run stripe_e2e_setup.py"
ep = stripe.WebhookEndpoint.retrieve(endpoint_id)
assert ep.status == "enabled", f"Endpoint status: {ep.status}"
ok(f"Webhook endpoint: {ep.url}")

# Webhook secret loaded in server
result = subprocess.run(
    ["curl", "-s", "-o", "/dev/null", "-w", "%{http_code}",
     "-X", "POST", "-H", "Content-Type: application/json",
     "-d", "{}", "http://localhost:5000/billing/webhook/stripe"],
    capture_output=True, text=True, timeout=5,
)
assert result.stdout.strip() == "400", f"Webhook returns {result.stdout.strip()} (need 400 = sig check active)"
ok("Webhook signature verification active")

# Price map
products = query_db("SELECT key, provider_price_id, billing_type FROM payment_products WHERE provider = 'stripe'")
price_map = {p["key"]: p for p in products}
assert len(price_map) >= 17, f"Only {len(price_map)} products"
ok(f"{len(price_map)} Stripe products loaded")

# Test data
users = query_db("SELECT id, email FROM users LIMIT 10")
assert users
test_user = users[0]
ok(f"User: {test_user['email']} (id={test_user['id']})")

suppliers = query_db("SELECT id, name, claimed_by, credit_balance, tier FROM suppliers LIMIT 5")
assert suppliers
# Pick a supplier with claimed_by set (has an owner user)
test_supplier = next((s for s in suppliers if s["claimed_by"]), suppliers[0])
supplier_user_id = test_supplier["claimed_by"] or test_user["id"]
ok(f"Supplier: {test_supplier['name']} (id={test_supplier['id']}, owner={supplier_user_id})")

# Record initial supplier state for later comparison
initial_credit_balance = test_supplier["credit_balance"]


# ═══════════════════════════════════════════════════════════
# 1. PLANNER SUBSCRIPTIONS
# ═══════════════════════════════════════════════════════════

section("1a. Planner Starter — create → verify DB → cancel → verify cancelled")

cus_starter = get_or_create_customer("e2e-starter@sandbox.padelnomics.com", "E2E Starter")
pm_starter = attach_pm(cus_starter.id)

sub = create_sub(cus_starter.id, price_map["starter"]["provider_price_id"],
                 {"user_id": str(test_user["id"]), "plan": "starter"}, pm_starter)
ok(f"Created: {sub.id} (status={sub.status})")

rows = wait_for_row("SELECT * FROM subscriptions WHERE provider_subscription_id = ?", (sub.id,))
if rows:
    r = rows[0]
    ok(f"DB: plan={r['plan']}, status={r['status']}") if r["plan"] == "starter" and r["status"] == "active" else fail(f"DB: plan={r['plan']}, status={r['status']}")
    if r.get("current_period_end"):
        ok(f"period_end set: {r['current_period_end'][:10]}")
    else:
        fail("period_end is NULL")
else:
    fail("Subscription NOT in DB")

# billing_customers
bc = query_db("SELECT * FROM billing_customers WHERE user_id = ?", (test_user["id"],))
ok("billing_customers created") if bc else fail("billing_customers NOT created")

# Cancel
cancel_sub(sub.id)
result = wait_for_value("SELECT status FROM subscriptions WHERE provider_subscription_id = ?",
                        (sub.id,), "status", "cancelled")
ok("Status → cancelled") if result and result["status"] == "cancelled" else fail(f"Status: {result['status'] if result else '?'}")


section("1b. Planner Pro — subscription lifecycle")

pro_user = users[1] if len(users) > 1 else users[0]
cus_pro = get_or_create_customer("e2e-pro@sandbox.padelnomics.com", "E2E Pro")
pm_pro = attach_pm(cus_pro.id)

sub = create_sub(cus_pro.id, price_map["pro"]["provider_price_id"],
                 {"user_id": str(pro_user["id"]), "plan": "pro"}, pm_pro)
ok(f"Created: {sub.id}")

rows = wait_for_row("SELECT * FROM subscriptions WHERE provider_subscription_id = ?", (sub.id,))
if rows and rows[0]["plan"] == "pro" and rows[0]["status"] == "active":
    ok("DB: plan=pro, status=active")
else:
    fail(f"DB: {rows[0] if rows else 'not found'}")

cancel_sub(sub.id)
ok("Cleaned up")


# ═══════════════════════════════════════════════════════════
# 2. SUPPLIER SUBSCRIPTIONS (all 4 variants)
# ═══════════════════════════════════════════════════════════

section("2a. Supplier Growth (monthly) — tier, credits, verified")

cus_sup = get_or_create_customer("e2e-supplier@sandbox.padelnomics.com", "E2E Supplier")
pm_sup = attach_pm(cus_sup.id)

sub = create_sub(cus_sup.id, price_map["supplier_growth"]["provider_price_id"], {
    "user_id": str(supplier_user_id),
    "supplier_id": str(test_supplier["id"]),
    "plan": "supplier_growth",
}, pm_sup)
ok(f"Created: {sub.id}")

result = wait_for_value(
    "SELECT tier, is_verified, monthly_credits, credit_balance FROM suppliers WHERE id = ?",
    (test_supplier["id"],), "tier", "growth",
)
if result:
    ok("tier=growth") if result["tier"] == "growth" else fail(f"tier={result['tier']}")
    ok("is_verified=1") if result["is_verified"] == 1 else fail(f"is_verified={result['is_verified']}")
    ok("monthly_credits=30") if result["monthly_credits"] == 30 else fail(f"monthly_credits={result['monthly_credits']}")
    ok(f"credit_balance={result['credit_balance']}") if result["credit_balance"] >= 30 else fail(f"credit_balance={result['credit_balance']}")
else:
    fail("Tier not updated")

# Check credit ledger entry was created
ledger = query_db(
    "SELECT * FROM credit_ledger WHERE supplier_id = ? AND event_type = 'monthly_allocation' ORDER BY id DESC LIMIT 1",
    (test_supplier["id"],),
)
ok("Credit ledger entry created") if ledger else fail("No credit ledger entry")

cancel_sub(sub.id)
ok("Cleaned up")


section("2b. Supplier Pro (monthly) — 100 credits")

# Reset supplier to basic first
query_conn = sqlite3.connect(DATABASE_PATH)
query_conn.execute("UPDATE suppliers SET tier='free', monthly_credits=0, credit_balance=0, is_verified=0 WHERE id=?",
                   (test_supplier["id"],))
query_conn.commit()
query_conn.close()
time.sleep(1)

sub = create_sub(cus_sup.id, price_map["supplier_pro"]["provider_price_id"], {
    "user_id": str(supplier_user_id),
    "supplier_id": str(test_supplier["id"]),
    "plan": "supplier_pro",
}, pm_sup)
ok(f"Created: {sub.id}")

result = wait_for_value(
    "SELECT tier, monthly_credits, credit_balance FROM suppliers WHERE id = ?",
    (test_supplier["id"],), "tier", "pro",
)
if result:
    ok("tier=pro") if result["tier"] == "pro" else fail(f"tier={result['tier']}")
    ok("monthly_credits=100") if result["monthly_credits"] == 100 else fail(f"monthly_credits={result['monthly_credits']}")
    ok(f"credit_balance={result['credit_balance']}") if result["credit_balance"] >= 100 else fail(f"credit_balance={result['credit_balance']}")
else:
    fail("Tier not updated to pro")

cancel_sub(sub.id)
ok("Cleaned up")


section("2c. Supplier Growth (yearly)")

# Reset
query_conn = sqlite3.connect(DATABASE_PATH)
query_conn.execute("UPDATE suppliers SET tier='free', monthly_credits=0, credit_balance=0, is_verified=0 WHERE id=?",
                   (test_supplier["id"],))
query_conn.commit()
query_conn.close()
time.sleep(1)

sub = create_sub(cus_sup.id, price_map["supplier_growth_yearly"]["provider_price_id"], {
    "user_id": str(supplier_user_id),
    "supplier_id": str(test_supplier["id"]),
    "plan": "supplier_growth_yearly",
}, pm_sup)
ok(f"Created: {sub.id}")

result = wait_for_value(
    "SELECT tier, monthly_credits FROM suppliers WHERE id = ?",
    (test_supplier["id"],), "tier", "growth",
)
if result:
    ok("tier=growth (yearly maps to growth)")
    ok("monthly_credits=30") if result["monthly_credits"] == 30 else fail(f"monthly_credits={result['monthly_credits']}")
else:
    fail("Yearly growth not processed")

cancel_sub(sub.id)
ok("Cleaned up")


section("2d. Supplier Pro (yearly)")

query_conn = sqlite3.connect(DATABASE_PATH)
query_conn.execute("UPDATE suppliers SET tier='free', monthly_credits=0, credit_balance=0, is_verified=0 WHERE id=?",
                   (test_supplier["id"],))
query_conn.commit()
query_conn.close()
time.sleep(1)

sub = create_sub(cus_sup.id, price_map["supplier_pro_yearly"]["provider_price_id"], {
    "user_id": str(supplier_user_id),
    "supplier_id": str(test_supplier["id"]),
    "plan": "supplier_pro_yearly",
}, pm_sup)
ok(f"Created: {sub.id}")

result = wait_for_value(
    "SELECT tier, monthly_credits FROM suppliers WHERE id = ?",
    (test_supplier["id"],), "tier", "pro",
)
if result:
    ok("tier=pro (yearly maps to pro)")
    ok("monthly_credits=100") if result["monthly_credits"] == 100 else fail(f"monthly_credits={result['monthly_credits']}")
else:
    fail("Yearly pro not processed")

cancel_sub(sub.id)
ok("Cleaned up")


# ═══════════════════════════════════════════════════════════
# 3. BOOST ADD-ON SUBSCRIPTIONS (all 4)
# ═══════════════════════════════════════════════════════════

section("3. Boost add-on subscriptions (Logo, Highlight, Verified, Card Color)")

cus_boost = get_or_create_customer("e2e-boost@sandbox.padelnomics.com", "E2E Boost")
pm_boost = attach_pm(cus_boost.id)

boost_keys = ["boost_logo", "boost_highlight", "boost_verified", "boost_card_color"]
for key in boost_keys:
    price_id = price_map[key]["provider_price_id"]
    sub = create_sub(cus_boost.id, price_id, {
        "user_id": str(supplier_user_id),
        "supplier_id": str(test_supplier["id"]),
        "plan": key,
    }, pm_boost)
    ok(f"{key}: {sub.id} (active)")
    # Let webhook arrive
    time.sleep(2)
    cancel_sub(sub.id)

# Boosts with plan starting "boost_" don't hit supplier handler (only supplier_ plans do).
# They go through the user subscription path. Verify at least the webhooks were accepted.
# Check ngrok logs for 200s
import json
import urllib.request
try:
    resp = urllib.request.urlopen("http://localhost:4040/api/requests/http?limit=50", timeout=5)
    requests_data = json.loads(resp.read())
    webhook_200s = sum(1 for r in requests_data.get("requests", [])
                       if r.get("request", {}).get("uri") == "/billing/webhook/stripe"
                       and r.get("response", {}).get("status_code") == 200)
    ok(f"Webhook 200 responses seen: {webhook_200s}")
except Exception:
    print("  (could not verify ngrok logs)")

ok("All 4 boost add-ons tested")


# ═══════════════════════════════════════════════════════════
# 4. CHECKOUT SESSIONS — every product
# ═══════════════════════════════════════════════════════════

section("4. Checkout session creation (all 17 products)")

try:
    ngrok_resp = urllib.request.urlopen("http://localhost:4040/api/tunnels", timeout=5)
    tunnel_url = json.loads(ngrok_resp.read())["tunnels"][0]["public_url"]
except Exception:
    tunnel_url = "http://localhost:5000"

checkout_ok = 0
for key, p in sorted(price_map.items()):
    mode = "subscription" if p["billing_type"] == "subscription" else "payment"
    try:
        stripe.checkout.Session.create(
            mode=mode,
            customer=cus_starter.id,
            line_items=[{"price": p["provider_price_id"], "quantity": 1}],
            metadata={"user_id": str(test_user["id"]), "plan": key, "test": "true"},
            success_url=f"{tunnel_url}/billing/success?session_id={{CHECKOUT_SESSION_ID}}",
            cancel_url=f"{tunnel_url}/billing/pricing",
        )
        checkout_ok += 1
    except stripe.StripeError as e:
        fail(f"Checkout failed: {key} -> {e}")

if checkout_ok == len(price_map):
    ok(f"All {checkout_ok} checkout sessions created")
else:
    fail(f"{len(price_map) - checkout_ok} checkout sessions failed")


# ═══════════════════════════════════════════════════════════
# 5. PAYMENT FAILURE — declined card
# ═══════════════════════════════════════════════════════════

section("5. Payment failure — declined card scenarios")

cus_fail = get_or_create_customer("e2e-failure@sandbox.padelnomics.com", "E2E Failure")
fail_user = users[2] if len(users) > 2 else users[0]

# 5a. First create a valid subscription, then simulate payment failure
pm_valid = attach_pm(cus_fail.id)
try:
    sub_fail = stripe.Subscription.create(
        customer=cus_fail.id,
        items=[{"price": price_map["starter"]["provider_price_id"]}],
        metadata={"user_id": str(fail_user["id"]), "plan": "starter"},
        default_payment_method=pm_valid,
    )
    cleanup_sub_ids.append(sub_fail.id)
    ok(f"Created valid sub first: {sub_fail.id} (status={sub_fail.status})")

    # Wait for subscription.created webhook
    rows = wait_for_row("SELECT * FROM subscriptions WHERE provider_subscription_id = ?", (sub_fail.id,))
    ok("DB row created") if rows else fail("No DB row after valid sub creation")

    # Now swap to a declined card — next invoice will fail
    try:
        pm_decline = stripe.PaymentMethod.create(type="card", card={"token": "tok_chargeDeclined"})
        stripe.PaymentMethod.attach(pm_decline.id, customer=cus_fail.id)
        stripe.Customer.modify(cus_fail.id, invoice_settings={"default_payment_method": pm_decline.id})
        ok("Swapped to declined card for next billing cycle")
    except stripe.CardError:
        ok("tok_chargeDeclined rejected at attach (newer API) — card swap skipped")

    cancel_sub(sub_fail.id)
    result = wait_for_value("SELECT status FROM subscriptions WHERE provider_subscription_id = ?",
                            (sub_fail.id,), "status", "cancelled")
    ok("Cancelled after failure test") if result else ok("Cleanup done")

except stripe.CardError as e:
    ok(f"Card declined at subscription level: {e.user_message}")

# 5b. Try creating subscription with payment_behavior=default_incomplete
try:
    pm_ok = stripe.PaymentMethod.create(type="card", card={"token": "tok_visa"})
    stripe.PaymentMethod.attach(pm_ok.id, customer=cus_fail.id)
    sub_inc = stripe.Subscription.create(
        customer=cus_fail.id,
        items=[{"price": price_map["pro"]["provider_price_id"]}],
        metadata={"user_id": str(fail_user["id"]), "plan": "pro"},
        default_payment_method=pm_ok.id,
        payment_behavior="default_incomplete",
    )
    cleanup_sub_ids.append(sub_inc.id)
    ok(f"Incomplete-mode sub: {sub_inc.id} (status={sub_inc.status})")
    cancel_sub(sub_inc.id)
except stripe.StripeError as e:
    ok(f"Incomplete mode handled: {e}")


# ═══════════════════════════════════════════════════════════
# 6. EDGE CASES
# ═══════════════════════════════════════════════════════════

section("6a. Edge case — missing user_id in metadata")

cus_edge = get_or_create_customer("e2e-edge@sandbox.padelnomics.com", "E2E Edge")
pm_edge = attach_pm(cus_edge.id)

sub = create_sub(cus_edge.id, price_map["starter"]["provider_price_id"],
                 {"plan": "starter"},  # NO user_id
                 pm_edge)
ok(f"Created sub without user_id: {sub.id}")

# Webhook should arrive but handler should not crash (no DB write expected)
time.sleep(5)

# Server should not have crashed — verify it's still up
result = subprocess.run(
    ["curl", "-s", "-o", "/dev/null", "-w", "%{http_code}", "http://localhost:5000/"],
    capture_output=True, text=True, timeout=5,
)
ok("Server still alive after missing user_id") if result.stdout.strip() in ("200", "301", "302") else fail("Server crashed!")

cancel_sub(sub.id)


section("6b. Edge case — missing supplier_id for supplier plan")

sub = create_sub(cus_edge.id, price_map["supplier_growth"]["provider_price_id"],
                 {"user_id": str(test_user["id"]), "plan": "supplier_growth"},  # NO supplier_id
                 pm_edge)
ok(f"Created supplier sub without supplier_id: {sub.id}")
time.sleep(5)

result = subprocess.run(
    ["curl", "-s", "-o", "/dev/null", "-w", "%{http_code}", "http://localhost:5000/"],
    capture_output=True, text=True, timeout=5,
)
ok("Server still alive after missing supplier_id") if result.stdout.strip() in ("200", "301", "302") else fail("Server crashed!")

cancel_sub(sub.id)


section("6c. Edge case — duplicate subscription (idempotency)")

# Create same subscription twice for same user
cus_dup = get_or_create_customer("e2e-dup@sandbox.padelnomics.com", "E2E Dup")
pm_dup = attach_pm(cus_dup.id)
dup_user = users[3] if len(users) > 3 else users[0]

sub1 = create_sub(cus_dup.id, price_map["starter"]["provider_price_id"],
                  {"user_id": str(dup_user["id"]), "plan": "starter"}, pm_dup)
time.sleep(3)

sub2 = create_sub(cus_dup.id, price_map["pro"]["provider_price_id"],
                  {"user_id": str(dup_user["id"]), "plan": "pro"}, pm_dup)
time.sleep(3)

rows = query_db("SELECT * FROM subscriptions WHERE user_id = ? ORDER BY created_at", (dup_user["id"],))
ok(f"Two subscriptions exist: {len(rows)} rows") if len(rows) >= 2 else fail(f"Expected 2+ rows, got {len(rows)}")

# get_subscription returns most recent
latest = query_db("SELECT * FROM subscriptions WHERE user_id = ? ORDER BY created_at DESC LIMIT 1", (dup_user["id"],))
if latest and latest[0]["plan"] == "pro":
    ok("Latest subscription is 'pro' (upgrade scenario)")
else:
    fail(f"Latest plan: {latest[0]['plan'] if latest else '?'}")

cancel_sub(sub1.id)
cancel_sub(sub2.id)


section("6d. Edge case — rapid create + cancel (race condition)")

cus_race = get_or_create_customer("e2e-race@sandbox.padelnomics.com", "E2E Race")
pm_race = attach_pm(cus_race.id)
race_user = users[4] if len(users) > 4 else users[0]

sub = create_sub(cus_race.id, price_map["starter"]["provider_price_id"],
                 {"user_id": str(race_user["id"]), "plan": "starter"}, pm_race)
# Cancel immediately — webhooks may arrive out of order
stripe.Subscription.cancel(sub.id)
ok(f"Created and immediately cancelled: {sub.id}")

time.sleep(8)  # Wait for both webhooks

rows = query_db("SELECT * FROM subscriptions WHERE provider_subscription_id = ?", (sub.id,))
if rows:
    ok(f"Final DB status: {rows[0]['status']}")
else:
    ok("No DB row (created webhook may have arrived after deleted)")

result = subprocess.run(
    ["curl", "-s", "-o", "/dev/null", "-w", "%{http_code}", "http://localhost:5000/"],
    capture_output=True, text=True, timeout=5,
)
ok("Server survived race condition") if result.stdout.strip() in ("200", "301", "302") else fail("Server crashed!")


# ═══════════════════════════════════════════════════════════
# 7. BILLING PORTAL
# ═══════════════════════════════════════════════════════════

section("7. Billing Portal session")

try:
    portal = stripe.billing_portal.Session.create(
        customer=cus_starter.id,
        return_url=f"{tunnel_url}/billing/success",
    )
    ok(f"Portal URL: {portal.url[:50]}...")
except stripe.StripeError as e:
    fail(f"Portal failed: {e}")


# ═══════════════════════════════════════════════════════════
# 8. ONE-TIME PAYMENTS (via PaymentIntent — simulates completed checkout)
# ═══════════════════════════════════════════════════════════

section("8. One-time payments (PaymentIntents — all credit packs + boosts + PDF)")

cus_buyer = get_or_create_customer("e2e-buyer@sandbox.padelnomics.com", "E2E Buyer")
pm_buyer = attach_pm(cus_buyer.id)

one_time_products = [
    ("credits_25", 9900),
    ("credits_50", 17900),
    ("credits_100", 32900),
    ("credits_250", 74900),
    ("boost_sticky_week", 7900),
    ("boost_sticky_month", 19900),
    ("business_plan", 14900),
]

for key, amount_cents in one_time_products:
    try:
        pi = stripe.PaymentIntent.create(
            amount=amount_cents,
            currency="eur",
            customer=cus_buyer.id,
            payment_method=pm_buyer,
            confirm=True,
            automatic_payment_methods={"enabled": True, "allow_redirects": "never"},
            metadata={
                "user_id": str(test_user["id"]),
                "supplier_id": str(test_supplier["id"]),
                "plan": key,
            },
        )
        if pi.status == "succeeded":
            ok(f"{key}: \u20ac{amount_cents/100:.2f} succeeded ({pi.id[:20]}...)")
        else:
            fail(f"{key}: status={pi.status}")
    except stripe.StripeError as e:
        fail(f"{key}: {e}")

# Note: PaymentIntents don't trigger checkout.session.completed webhooks.
# The actual credit/boost/PDF creation requires a Checkout Session completion,
# which can only happen via browser. These tests verify the payments succeed.
print("  (PaymentIntents succeed but don't trigger checkout webhooks —")
print("   credit/boost/PDF creation requires browser checkout completion)")


# ═══════════════════════════════════════════════════════════
# 9. DECLINED CARDS — different failure modes
# ═══════════════════════════════════════════════════════════

section("9. Declined card scenarios (PaymentIntent level)")

decline_tokens = [
    ("tok_chargeDeclined", "generic decline"),
    ("tok_chargeDeclinedInsufficientFunds", "insufficient funds"),
    ("tok_chargeDeclinedExpiredCard", "expired card"),
    ("tok_chargeDeclinedProcessingError", "processing error"),
]

for token, description in decline_tokens:
    try:
        pm = stripe.PaymentMethod.create(type="card", card={"token": token})
        stripe.PaymentMethod.attach(pm.id, customer=cus_buyer.id)
        pi = stripe.PaymentIntent.create(
            amount=1900,
            currency="eur",
            customer=cus_buyer.id,
            payment_method=pm.id,
            confirm=True,
            automatic_payment_methods={"enabled": True, "allow_redirects": "never"},
        )
        fail(f"{description}: should have been declined but succeeded")
    except stripe.CardError as e:
        ok(f"{description}: correctly declined ({e.code})")
    except stripe.StripeError as e:
        ok(f"{description}: rejected ({type(e).__name__})")


# ═══════════════════════════════════════════════════════════
# Summary
# ═══════════════════════════════════════════════════════════

section("RESULTS")

total = passed + failed
print(f"\n  {passed}/{total} passed, {failed} failed\n")

if errors:
    print("  Failures:")
    for err in errors:
        print(f"    - {err}")
    print()

# Final cleanup: cancel any remaining subs
for sid in cleanup_sub_ids:
    try:
        stripe.Subscription.cancel(sid)
    except Exception:
        pass

sys.exit(1 if failed else 0)