fix(billing): handle customer.subscription.created webhook + test isolation

- Add customer.subscription.created → subscription.activated mapping in
  stripe.parse_webhook so direct API subscription creation also creates DB rows
- Add customer.subscription.created to setup_stripe.py enabled_events
- Pin PAYMENT_PROVIDER=paddle and STRIPE_WEBHOOK_SECRET="" in test conftest
  so billing tests don't hit real Stripe API when env has Stripe keys
- Add 8 unit tests for stripe.parse_webhook covering all event types

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Deeman
2026-03-03 17:29:13 +01:00
parent 7c5fa86fb8
commit 72c4de91b0
4 changed files with 119 additions and 1 deletions

View File

@@ -152,6 +152,7 @@ def parse_webhook(payload: bytes) -> dict:
Maps Stripe event types to the shared format used by _handle_webhook_event():
- checkout.session.completed (mode=subscription) → subscription.activated
- customer.subscription.created → subscription.activated
- customer.subscription.updated → subscription.updated
- customer.subscription.deleted → subscription.canceled
- invoice.payment_failed → subscription.past_due
@@ -218,6 +219,23 @@ def parse_webhook(payload: bytes) -> dict:
"custom_data": metadata,
}
elif stripe_type == "customer.subscription.created":
# New subscription — map to subscription.activated so the handler creates the DB row
status = _map_stripe_status(obj.get("status", ""))
return {
"event_type": "subscription.activated",
"subscription_id": obj.get("id", ""),
"customer_id": str(customer_id),
"user_id": user_id,
"supplier_id": supplier_id,
"plan": plan,
"status": status,
"current_period_end": _unix_to_iso(obj.get("current_period_end")),
"data": obj,
"items": _extract_sub_items(obj),
"custom_data": metadata,
}
elif stripe_type == "customer.subscription.updated":
status = _map_stripe_status(obj.get("status", ""))
return {

View File

@@ -189,6 +189,7 @@ def create(conn):
webhook_url = f"{BASE_URL}/billing/webhook/stripe"
enabled_events = [
"checkout.session.completed",
"customer.subscription.created",
"customer.subscription.updated",
"customer.subscription.deleted",
"invoice.payment_failed",

View File

@@ -213,8 +213,10 @@ def patch_config():
"""Set test Paddle config values."""
original_values = {}
test_values = {
"PAYMENT_PROVIDER": "paddle", # default to Paddle so mocks work
"PADDLE_API_KEY": "test_api_key_123",
"PADDLE_WEBHOOK_SECRET": "whsec_test_secret",
"STRIPE_WEBHOOK_SECRET": "", # no Stripe in default tests
"RESEND_API_KEY": "", # never send real emails in tests
"BASE_URL": "http://localhost:5000",
"DEBUG": True,

View File

@@ -1,5 +1,5 @@
"""
Integration tests for Paddle webhook handling.
Integration tests for Paddle webhook handling + Stripe parse_webhook unit tests.
Covers signature verification, event parsing, subscription lifecycle transitions, and Hypothesis fuzzing.
"""
import json
@@ -10,6 +10,7 @@ from hypothesis import HealthCheck, given
from hypothesis import settings as h_settings
from hypothesis import strategies as st
from padelnomics.billing.routes import get_subscription
from padelnomics.billing.stripe import parse_webhook as stripe_parse_webhook
WEBHOOK_PATH = "/billing/webhook/paddle"
SIG_HEADER = "Paddle-Signature"
@@ -306,3 +307,99 @@ class TestWebhookHypothesis:
headers={SIG_HEADER: sig, "Content-Type": "application/json"},
)
assert response.status_code < 500
# ════════════════════════════════════════════════════════════
# Stripe parse_webhook unit tests
# ════════════════════════════════════════════════════════════
def _stripe_event(event_type, obj, metadata=None):
"""Build a minimal Stripe event payload."""
if metadata:
obj["metadata"] = metadata
return json.dumps({"type": event_type, "data": {"object": obj}}).encode()
class TestStripeParseWebhook:
def test_subscription_created_maps_to_activated(self):
payload = _stripe_event(
"customer.subscription.created",
{"id": "sub_123", "customer": "cus_456", "status": "active", "current_period_end": 1740000000,
"items": {"data": [{"price": {"id": "price_abc"}}]}},
metadata={"user_id": "42", "plan": "starter"},
)
ev = stripe_parse_webhook(payload)
assert ev["event_type"] == "subscription.activated"
assert ev["subscription_id"] == "sub_123"
assert ev["customer_id"] == "cus_456"
assert ev["user_id"] == "42"
assert ev["plan"] == "starter"
assert ev["status"] == "active"
assert ev["items"] == [{"price": {"id": "price_abc"}}]
def test_subscription_updated_maps_correctly(self):
payload = _stripe_event(
"customer.subscription.updated",
{"id": "sub_123", "customer": "cus_456", "status": "past_due", "current_period_end": 1740000000,
"items": {"data": [{"price": {"id": "price_abc"}}]}},
metadata={"user_id": "42", "plan": "pro"},
)
ev = stripe_parse_webhook(payload)
assert ev["event_type"] == "subscription.updated"
assert ev["status"] == "past_due"
def test_subscription_deleted_maps_to_canceled(self):
payload = _stripe_event(
"customer.subscription.deleted",
{"id": "sub_123", "customer": "cus_456", "status": "canceled", "current_period_end": 1740000000,
"items": {"data": []}},
metadata={"user_id": "42"},
)
ev = stripe_parse_webhook(payload)
assert ev["event_type"] == "subscription.canceled"
assert ev["status"] == "cancelled"
def test_invoice_payment_failed_maps_to_past_due(self):
payload = _stripe_event(
"invoice.payment_failed",
{"subscription": "sub_123", "customer": "cus_456"},
metadata={},
)
ev = stripe_parse_webhook(payload)
assert ev["event_type"] == "subscription.past_due"
assert ev["status"] == "past_due"
def test_checkout_session_subscription_maps_to_activated(self):
payload = _stripe_event(
"checkout.session.completed",
{"mode": "subscription", "subscription": "sub_123", "customer": "cus_456"},
metadata={"user_id": "42", "plan": "starter"},
)
ev = stripe_parse_webhook(payload)
assert ev["event_type"] == "subscription.activated"
assert ev["subscription_id"] == "sub_123"
def test_checkout_session_payment_maps_to_transaction(self):
payload = _stripe_event(
"checkout.session.completed",
{"mode": "payment", "customer": "cus_456"},
metadata={"user_id": "42"},
)
ev = stripe_parse_webhook(payload)
assert ev["event_type"] == "transaction.completed"
def test_unknown_event_returns_empty_type(self):
payload = _stripe_event("some.unknown.event", {"customer": "cus_456"})
ev = stripe_parse_webhook(payload)
assert ev["event_type"] == ""
def test_trialing_status_maps_to_on_trial(self):
payload = _stripe_event(
"customer.subscription.created",
{"id": "sub_123", "customer": "cus_456", "status": "trialing", "current_period_end": 1740000000,
"items": {"data": []}},
metadata={"user_id": "42"},
)
ev = stripe_parse_webhook(payload)
assert ev["event_type"] == "subscription.activated"
assert ev["status"] == "on_trial"