resolutionflow/backend/tests/test_stripe_webhook_handler.py

import json
import uuid
import pytest
from sqlalchemy import delete, select
from unittest.mock import patch
from app.models.subscription import Subscription


def _make_event(event_id, event_type, obj):
    return {
        "id": event_id,
        "type": event_type,
        "data": {"object": obj},
    }


@pytest.mark.asyncio
async def test_checkout_completed_activates_subscription(
    client, test_db, test_user, auth_headers, monkeypatch
):
    from app.core.config import settings
    monkeypatch.setattr(settings, "STRIPE_SECRET_KEY", "sk_test_dummy")
    monkeypatch.setattr(settings, "STRIPE_WEBHOOK_SECRET", "whsec_dummy")

    account_id = uuid.UUID(test_user["user_data"]["account_id"])
    # Replace seeded sub with trialing + stripe_customer_id linkage
    from app.models.account import Account
    account = (await test_db.execute(select(Account).where(Account.id == account_id))).scalar_one()
    account.stripe_customer_id = "cus_xxx"
    await test_db.execute(delete(Subscription).where(Subscription.account_id == account_id))
    test_db.add(Subscription(account_id=account_id, plan="pro", status="trialing"))
    await test_db.commit()

    event = _make_event("evt_co_1", "checkout.session.completed", {
        "id": "cs_xxx",
        "customer": "cus_xxx",
        "subscription": "sub_xxx",
    })

    with patch("stripe.Subscription.retrieve", return_value={
        "id": "sub_xxx",
        "status": "active",
        "current_period_start": 1714521600,
        "current_period_end": 1717113600,
        "items": {"data": [{
            "price": {"id": "price_test_monthly"},
            "quantity": 5,
        }]},
        "cancel_at_period_end": False,
    }), patch("stripe.Webhook.construct_event", return_value=event):
        response = await client.post(
            "/api/v1/webhooks/stripe",
            content=json.dumps(event),
            headers={"stripe-signature": "fake-sig"},
        )
    assert response.status_code == 200, response.json()

    sub = (await test_db.execute(
        select(Subscription).where(Subscription.account_id == account_id)
    )).scalar_one()
    assert sub.status == "active"
    assert sub.stripe_subscription_id == "sub_xxx"


@pytest.mark.asyncio
async def test_subscription_deleted_cancels_account(
    client, test_db, test_user, auth_headers, monkeypatch
):
    from app.core.config import settings
    monkeypatch.setattr(settings, "STRIPE_SECRET_KEY", "sk_test_dummy")
    monkeypatch.setattr(settings, "STRIPE_WEBHOOK_SECRET", "whsec_dummy")

    account_id = uuid.UUID(test_user["user_data"]["account_id"])
    await test_db.execute(delete(Subscription).where(Subscription.account_id == account_id))
    test_db.add(Subscription(
        account_id=account_id, plan="pro", status="active",
        stripe_subscription_id="sub_xxx",
    ))
    await test_db.commit()

    event = _make_event("evt_del_1", "customer.subscription.deleted", {
        "id": "sub_xxx",
        "current_period_start": 1714521600,
        "current_period_end": 1717113600,
        "items": {"data": [{"quantity": 1}]},
    })

    with patch("stripe.Webhook.construct_event", return_value=event):
        response = await client.post(
            "/api/v1/webhooks/stripe",
            content=json.dumps(event),
            headers={"stripe-signature": "fake-sig"},
        )
    assert response.status_code == 200

    sub = (await test_db.execute(
        select(Subscription).where(Subscription.account_id == account_id)
    )).scalar_one()
    assert sub.status == "canceled"


@pytest.mark.asyncio
async def test_webhook_signature_failure_returns_400(client, monkeypatch):
    from app.core.config import settings
    monkeypatch.setattr(settings, "STRIPE_SECRET_KEY", "sk_test_dummy")
    monkeypatch.setattr(settings, "STRIPE_WEBHOOK_SECRET", "whsec_dummy")

    with patch("stripe.Webhook.construct_event", side_effect=ValueError("bad sig")):
        response = await client.post(
            "/api/v1/webhooks/stripe",
            content=b"{}",
            headers={"stripe-signature": "fake-sig"},
        )
    assert response.status_code == 400


@pytest.mark.asyncio
async def test_webhook_idempotency(
    client, test_db, test_user, auth_headers, monkeypatch
):
    from app.core.config import settings
    monkeypatch.setattr(settings, "STRIPE_SECRET_KEY", "sk_test_dummy")
    monkeypatch.setattr(settings, "STRIPE_WEBHOOK_SECRET", "whsec_dummy")

    account_id = uuid.UUID(test_user["user_data"]["account_id"])
    await test_db.execute(delete(Subscription).where(Subscription.account_id == account_id))
    test_db.add(Subscription(account_id=account_id, plan="pro", status="trialing"))
    await test_db.commit()

    event = _make_event("evt_dup_1", "customer.subscription.updated", {
        "id": "sub_yyy",
        "status": "active",
        "current_period_start": 1714521600,
        "current_period_end": 1717113600,
        "items": {"data": [{"quantity": 1}]},
        "cancel_at_period_end": False,
    })
    with patch("stripe.Webhook.construct_event", return_value=event):
        r1 = await client.post("/api/v1/webhooks/stripe", content=json.dumps(event), headers={"stripe-signature": "x"})
        r2 = await client.post("/api/v1/webhooks/stripe", content=json.dumps(event), headers={"stripe-signature": "x"})
    assert r1.status_code == 200
    assert r2.status_code == 200
    assert r1.json()["applied"] is True
    assert r2.json()["applied"] is False


# ----------------------------------------------------------------------------
# Atomic-idempotency regression tests
# ----------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_apply_event_handler_failure_does_not_persist_idempotency_mark(
    test_db, test_user,
):
    """If the handler raises, the StripeEvent row must NOT be persisted —
    otherwise Stripe's retry will be silently dropped as a duplicate and the
    subscription state will desync from Stripe."""
    from app.services.billing import BillingService
    from app.models.stripe_event import StripeEvent

    event_id = "evt_handler_fail_1"
    payload = {"data": {"object": {
        "id": "sub_doesnotmatter",
        "status": "active",
        "current_period_start": 1714521600,
        "current_period_end": 1717113600,
        "items": {"data": [{"quantity": 1}]},
        "cancel_at_period_end": False,
    }}}

    boom = RuntimeError("simulated handler failure")
    with patch(
        "app.services.billing._handle_subscription_updated",
        side_effect=boom,
    ):
        with pytest.raises(RuntimeError, match="simulated handler failure"):
            await BillingService.apply_subscription_event(
                test_db,
                event_id=event_id,
                event_type="customer.subscription.updated",
                payload=payload,
            )

    # The StripeEvent row must not exist — handler raised, the entire
    # transaction (idempotency mark + partial mutations) was rolled back.
    row = (await test_db.execute(
        select(StripeEvent).where(StripeEvent.id == event_id)
    )).scalar_one_or_none()
    assert row is None, (
        "StripeEvent row was persisted despite handler failure — "
        "Stripe retry will be silently dropped"
    )


@pytest.mark.asyncio
async def test_apply_event_retry_after_failure_succeeds(
    test_db, test_user,
):
    """A failed first attempt followed by a successful retry must apply state.
    This is the core Stripe webhook retry contract."""
    from app.services.billing import BillingService
    from app.models.stripe_event import StripeEvent

    account_id = uuid.UUID(test_user["user_data"]["account_id"])
    await test_db.execute(delete(Subscription).where(Subscription.account_id == account_id))
    test_db.add(Subscription(
        account_id=account_id, plan="pro", status="trialing",
        stripe_subscription_id="sub_retry",
    ))
    await test_db.commit()

    event_id = "evt_retry_1"
    payload = {"data": {"object": {
        "id": "sub_retry",
        "status": "active",
        "current_period_start": 1714521600,
        "current_period_end": 1717113600,
        "items": {"data": [{"quantity": 3}]},
        "cancel_at_period_end": False,
    }}}

    # First attempt — handler raises mid-flight.
    with patch(
        "app.services.billing._handle_subscription_updated",
        side_effect=RuntimeError("transient blip"),
    ):
        with pytest.raises(RuntimeError):
            await BillingService.apply_subscription_event(
                test_db,
                event_id=event_id,
                event_type="customer.subscription.updated",
                payload=payload,
            )

    # No idempotency mark, sub still trialing.
    row = (await test_db.execute(
        select(StripeEvent).where(StripeEvent.id == event_id)
    )).scalar_one_or_none()
    assert row is None
    sub = (await test_db.execute(
        select(Subscription).where(Subscription.account_id == account_id)
    )).scalar_one()
    assert sub.status == "trialing"

    # Second attempt — same event_id, handler succeeds.
    applied = await BillingService.apply_subscription_event(
        test_db,
        event_id=event_id,
        event_type="customer.subscription.updated",
        payload=payload,
    )
    assert applied is True

    # Idempotency mark now persisted, sub state reconciled.
    row = (await test_db.execute(
        select(StripeEvent).where(StripeEvent.id == event_id)
    )).scalar_one()
    assert row.id == event_id
    await test_db.refresh(sub)
    assert sub.status == "active"
    assert sub.seat_limit == 3


@pytest.mark.asyncio
async def test_apply_event_duplicate_event_id_skips(
    test_db, test_user,
):
    """Two successful invocations with the same event_id must not double-apply.
    Second call returns False; mutations are not repeated."""
    from app.services.billing import BillingService

    account_id = uuid.UUID(test_user["user_data"]["account_id"])
    await test_db.execute(delete(Subscription).where(Subscription.account_id == account_id))
    test_db.add(Subscription(
        account_id=account_id, plan="pro", status="trialing",
        stripe_subscription_id="sub_dup",
    ))
    await test_db.commit()

    event_id = "evt_dedupe_1"
    payload = {"data": {"object": {
        "id": "sub_dup",
        "status": "active",
        "current_period_start": 1714521600,
        "current_period_end": 1717113600,
        "items": {"data": [{"quantity": 7}]},
        "cancel_at_period_end": False,
    }}}

    applied1 = await BillingService.apply_subscription_event(
        test_db,
        event_id=event_id,
        event_type="customer.subscription.updated",
        payload=payload,
    )
    assert applied1 is True

    sub = (await test_db.execute(
        select(Subscription).where(Subscription.account_id == account_id)
    )).scalar_one()
    assert sub.status == "active"
    assert sub.seat_limit == 7

    # Mutate locally so we can prove the second call doesn't re-run the handler.
    sub.seat_limit = 99
    await test_db.commit()

    applied2 = await BillingService.apply_subscription_event(
        test_db,
        event_id=event_id,
        event_type="customer.subscription.updated",
        payload=payload,
    )
    assert applied2 is False

    await test_db.refresh(sub)
    # Handler did NOT run again — our local mutation is preserved.
    assert sub.seat_limit == 99