Testing Stripe Webhooks End to End

The Feature

A developer can run automated tests that verify every Stripe webhook handler produces the correct database state, handles duplicate deliveries gracefully, and processes events in the correct order.

The Decision

Webhook handlers are tested with real HTTP requests to the FastAPI test client, using Stripe’s event fixture data. The tests do not call the actual Stripe API. They construct events with the same structure Stripe sends and verify the handler’s database effects. The Stripe CLI provides manual testing for the full end-to-end flow.

The Implementation

Test Fixtures

# backend/tests/conftest.py
import uuid
from datetime import datetime

import pytest
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker

from app.main import app
from app.database import get_db
from app.models.base import Base
from app.models.organizer import OrganizerProfile, SubscriptionTier

TEST_DATABASE_URL = "sqlite+aiosqlite:///./test.db"

engine = create_async_engine(TEST_DATABASE_URL)
TestSession = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)


@pytest_asyncio.fixture
async def db_session():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    async with TestSession() as session:
        yield session

    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)


@pytest_asyncio.fixture
async def client(db_session):
    async def override_get_db():
        yield db_session

    app.dependency_overrides[get_db] = override_get_db
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test",
    ) as client:
        yield client
    app.dependency_overrides.clear()


@pytest.fixture
def organizer(db_session) -> OrganizerProfile:
    org = OrganizerProfile(
        id=uuid.uuid4(),
        user_id=uuid.uuid4(),
        display_name="Test Organizer",
        email="[email protected]",
        stripe_customer_id="cus_test123",
        subscription_tier=SubscriptionTier.FREE,
    )
    return org

Webhook Handler Tests

# backend/tests/test_webhooks.py
import json
import hmac
import hashlib
import time

import pytest
import pytest_asyncio
from sqlalchemy import select

from app.models.organizer import OrganizerProfile, SubscriptionTier
from app.config import settings


def generate_stripe_signature(payload: bytes, secret: str) -> str:
    """Generate a valid Stripe webhook signature for testing."""
    timestamp = str(int(time.time()))
    signed_payload = f"{timestamp}.{payload.decode()}"
    signature = hmac.new(
        secret.encode(),
        signed_payload.encode(),
        hashlib.sha256,
    ).hexdigest()
    return f"t={timestamp},v1={signature}"


def make_stripe_event(event_type: str, data: dict) -> dict:
    return {
        "id": f"evt_test_{event_type}",
        "type": event_type,
        "data": {"object": data},
        "created": int(time.time()),
    }


@pytest.mark.asyncio
async def test_checkout_completed_upgrades_organizer(client, db_session, organizer):
    db_session.add(organizer)
    await db_session.commit()

    event = make_stripe_event("checkout.session.completed", {
        "customer": "cus_test123",
        "subscription": "sub_test456",
    })

    payload = json.dumps(event).encode()
    signature = generate_stripe_signature(payload, settings.stripe_webhook_secret)

    response = await client.post(
        "/api/webhooks/stripe",
        content=payload,
        headers={
            "stripe-signature": signature,
            "content-type": "application/json",
        },
    )

    assert response.status_code == 200

    result = await db_session.execute(
        select(OrganizerProfile).where(
            OrganizerProfile.stripe_customer_id == "cus_test123"
        )
    )
    updated = result.scalar_one()
    assert updated.subscription_tier == SubscriptionTier.PAID
    assert updated.stripe_subscription_id == "sub_test456"


@pytest.mark.asyncio
async def test_subscription_deleted_downgrades(client, db_session, organizer):
    organizer.subscription_tier = SubscriptionTier.PAID
    organizer.stripe_subscription_id = "sub_test456"
    db_session.add(organizer)
    await db_session.commit()

    event = make_stripe_event("customer.subscription.deleted", {
        "customer": "cus_test123",
        "id": "sub_test456",
    })

    payload = json.dumps(event).encode()
    signature = generate_stripe_signature(payload, settings.stripe_webhook_secret)

    response = await client.post(
        "/api/webhooks/stripe",
        content=payload,
        headers={"stripe-signature": signature, "content-type": "application/json"},
    )

    assert response.status_code == 200

    result = await db_session.execute(
        select(OrganizerProfile).where(
            OrganizerProfile.stripe_customer_id == "cus_test123"
        )
    )
    updated = result.scalar_one()
    assert updated.subscription_tier == SubscriptionTier.FREE
    assert updated.stripe_subscription_id is None


@pytest.mark.asyncio
async def test_duplicate_webhook_is_idempotent(client, db_session, organizer):
    db_session.add(organizer)
    await db_session.commit()

    event = make_stripe_event("checkout.session.completed", {
        "customer": "cus_test123",
        "subscription": "sub_test456",
    })

    payload = json.dumps(event).encode()
    signature = generate_stripe_signature(payload, settings.stripe_webhook_secret)
    headers = {"stripe-signature": signature, "content-type": "application/json"}

    # Process the same event twice
    await client.post("/api/webhooks/stripe", content=payload, headers=headers)
    await client.post("/api/webhooks/stripe", content=payload, headers=headers)

    result = await db_session.execute(
        select(OrganizerProfile).where(
            OrganizerProfile.stripe_customer_id == "cus_test123"
        )
    )
    # Should still be exactly one organizer, not duplicated
    organizers = result.scalars().all()
    assert len(organizers) == 1
    assert organizers[0].subscription_tier == SubscriptionTier.PAID


@pytest.mark.asyncio
async def test_invalid_signature_rejected(client):
    payload = json.dumps({"type": "test"}).encode()

    response = await client.post(
        "/api/webhooks/stripe",
        content=payload,
        headers={
            "stripe-signature": "t=123,v1=invalid",
            "content-type": "application/json",
        },
    )

    assert response.status_code == 400

Manual Testing with Stripe CLI

# Trigger a complete subscription lifecycle
docker compose exec stripe-cli stripe trigger checkout.session.completed
docker compose exec stripe-cli stripe trigger customer.subscription.created
docker compose exec stripe-cli stripe trigger invoice.payment_succeeded

# Verify the organizer was upgraded
curl http://localhost:8000/api/billing/status \
  -H "Authorization: Bearer <jwt_token>"

# Trigger a cancellation
docker compose exec stripe-cli stripe trigger customer.subscription.deleted

# Trigger a payment failure
docker compose exec stripe-cli stripe trigger invoice.payment_failed

The Trap

# TRAP: Testing webhooks without signature verification
@pytest.mark.asyncio
async def test_webhook(client):
    # Sending raw JSON without a signature
    response = await client.post(
        "/api/webhooks/stripe",
        json={"type": "checkout.session.completed", "data": {...}},
    )
    # This passes in tests but the production handler rejects unsigned requests
    # The test proves nothing about the actual webhook flow

# SAFE: Generate a valid signature in tests (shown above)
# The test exercises the same code path as production

Tests that skip signature verification test a different code path than production. The test passes, the production handler rejects the request, and the developer spends hours debugging why webhooks “work in tests but not in production.”

The Cost

Test	Time to Run
4 webhook tests	~2 seconds
Manual Stripe CLI test	~30 seconds per event
Full lifecycle test	~3 minutes

Automated tests run in CI on every push. Manual Stripe CLI tests run during development when changing webhook handlers. The total testing time is negligible compared to the cost of a billing bug that overcharges or undercharges a customer.