deemanone/padelnomics
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
4dbded74cac42330882c3a3bdfc859a30bede655
padelnomics/web/tests/test_billing_webhooks.py
Deeman 7da6a4737d fix(billing): extract current_period_end from Stripe subscription items 
Stripe API 2026-02+ moved current_period_end from subscription to
subscription items. Add _get_period_end() helper that falls back to
items[0].current_period_end when the subscription-level field is None.

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

418 lines
17 KiB
Python
Raw Blame History

"""
Integration tests for Paddle webhook handling + Stripe parse_webhook unit tests.
Covers signature verification, event parsing, subscription lifecycle transitions, and Hypothesis fuzzing.
"""
import json
import pytest
from conftest import make_webhook_payload, sign_payload
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"
# ════════════════════════════════════════════════════════════
# Signature Verification
# ════════════════════════════════════════════════════════════
class TestWebhookSignature:
async def test_missing_signature_rejected(self, client, db):
payload = make_webhook_payload("subscription.activated")
payload_bytes = json.dumps(payload).encode()
response = await client.post(
WEBHOOK_PATH,
data=payload_bytes,
headers={"Content-Type": "application/json"},
)
assert response.status_code in (400, 401)
async def test_invalid_signature_rejected(self, client, db):
payload = make_webhook_payload("subscription.activated")
payload_bytes = json.dumps(payload).encode()
response = await client.post(
WEBHOOK_PATH,
data=payload_bytes,
headers={SIG_HEADER: "invalid_signature", "Content-Type": "application/json"},
)
assert response.status_code in (400, 401)
async def test_valid_signature_accepted(self, client, db, test_user):
payload = make_webhook_payload("subscription.activated", user_id=str(test_user["id"]))
payload_bytes = json.dumps(payload).encode()
sig = sign_payload(payload_bytes)
response = await client.post(
WEBHOOK_PATH,
data=payload_bytes,
headers={SIG_HEADER: sig, "Content-Type": "application/json"},
)
assert response.status_code in (200, 204)
async def test_modified_payload_rejected(self, client, db, test_user):
payload = make_webhook_payload("subscription.activated", user_id=str(test_user["id"]))
payload_bytes = json.dumps(payload).encode()
sig = sign_payload(payload_bytes)
tampered = payload_bytes + b"extra"
response = await client.post(
WEBHOOK_PATH,
data=tampered,
headers={SIG_HEADER: sig, "Content-Type": "application/json"},
)
assert response.status_code in (400, 401)
async def test_empty_payload_rejected(self, client, db):
sig = sign_payload(b"")
response = await client.post(
WEBHOOK_PATH,
data=b"",
headers={SIG_HEADER: sig, "Content-Type": "application/json"},
)
assert response.status_code == 400
async def test_null_custom_data_does_not_crash(self, client, db):
"""Paddle sends null custom_data on lifecycle events like subscription.updated."""
payload = {
"event_type": "subscription.updated",
"data": {
"id": "sub_test456",
"status": "active",
"customer_id": "ctm_test123",
"custom_data": None,
"current_billing_period": {
"starts_at": "2025-02-01T00:00:00.000000Z",
"ends_at": "2025-03-01T00:00:00.000000Z",
},
},
}
payload_bytes = json.dumps(payload).encode()
sig = sign_payload(payload_bytes)
response = await client.post(
WEBHOOK_PATH,
data=payload_bytes,
headers={SIG_HEADER: sig, "Content-Type": "application/json"},
)
assert response.status_code == 200
async def test_null_custom_data_activated_does_not_crash(self, client, db):
"""subscription.activated with null custom_data must not FK-violate on user_id=0."""
payload = {
"event_type": "subscription.activated",
"data": {
"id": "sub_test456",
"status": "active",
"customer_id": "ctm_test123",
"custom_data": None,
"current_billing_period": {
"starts_at": "2025-02-01T00:00:00.000000Z",
"ends_at": "2025-03-01T00:00:00.000000Z",
},
},
}
payload_bytes = json.dumps(payload).encode()
sig = sign_payload(payload_bytes)
response = await client.post(
WEBHOOK_PATH,
data=payload_bytes,
headers={SIG_HEADER: sig, "Content-Type": "application/json"},
)
assert response.status_code == 200
async def test_null_data_does_not_crash(self, client, db):
"""Guard against data being null in the event payload."""
payload = {
"event_type": "subscription.updated",
"data": None,
}
payload_bytes = json.dumps(payload).encode()
sig = sign_payload(payload_bytes)
response = await client.post(
WEBHOOK_PATH,
data=payload_bytes,
headers={SIG_HEADER: sig, "Content-Type": "application/json"},
)
assert response.status_code == 200
# ════════════════════════════════════════════════════════════
# Subscription Lifecycle Events
# ════════════════════════════════════════════════════════════
class TestWebhookSubscriptionActivated:
async def test_creates_subscription(self, client, db, test_user):
payload = make_webhook_payload(
"subscription.activated",
user_id=str(test_user["id"]),
plan="starter",
)
payload_bytes = json.dumps(payload).encode()
sig = sign_payload(payload_bytes)
response = await client.post(
WEBHOOK_PATH,
data=payload_bytes,
headers={SIG_HEADER: sig, "Content-Type": "application/json"},
)
assert response.status_code in (200, 204)
sub = await get_subscription(test_user["id"])
assert sub is not None
assert sub["plan"] == "starter"
assert sub["status"] == "active"
class TestWebhookSubscriptionUpdated:
async def test_updates_subscription_status(self, client, db, test_user, create_subscription):
await create_subscription(test_user["id"], status="active", provider_subscription_id="sub_test456")
payload = make_webhook_payload(
"subscription.updated",
subscription_id="sub_test456",
status="paused",
)
payload_bytes = json.dumps(payload).encode()
sig = sign_payload(payload_bytes)
response = await client.post(
WEBHOOK_PATH,
data=payload_bytes,
headers={SIG_HEADER: sig, "Content-Type": "application/json"},
)
assert response.status_code in (200, 204)
sub = await get_subscription(test_user["id"])
assert sub["status"] == "paused"
class TestWebhookSubscriptionCanceled:
async def test_marks_subscription_cancelled(self, client, db, test_user, create_subscription):
await create_subscription(test_user["id"], status="active", provider_subscription_id="sub_test456")
payload = make_webhook_payload(
"subscription.canceled",
subscription_id="sub_test456",
)
payload_bytes = json.dumps(payload).encode()
sig = sign_payload(payload_bytes)
response = await client.post(
WEBHOOK_PATH,
data=payload_bytes,
headers={SIG_HEADER: sig, "Content-Type": "application/json"},
)
assert response.status_code in (200, 204)
sub = await get_subscription(test_user["id"])
assert sub["status"] == "cancelled"
class TestWebhookSubscriptionPastDue:
async def test_marks_subscription_past_due(self, client, db, test_user, create_subscription):
await create_subscription(test_user["id"], status="active", provider_subscription_id="sub_test456")
payload = make_webhook_payload(
"subscription.past_due",
subscription_id="sub_test456",
)
payload_bytes = json.dumps(payload).encode()
sig = sign_payload(payload_bytes)
response = await client.post(
WEBHOOK_PATH,
data=payload_bytes,
headers={SIG_HEADER: sig, "Content-Type": "application/json"},
)
assert response.status_code in (200, 204)
sub = await get_subscription(test_user["id"])
assert sub["status"] == "past_due"
# ════════════════════════════════════════════════════════════
# Parameterized: event -> status transitions
# ════════════════════════════════════════════════════════════
@pytest.mark.parametrize("event_type,expected_status", [
("subscription.activated", "active"),
("subscription.updated", "active"),
("subscription.canceled", "cancelled"),
("subscription.past_due", "past_due"),
])
async def test_event_status_transitions(client, db, test_user, create_subscription, event_type, expected_status):
if event_type != "subscription.activated":
await create_subscription(test_user["id"], provider_subscription_id="sub_test456")
payload = make_webhook_payload(event_type, user_id=str(test_user["id"]))
payload_bytes = json.dumps(payload).encode()
sig = sign_payload(payload_bytes)
response = await client.post(
WEBHOOK_PATH,
data=payload_bytes,
headers={SIG_HEADER: sig, "Content-Type": "application/json"},
)
assert response.status_code in (200, 204)
sub = await get_subscription(test_user["id"])
assert sub["status"] == expected_status
# ════════════════════════════════════════════════════════════
# Hypothesis: fuzz webhook payloads
# ════════════════════════════════════════════════════════════
fuzz_event_type = st.sampled_from([
"subscription.activated",
"subscription.updated",
"subscription.canceled",
"subscription.past_due",
])
fuzz_status = st.sampled_from(["active", "paused", "past_due", "canceled"])
@st.composite
def fuzz_payload(draw):
event_type = draw(fuzz_event_type)
return make_webhook_payload(
event_type=event_type,
subscription_id=f"sub_{draw(st.integers(min_value=100, max_value=999999))}",
user_id=str(draw(st.integers(min_value=1, max_value=999999))),
status=draw(fuzz_status),
)
class TestWebhookHypothesis:
@given(payload_dict=fuzz_payload())
@h_settings(max_examples=50, deadline=5000, suppress_health_check=[HealthCheck.function_scoped_fixture])
async def test_webhook_never_500s(self, client, db, test_user, payload_dict):
# Pin user_id to the test user so subscription_created/activated events don't hit FK violations
payload_dict["data"]["custom_data"]["user_id"] = str(test_user["id"])
payload_bytes = json.dumps(payload_dict).encode()
sig = sign_payload(payload_bytes)
response = await client.post(
WEBHOOK_PATH,
data=payload_bytes,
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_period_end_from_items_fallback(self):
"""Stripe API 2026-02+ puts current_period_end on items, not subscription."""
payload = _stripe_event(
"customer.subscription.created",
{"id": "sub_123", "customer": "cus_456", "status": "active",
"items": {"data": [{"price": {"id": "price_abc"}, "current_period_end": 1740000000}]}},
metadata={"user_id": "42"},
)
ev = stripe_parse_webhook(payload)
assert ev["current_period_end"] is not None
assert "2025-02-19" in ev["current_period_end"]
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"
Reference in New Issue View Git Blame Copy Permalink