padelnomics/scripts/test_stripe_sandbox.py

"""
Stripe Sandbox Integration Test — verifies all products work end-to-end.

Creates multiple test customers with different personas, tests:
- Checkout session creation for every product
- Subscription creation + cancellation lifecycle
- One-time payment intents
- Price/product consistency

Run: uv run python scripts/test_stripe_sandbox.py
"""

import os
import sys
import time

from dotenv import load_dotenv

load_dotenv()

import stripe

STRIPE_SECRET_KEY = os.getenv("STRIPE_SECRET_KEY", "") or os.getenv("STRIPE_API_PRIVATE_KEY", "")
if not STRIPE_SECRET_KEY:
    print("ERROR: STRIPE_SECRET_KEY / STRIPE_API_PRIVATE_KEY not set in .env")
    sys.exit(1)

stripe.api_key = STRIPE_SECRET_KEY
stripe.max_network_retries = 2

BASE_URL = os.getenv("BASE_URL", "http://localhost:5000")

# ═══════════════════════════════════════════════════════════
# Expected product catalog — must match setup_stripe.py
# ═══════════════════════════════════════════════════════════

EXPECTED_PRODUCTS = {
    "Supplier Growth": {"price_cents": 19900, "billing": "subscription", "interval": "month"},
    "Supplier Growth (Yearly)": {"price_cents": 179900, "billing": "subscription", "interval": "year"},
    "Supplier Pro": {"price_cents": 49900, "billing": "subscription", "interval": "month"},
    "Supplier Pro (Yearly)": {"price_cents": 449900, "billing": "subscription", "interval": "year"},
    "Boost: Logo": {"price_cents": 2900, "billing": "subscription", "interval": "month"},
    "Boost: Highlight": {"price_cents": 3900, "billing": "subscription", "interval": "month"},
    "Boost: Verified Badge": {"price_cents": 4900, "billing": "subscription", "interval": "month"},
    "Boost: Custom Card Color": {"price_cents": 5900, "billing": "subscription", "interval": "month"},
    "Boost: Sticky Top 1 Week": {"price_cents": 7900, "billing": "one_time"},
    "Boost: Sticky Top 1 Month": {"price_cents": 19900, "billing": "one_time"},
    "Credit Pack 25": {"price_cents": 9900, "billing": "one_time"},
    "Credit Pack 50": {"price_cents": 17900, "billing": "one_time"},
    "Credit Pack 100": {"price_cents": 32900, "billing": "one_time"},
    "Credit Pack 250": {"price_cents": 74900, "billing": "one_time"},
    "Padel Business Plan (PDF)": {"price_cents": 14900, "billing": "one_time"},
    "Planner Starter": {"price_cents": 1900, "billing": "subscription", "interval": "month"},
    "Planner Pro": {"price_cents": 4900, "billing": "subscription", "interval": "month"},
}

# Test customer personas
TEST_CUSTOMERS = [
    {"email": "planner-starter@sandbox.padelnomics.com", "name": "Anna Planner (Starter)"},
    {"email": "planner-pro@sandbox.padelnomics.com", "name": "Ben Planner (Pro)"},
    {"email": "supplier-growth@sandbox.padelnomics.com", "name": "Carlos Supplier (Growth)"},
    {"email": "supplier-pro@sandbox.padelnomics.com", "name": "Diana Supplier (Pro)"},
    {"email": "one-time-buyer@sandbox.padelnomics.com", "name": "Eva Buyer (Credits+Boosts)"},
]

passed = 0
failed = 0
errors = []


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


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


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


# ═══════════════════════════════════════════════════════════
# Phase 1: Verify all products and prices exist
# ═══════════════════════════════════════════════════════════

section("Phase 1: Product & Price Verification")

products = list(stripe.Product.list(limit=100, active=True).auto_paging_iter())
product_map = {}  # name -> {product_id, price_id, price_amount, price_type, interval}

for product in products:
    prices = stripe.Price.list(product=product.id, active=True, limit=1)
    if not prices.data:
        continue
    price = prices.data[0]
    product_map[product.name] = {
        "product_id": product.id,
        "price_id": price.id,
        "price_amount": price.unit_amount,
        "price_type": price.type,
        "interval": price.recurring.interval if price.recurring else None,
    }

for name, expected in EXPECTED_PRODUCTS.items():
    if name not in product_map:
        fail(f"MISSING product: {name}")
        continue

    actual = product_map[name]
    if actual["price_amount"] != expected["price_cents"]:
        fail(f"{name}: price {actual['price_amount']} != expected {expected['price_cents']}")
    elif expected["billing"] == "subscription" and actual["price_type"] != "recurring":
        fail(f"{name}: expected recurring, got {actual['price_type']}")
    elif expected["billing"] == "one_time" and actual["price_type"] != "one_time":
        fail(f"{name}: expected one_time, got {actual['price_type']}")
    elif expected.get("interval") and actual["interval"] != expected["interval"]:
        fail(f"{name}: interval {actual['interval']} != expected {expected['interval']}")
    else:
        ok(f"{name}: €{actual['price_amount']/100:.2f} ({actual['price_type']}"
           f"{', ' + actual['interval'] if actual['interval'] else ''})")

extra_products = set(product_map.keys()) - set(EXPECTED_PRODUCTS.keys())
if extra_products:
    print(f"\n  ℹ Extra products in Stripe (not in catalog): {extra_products}")


# ═══════════════════════════════════════════════════════════
# Phase 2: Create test customers (idempotent)
# ═══════════════════════════════════════════════════════════

section("Phase 2: Create Test Customers")

customer_ids = {}  # email -> customer_id

for persona in TEST_CUSTOMERS:
    existing = stripe.Customer.list(email=persona["email"], limit=1)
    if existing.data:
        cus = existing.data[0]
        ok(f"Reusing: {persona['name']} ({cus.id})")
    else:
        cus = stripe.Customer.create(
            email=persona["email"],
            name=persona["name"],
            metadata={"test": "true", "persona": persona["name"]},
        )
        ok(f"Created: {persona['name']} ({cus.id})")
    customer_ids[persona["email"]] = cus.id


# ═══════════════════════════════════════════════════════════
# Phase 3: Test Checkout Sessions for every product
# ═══════════════════════════════════════════════════════════

section("Phase 3: Checkout Session Creation (all products)")

success_url = f"{BASE_URL}/billing/success?session_id={{CHECKOUT_SESSION_ID}}"
cancel_url = f"{BASE_URL}/billing/pricing"

# Use the first customer for checkout tests
checkout_customer = customer_ids["planner-starter@sandbox.padelnomics.com"]

for name, info in product_map.items():
    if name not in EXPECTED_PRODUCTS:
        continue

    mode = "subscription" if info["price_type"] == "recurring" else "payment"
    try:
        session = stripe.checkout.Session.create(
            mode=mode,
            customer=checkout_customer,
            line_items=[{"price": info["price_id"], "quantity": 1}],
            metadata={"user_id": "999", "plan": name, "test": "true"},
            success_url=success_url,
            cancel_url=cancel_url,
        )
        ok(f"Checkout ({mode}): {name} -> {session.id[:30]}...")
    except stripe.StripeError as e:
        fail(f"Checkout FAILED for {name}: {e.user_message or str(e)}")


# ═══════════════════════════════════════════════════════════
# Phase 4: Subscription lifecycle tests (per persona)
# ═══════════════════════════════════════════════════════════

section("Phase 4: Subscription Lifecycle Tests")

created_subs = []

# Cache: customer_id -> payment_method_id
_customer_pms = {}


def _ensure_payment_method(cus_id):
    """Create and attach a test Visa card to a customer (cached)."""
    if cus_id in _customer_pms:
        return _customer_pms[cus_id]

    pm = stripe.PaymentMethod.create(type="card", card={"token": "tok_visa"})
    stripe.PaymentMethod.attach(pm.id, customer=cus_id)
    stripe.Customer.modify(
        cus_id,
        invoice_settings={"default_payment_method": pm.id},
    )
    _customer_pms[cus_id] = pm.id
    return pm.id


def test_subscription(customer_email, product_name, user_id, extra_metadata=None):
    """Create a subscription, verify it's active, then cancel it."""
    cus_id = customer_ids[customer_email]
    info = product_map.get(product_name)
    if not info:
        fail(f"Product not found: {product_name}")
        return

    metadata = {"user_id": str(user_id), "plan": product_name, "test": "true"}
    if extra_metadata:
        metadata.update(extra_metadata)

    pm_id = _ensure_payment_method(cus_id)

    # Create subscription
    sub = stripe.Subscription.create(
        customer=cus_id,
        items=[{"price": info["price_id"]}],
        metadata=metadata,
        default_payment_method=pm_id,
    )
    created_subs.append(sub.id)

    if sub.status == "active":
        ok(f"Sub created: {product_name} for {customer_email} -> {sub.id} (active)")
    else:
        fail(f"Sub status unexpected: {product_name} -> {sub.status} (expected active)")

    # Verify subscription items
    items = sub["items"]["data"]
    if len(items) == 1 and items[0]["price"]["id"] == info["price_id"]:
        ok(f"Sub items correct: price={info['price_id'][:20]}...")
    else:
        fail(f"Sub items mismatch for {product_name}")

    # Cancel at period end
    updated = stripe.Subscription.modify(sub.id, cancel_at_period_end=True)
    if updated.cancel_at_period_end:
        ok(f"Cancel scheduled: {product_name} (cancel_at_period_end=True)")
    else:
        fail(f"Cancel failed for {product_name}")

    # Immediately cancel to clean up
    deleted = stripe.Subscription.cancel(sub.id)
    if deleted.status == "canceled":
        ok(f"Cancelled: {product_name} -> {deleted.status}")
    else:
        fail(f"Final cancel status: {product_name} -> {deleted.status}")


# Planner Starter
test_subscription(
    "planner-starter@sandbox.padelnomics.com", "Planner Starter", user_id=101,
)

# Planner Pro
test_subscription(
    "planner-pro@sandbox.padelnomics.com", "Planner Pro", user_id=102,
)

# Supplier Growth (monthly)
test_subscription(
    "supplier-growth@sandbox.padelnomics.com", "Supplier Growth", user_id=103,
    extra_metadata={"supplier_id": "201"},
)

# Supplier Pro (monthly)
test_subscription(
    "supplier-pro@sandbox.padelnomics.com", "Supplier Pro", user_id=104,
    extra_metadata={"supplier_id": "202"},
)


# ═══════════════════════════════════════════════════════════
# Phase 5: One-time payment tests
# ═══════════════════════════════════════════════════════════

section("Phase 5: One-Time Payment Tests")

buyer_id = customer_ids["one-time-buyer@sandbox.padelnomics.com"]
buyer_pm = _ensure_payment_method(buyer_id)

ONE_TIME_PRODUCTS = [
    "Credit Pack 25",
    "Credit Pack 50",
    "Credit Pack 100",
    "Credit Pack 250",
    "Boost: Sticky Top 1 Week",
    "Boost: Sticky Top 1 Month",
    "Padel Business Plan (PDF)",
]

for product_name in ONE_TIME_PRODUCTS:
    info = product_map.get(product_name)
    if not info:
        fail(f"Product not found: {product_name}")
        continue

    try:
        pi = stripe.PaymentIntent.create(
            amount=info["price_amount"],
            currency="eur",
            customer=buyer_id,
            payment_method=buyer_pm,
            confirm=True,
            automatic_payment_methods={"enabled": True, "allow_redirects": "never"},
            metadata={
                "user_id": "105",
                "supplier_id": "203",
                "plan": product_name,
                "test": "true",
            },
        )
        if pi.status == "succeeded":
            ok(f"Payment: {product_name} -> €{info['price_amount']/100:.2f} ({pi.id[:25]}...)")
        else:
            fail(f"Payment status: {product_name} -> {pi.status}")
    except stripe.StripeError as e:
        fail(f"Payment FAILED for {product_name}: {e.user_message or str(e)}")


# ═══════════════════════════════════════════════════════════
# Phase 6: Boost subscription add-ons
# ═══════════════════════════════════════════════════════════

section("Phase 6: Boost Add-on Subscriptions")

BOOST_PRODUCTS = [
    "Boost: Logo",
    "Boost: Highlight",
    "Boost: Verified Badge",
    "Boost: Custom Card Color",
]

boost_customer = customer_ids["supplier-pro@sandbox.padelnomics.com"]
boost_pm = _ensure_payment_method(boost_customer)

for product_name in BOOST_PRODUCTS:
    info = product_map.get(product_name)
    if not info:
        fail(f"Product not found: {product_name}")
        continue

    try:
        sub = stripe.Subscription.create(
            customer=boost_customer,
            items=[{"price": info["price_id"]}],
            metadata={
                "user_id": "104",
                "supplier_id": "202",
                "plan": product_name,
                "test": "true",
            },
            default_payment_method=boost_pm,
        )
        created_subs.append(sub.id)

        if sub.status == "active":
            ok(f"Boost sub: {product_name} -> €{info['price_amount']/100:.2f}/mo ({sub.id[:25]}...)")
        else:
            fail(f"Boost sub status: {product_name} -> {sub.status}")

        # Clean up
        stripe.Subscription.cancel(sub.id)
    except stripe.StripeError as e:
        fail(f"Boost sub FAILED for {product_name}: {e.user_message or str(e)}")


# ═══════════════════════════════════════════════════════════
# Phase 7: Billing Portal access
# ═══════════════════════════════════════════════════════════

section("Phase 7: Billing Portal")

try:
    portal = stripe.billing_portal.Session.create(
        customer=checkout_customer,
        return_url=f"{BASE_URL}/billing/success",
    )
    ok(f"Portal URL generated: {portal.url[:50]}...")
except stripe.StripeError as e:
    fail(f"Portal creation failed: {e.user_message or str(e)}")


# ═══════════════════════════════════════════════════════════
# 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()

# Customer summary
print("  Test customers in sandbox:")
for persona in TEST_CUSTOMERS:
    cid = customer_ids.get(persona["email"], "?")
    print(f"    {persona['name']}: {cid}")

print()
sys.exit(1 if failed else 0)