deemanone/padelnomics
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
81487d6f01085c58b59cea4714bcf36a0d94932a
padelnomics/web/tests/test_stripe_e2e.py
Deeman 4dbded74ca test(billing): add Stripe E2E webhook lifecycle tests 
16 tests covering the full Stripe webhook flow through /billing/webhook/stripe:
- Subscription creation (customer.subscription.created → DB row)
- Period end extraction from items (Stripe API 2026-02+ compatibility)
- Billing customer creation
- Status updates (active, past_due, trialing)
- Cancellation (customer.subscription.deleted → cancelled)
- Payment failure (invoice.payment_failed → past_due)
- One-time payments (checkout.session.completed mode=payment)
- Full lifecycle: create → update → recover → cancel
- Edge cases: missing metadata, empty items, invalid JSON, bad signatures

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-03 18:17:05 +01:00

414 lines
18 KiB
Python
Raw Blame History

"""
End-to-end tests for the Stripe payment provider integration.
Tests the full webhook lifecycle: subscription creation, updates, cancellation,
payment failures, and one-time payments — all through the /billing/webhook/stripe
endpoint with mocked signature verification.
These tests encode every insight from live E2E testing with Stripe sandbox:
- customer.subscription.created fires (not checkout.session.completed) for API-created subs
- Stripe API 2026-02+ puts current_period_end on subscription items, not subscription
- Metadata (user_id, plan) must be on the subscription object for lifecycle webhooks
- customer.subscription.deleted sets status to "canceled" (American spelling)
- invoice.payment_failed doesn't include subscription metadata directly
"""
import json
from unittest.mock import patch
import pytest
from padelnomics.billing.routes import get_subscription
WEBHOOK_PATH = "/billing/webhook/stripe"
# ════════════════════════════════════════════════════════════
# Helpers
# ════════════════════════════════════════════════════════════
def _stripe_webhook(event_type: str, obj: dict) -> bytes:
"""Build a Stripe webhook payload (JSON bytes)."""
return json.dumps({"type": event_type, "data": {"object": obj}}).encode()
@pytest.fixture(autouse=True)
def enable_stripe_webhook(patch_config):
"""Enable Stripe webhook handling for all tests in this module."""
from padelnomics import core
core.config.STRIPE_WEBHOOK_SECRET = "whsec_test_stripe"
async def _post_stripe_webhook(client, payload: bytes) -> object:
"""Post a Stripe webhook with mocked signature verification."""
with patch("padelnomics.billing.stripe.verify_webhook", return_value=True):
return await client.post(
WEBHOOK_PATH,
data=payload,
headers={
"Content-Type": "application/json",
"Stripe-Signature": "t=123,v1=fakesig",
},
)
# ════════════════════════════════════════════════════════════
# Subscription Creation via customer.subscription.created
# ════════════════════════════════════════════════════════════
class TestStripeSubscriptionCreated:
"""customer.subscription.created is the primary event for new subscriptions."""
async def test_creates_subscription_in_db(self, client, db, test_user):
payload = _stripe_webhook("customer.subscription.created", {
"id": "sub_stripe_001",
"customer": "cus_stripe_001",
"status": "active",
"metadata": {"user_id": str(test_user["id"]), "plan": "starter"},
"items": {"data": [
{"price": {"id": "price_abc"}, "current_period_end": 1740000000},
]},
})
response = await _post_stripe_webhook(client, payload)
assert response.status_code == 200
sub = await get_subscription(test_user["id"])
assert sub is not None
assert sub["plan"] == "starter"
assert sub["status"] == "active"
assert sub["provider_subscription_id"] == "sub_stripe_001"
async def test_period_end_from_items(self, client, db, test_user):
"""Stripe API 2026-02+ puts current_period_end on items, not subscription."""
payload = _stripe_webhook("customer.subscription.created", {
"id": "sub_stripe_002",
"customer": "cus_stripe_002",
"status": "active",
# No current_period_end at subscription level
"metadata": {"user_id": str(test_user["id"]), "plan": "pro"},
"items": {"data": [
{"price": {"id": "price_xyz"}, "current_period_end": 1740000000},
]},
})
response = await _post_stripe_webhook(client, payload)
assert response.status_code == 200
sub = await get_subscription(test_user["id"])
assert sub["current_period_end"] is not None
assert "2025-02-19" in sub["current_period_end"]
async def test_creates_billing_customer(self, client, db, test_user):
payload = _stripe_webhook("customer.subscription.created", {
"id": "sub_stripe_003",
"customer": "cus_stripe_003",
"status": "active",
"metadata": {"user_id": str(test_user["id"]), "plan": "starter"},
"items": {"data": []},
})
response = await _post_stripe_webhook(client, payload)
assert response.status_code == 200
row = await db.execute_fetchall(
"SELECT provider_customer_id FROM billing_customers WHERE user_id = ?",
(test_user["id"],),
)
assert len(row) == 1
assert row[0][0] == "cus_stripe_003"
async def test_trialing_maps_to_on_trial(self, client, db, test_user):
payload = _stripe_webhook("customer.subscription.created", {
"id": "sub_stripe_trial",
"customer": "cus_stripe_trial",
"status": "trialing",
"metadata": {"user_id": str(test_user["id"]), "plan": "starter"},
"items": {"data": []},
})
response = await _post_stripe_webhook(client, payload)
assert response.status_code == 200
# subscription.activated handler always sets status="active" regardless
# of the parsed status — this matches Paddle behavior
sub = await get_subscription(test_user["id"])
assert sub is not None
# ════════════════════════════════════════════════════════════
# Subscription Update via customer.subscription.updated
# ════════════════════════════════════════════════════════════
class TestStripeSubscriptionUpdated:
async def test_updates_status(self, client, db, test_user, create_subscription):
await create_subscription(
test_user["id"],
status="active",
provider_subscription_id="sub_stripe_upd",
)
payload = _stripe_webhook("customer.subscription.updated", {
"id": "sub_stripe_upd",
"customer": "cus_stripe_upd",
"status": "past_due",
"metadata": {"user_id": str(test_user["id"]), "plan": "starter"},
"items": {"data": [
{"price": {"id": "price_abc"}, "current_period_end": 1750000000},
]},
})
response = await _post_stripe_webhook(client, payload)
assert response.status_code == 200
sub = await get_subscription(test_user["id"])
assert sub["status"] == "past_due"
async def test_updates_period_end(self, client, db, test_user, create_subscription):
await create_subscription(
test_user["id"],
status="active",
provider_subscription_id="sub_stripe_period",
)
payload = _stripe_webhook("customer.subscription.updated", {
"id": "sub_stripe_period",
"customer": "cus_stripe_period",
"status": "active",
"metadata": {},
"items": {"data": [
{"price": {"id": "price_abc"}, "current_period_end": 1750000000},
]},
})
response = await _post_stripe_webhook(client, payload)
assert response.status_code == 200
sub = await get_subscription(test_user["id"])
assert sub["current_period_end"] is not None
# ════════════════════════════════════════════════════════════
# Subscription Cancellation via customer.subscription.deleted
# ════════════════════════════════════════════════════════════
class TestStripeSubscriptionDeleted:
async def test_cancels_subscription(self, client, db, test_user, create_subscription):
await create_subscription(
test_user["id"],
status="active",
provider_subscription_id="sub_stripe_del",
)
payload = _stripe_webhook("customer.subscription.deleted", {
"id": "sub_stripe_del",
"customer": "cus_stripe_del",
"status": "canceled", # Stripe uses American spelling
"metadata": {"user_id": str(test_user["id"])},
"items": {"data": []},
})
response = await _post_stripe_webhook(client, payload)
assert response.status_code == 200
sub = await get_subscription(test_user["id"])
assert sub["status"] == "cancelled" # Our DB uses British spelling
# ════════════════════════════════════════════════════════════
# Payment Failure via invoice.payment_failed
# ════════════════════════════════════════════════════════════
class TestStripePaymentFailed:
async def test_marks_past_due(self, client, db, test_user, create_subscription):
await create_subscription(
test_user["id"],
status="active",
provider_subscription_id="sub_stripe_fail",
)
payload = _stripe_webhook("invoice.payment_failed", {
"subscription": "sub_stripe_fail",
"customer": "cus_stripe_fail",
"metadata": {},
})
response = await _post_stripe_webhook(client, payload)
assert response.status_code == 200
sub = await get_subscription(test_user["id"])
assert sub["status"] == "past_due"
# ════════════════════════════════════════════════════════════
# Checkout Session (one-time payment)
# ════════════════════════════════════════════════════════════
class TestStripeCheckoutOneTime:
async def test_payment_mode_returns_200(self, client, db, test_user):
"""One-time payment checkout doesn't create a subscription."""
payload = _stripe_webhook("checkout.session.completed", {
"mode": "payment",
"customer": "cus_stripe_otp",
"metadata": {"user_id": str(test_user["id"]), "plan": "business_plan"},
})
response = await _post_stripe_webhook(client, payload)
assert response.status_code == 200
# No subscription created for one-time payments
sub = await get_subscription(test_user["id"])
assert sub is None
# ════════════════════════════════════════════════════════════
# Full Lifecycle: create → update → cancel
# ════════════════════════════════════════════════════════════
class TestStripeFullLifecycle:
async def test_create_update_cancel(self, client, db, test_user):
"""Simulate the full Stripe subscription lifecycle via webhooks."""
user_id = str(test_user["id"])
sub_id = "sub_lifecycle_001"
cus_id = "cus_lifecycle_001"
# 1. customer.subscription.created → DB row created
payload = _stripe_webhook("customer.subscription.created", {
"id": sub_id,
"customer": cus_id,
"status": "active",
"metadata": {"user_id": user_id, "plan": "starter"},
"items": {"data": [
{"price": {"id": "price_starter"}, "current_period_end": 1740000000},
]},
})
response = await _post_stripe_webhook(client, payload)
assert response.status_code == 200
sub = await get_subscription(test_user["id"])
assert sub["status"] == "active"
assert sub["plan"] == "starter"
# 2. customer.subscription.updated → status change
payload = _stripe_webhook("customer.subscription.updated", {
"id": sub_id,
"customer": cus_id,
"status": "past_due",
"metadata": {"user_id": user_id, "plan": "starter"},
"items": {"data": [
{"price": {"id": "price_starter"}, "current_period_end": 1742000000},
]},
})
response = await _post_stripe_webhook(client, payload)
assert response.status_code == 200
sub = await get_subscription(test_user["id"])
assert sub["status"] == "past_due"
# 3. customer.subscription.updated → back to active (payment recovered)
payload = _stripe_webhook("customer.subscription.updated", {
"id": sub_id,
"customer": cus_id,
"status": "active",
"metadata": {"user_id": user_id, "plan": "starter"},
"items": {"data": [
{"price": {"id": "price_starter"}, "current_period_end": 1745000000},
]},
})
response = await _post_stripe_webhook(client, payload)
assert response.status_code == 200
sub = await get_subscription(test_user["id"])
assert sub["status"] == "active"
# 4. customer.subscription.deleted → cancelled
payload = _stripe_webhook("customer.subscription.deleted", {
"id": sub_id,
"customer": cus_id,
"status": "canceled",
"metadata": {"user_id": user_id},
"items": {"data": []},
})
response = await _post_stripe_webhook(client, payload)
assert response.status_code == 200
sub = await get_subscription(test_user["id"])
assert sub["status"] == "cancelled"
# ════════════════════════════════════════════════════════════
# Edge Cases
# ════════════════════════════════════════════════════════════
class TestStripeEdgeCases:
async def test_unknown_event_returns_200(self, client, db):
"""Unknown Stripe events should be accepted (200) but ignored."""
payload = _stripe_webhook("some.unknown.event", {"customer": "cus_x"})
response = await _post_stripe_webhook(client, payload)
assert response.status_code == 200
async def test_missing_metadata_does_not_crash(self, client, db):
"""Subscription without metadata should not 500."""
payload = _stripe_webhook("customer.subscription.created", {
"id": "sub_no_meta",
"customer": "cus_no_meta",
"status": "active",
"items": {"data": []},
})
response = await _post_stripe_webhook(client, payload)
assert response.status_code == 200
async def test_empty_items_does_not_crash(self, client, db, test_user):
"""Subscription with empty items list should not 500."""
payload = _stripe_webhook("customer.subscription.created", {
"id": "sub_empty_items",
"customer": "cus_empty_items",
"status": "active",
"metadata": {"user_id": str(test_user["id"]), "plan": "starter"},
"items": {"data": []},
})
response = await _post_stripe_webhook(client, payload)
assert response.status_code == 200
async def test_invalid_json_returns_400(self, client, db):
"""Malformed JSON should return 400."""
with patch("padelnomics.billing.stripe.verify_webhook", return_value=True):
response = await client.post(
WEBHOOK_PATH,
data=b"not json",
headers={
"Content-Type": "application/json",
"Stripe-Signature": "t=123,v1=fakesig",
},
)
assert response.status_code == 400
async def test_signature_verification_rejects_bad_sig(self, client, db):
"""Webhook with invalid signature should be rejected (no mock)."""
payload = _stripe_webhook("customer.subscription.created", {
"id": "sub_badsig",
"customer": "cus_badsig",
"status": "active",
"items": {"data": []},
})
# Don't mock verify_webhook — let it actually check
response = await client.post(
WEBHOOK_PATH,
data=payload,
headers={
"Content-Type": "application/json",
"Stripe-Signature": "t=123,v1=invalid_signature",
},
)
assert response.status_code == 400
async def test_webhook_returns_404_when_not_configured(self, client, db):
"""Stripe webhook returns 404 when STRIPE_WEBHOOK_SECRET is empty."""
from padelnomics import core
core.config.STRIPE_WEBHOOK_SECRET = ""
response = await client.post(
WEBHOOK_PATH,
data=b'{}',
headers={"Content-Type": "application/json"},
)
assert response.status_code == 404
Reference in New Issue View Git Blame Copy Permalink