Spaces:
Runtime error
Runtime error
| """Stripe webhooks and checkout with a fake Stripe: signatures, retries, out of order events, 3DS, grace.""" | |
| import hashlib | |
| import hmac | |
| import json | |
| import time | |
| import uuid | |
| import pytest | |
| from sqlalchemy import select | |
| from app.config import settings | |
| from app.models import Subscription, Workspace | |
| from app.services import stripe_billing | |
| from app.services.email import ConsoleEmailProvider | |
| from app.services.plans import effective_plan | |
| from tests.conftest import last_code | |
| SECRET = "whsec_test_secret" | |
| class FakeStripe: | |
| """Just enough of StripeClient.v1 for the billing flows.""" | |
| def __init__(self): | |
| self.subs: dict[str, dict] = {} | |
| self.calls: list[tuple] = [] | |
| self.pending = False # next plan change needs 3DS | |
| outer = self | |
| class Subs: | |
| def retrieve(self, sid, params=None): | |
| return outer.subs[sid] | |
| def update(self, sid, params): | |
| outer.calls.append(("update", sid, params)) | |
| sub = outer.subs[sid] | |
| invoice = {"status": "open" if outer.pending else "paid", | |
| "hosted_invoice_url": "https://invoice.stripe.test/i_1"} | |
| if outer.pending: | |
| return {**sub, "pending_update": {"subscription_items": params["items"]}, "latest_invoice": invoice} | |
| sub["items"]["data"][0]["price"] = {"id": params["items"][0]["price"], | |
| "lookup_key": "notestack_studio_monthly"} | |
| return {**sub, "pending_update": None, "latest_invoice": invoice} | |
| class Prices: | |
| def list(self, params): | |
| return type("L", (), {"data": [type("P", (), {"id": f"price_{params['lookup_keys'][0]}"})()]})() | |
| class Customers: | |
| def create(self, params, options=None): | |
| outer.calls.append(("customer", params)) | |
| return type("C", (), {"id": "cus_1"})() | |
| class Sessions: | |
| def create(self, params): | |
| outer.calls.append(("checkout", params)) | |
| return type("S", (), {"url": "https://checkout.stripe.test/s_1"})() | |
| def retrieve(self, sid, params=None): | |
| return outer.session | |
| class V1: | |
| subscriptions = Subs() | |
| prices = Prices() | |
| customers = Customers() | |
| checkout = type("Checkout", (), {"sessions": Sessions()})() | |
| self.v1 = V1() | |
| self.session: dict = {} | |
| def add_sub(self, sid, workspace_id, plan="writer", status="active", period_end=None): | |
| self.subs[sid] = { | |
| "id": sid, "customer": "cus_1", "status": status, | |
| "metadata": {"workspace_id": str(workspace_id), "plan": plan}, | |
| "items": {"data": [{ | |
| "id": "si_1", | |
| "price": {"id": f"price_{plan}", "lookup_key": f"notestack_{plan}_monthly"}, | |
| "current_period_end": period_end or int(time.time()) + 30 * 86400, | |
| }]}, | |
| } | |
| def stripe_on(monkeypatch): | |
| fake = FakeStripe() | |
| monkeypatch.setattr(settings, "billing_enabled", True) | |
| monkeypatch.setattr(settings, "stripe_secret_key", "sk_test_x") | |
| monkeypatch.setattr(settings, "stripe_webhook_secret", SECRET) | |
| monkeypatch.setattr(stripe_billing, "client", lambda: fake) | |
| return fake | |
| def owner(client, db_session): | |
| client.post("/api/auth/email/register/start", json={"email": "grace@example.com", "password": "stardust-42", | |
| "name": "Grace Hopper"}) | |
| r = client.post("/api/auth/email/register/verify", json={"email": "grace@example.com", "code": last_code()}) | |
| ws = db_session.scalar(select(Workspace)) | |
| return {"Authorization": f"Bearer {r.json()['access_token']}"}, ws | |
| def send(client, event: dict, secret: str = SECRET): | |
| payload = json.dumps(event) | |
| t = int(time.time()) | |
| sig = hmac.new(secret.encode(), f"{t}.{payload}".encode(), hashlib.sha256).hexdigest() | |
| return client.post("/api/billing/webhook", content=payload, | |
| headers={"stripe-signature": f"t={t},v1={sig}", "content-type": "application/json"}) | |
| def event(kind: str, obj: dict) -> dict: | |
| return {"id": f"evt_{uuid.uuid4().hex}", "object": "event", "type": kind, "data": {"object": obj}} | |
| def plan_of(db, ws) -> str: | |
| db.expire_all() | |
| return effective_plan(db, ws).id | |
| def test_webhook_rejects_bad_signature(client, stripe_on, owner): | |
| r = send(client, event("customer.subscription.updated", {"id": "sub_x"}), secret="whsec_wrong") | |
| assert r.status_code == 400 | |
| def test_checkout_completed_then_retry_and_out_of_order(client, stripe_on, owner, db_session): | |
| _, ws = owner | |
| stripe_on.add_sub("sub_1", ws.id, plan="writer") | |
| done = event("checkout.session.completed", {"mode": "subscription", "subscription": "sub_1", | |
| "client_reference_id": str(ws.id)}) | |
| assert send(client, done).json()["result"] == "synced" | |
| assert plan_of(db_session, ws) == "writer" | |
| # Stripe retries the same delivery: still fine, nothing doubles. | |
| assert send(client, done).status_code == 200 | |
| assert db_session.scalar(select(Subscription).where(Subscription.workspace_id == ws.id)).plan == "writer" | |
| # An older "incomplete" payload arrives late: we re-read Stripe, which says active. | |
| stale = {**stripe_on.subs["sub_1"], "status": "incomplete"} | |
| send(client, event("customer.subscription.updated", stale)) | |
| assert plan_of(db_session, ws) == "writer" | |
| def test_past_due_keeps_plan_and_emails_once_then_cancel_drops_to_free(client, stripe_on, owner, db_session): | |
| _, ws = owner | |
| stripe_on.add_sub("sub_2", ws.id, plan="studio") | |
| send(client, event("customer.subscription.created", stripe_on.subs["sub_2"])) | |
| stripe_on.subs["sub_2"]["status"] = "past_due" | |
| ConsoleEmailProvider.sent.clear() | |
| failed = event("invoice.payment_failed", { | |
| "subscription": "sub_2", "customer": "cus_1", "customer_email": "grace@example.com", | |
| "billing_reason": "subscription_cycle", "hosted_invoice_url": "https://invoice.stripe.test/i_2"}) | |
| send(client, failed) | |
| send(client, failed) # retried delivery | |
| assert plan_of(db_session, ws) == "studio" # grace while Stripe retries | |
| assert len([m for m in ConsoleEmailProvider.sent if "payment failed" in m.subject.lower()]) == 1 | |
| stripe_on.subs["sub_2"]["status"] = "canceled" | |
| send(client, event("customer.subscription.deleted", stripe_on.subs["sub_2"])) | |
| assert plan_of(db_session, ws) == "free" | |
| def test_renewal_needing_3ds_emails_confirm_link(client, stripe_on, owner, db_session): | |
| _, ws = owner | |
| stripe_on.add_sub("sub_3", ws.id) | |
| ConsoleEmailProvider.sent.clear() | |
| send(client, event("invoice.payment_action_required", { | |
| "subscription": "sub_3", "customer": "cus_1", "customer_email": "grace@example.com", | |
| "billing_reason": "subscription_cycle", "hosted_invoice_url": "https://invoice.stripe.test/i_3"})) | |
| mail = [m for m in ConsoleEmailProvider.sent if "confirm" in m.subject.lower()] | |
| assert len(mail) == 1 and "https://invoice.stripe.test/i_3" in mail[0].text | |
| assert plan_of(db_session, ws) == "writer" | |
| def test_checkout_creates_one_customer_and_confirm_applies_plan(client, stripe_on, owner, db_session): | |
| headers, ws = owner | |
| r = client.post("/api/billing/checkout", json={"plan": "writer", "cycle": "annual"}, headers=headers) | |
| assert r.json()["url"] == "https://checkout.stripe.test/s_1" | |
| kinds = [c[0] for c in stripe_on.calls] | |
| assert kinds == ["customer", "checkout"] | |
| session = stripe_on.calls[1][1] | |
| assert session["customer"] == "cus_1" and session["client_reference_id"] == str(ws.id) | |
| assert "{CHECKOUT_SESSION_ID}" in session["success_url"] | |
| assert session["line_items"][0]["price"] == "price_notestack_writer_annual" | |
| # Back on the success page before any webhook: confirm applies the plan. | |
| stripe_on.add_sub("sub_4", ws.id, plan="writer") | |
| stripe_on.session = {"client_reference_id": str(ws.id), "status": "complete", | |
| "subscription": stripe_on.subs["sub_4"]} | |
| r = client.post("/api/billing/confirm", json={"session_id": "cs_1"}, headers=headers) | |
| assert r.json()["plan"]["id"] == "writer" | |
| # A second checkout while subscribed changes the plan in place instead of a second subscription. | |
| r = client.post("/api/billing/checkout", json={"plan": "studio", "cycle": "monthly"}, headers=headers) | |
| assert r.json()["url"].endswith("/app?checkout=success") | |
| assert [c[0] for c in stripe_on.calls].count("customer") == 1 | |
| assert plan_of(db_session, ws) == "studio" | |
| def test_upgrade_needing_3ds_sends_user_to_invoice(client, stripe_on, owner, db_session): | |
| headers, ws = owner | |
| stripe_on.add_sub("sub_5", ws.id, plan="writer") | |
| send(client, event("customer.subscription.created", stripe_on.subs["sub_5"])) | |
| stripe_on.pending = True | |
| r = client.post("/api/billing/checkout", json={"plan": "studio", "cycle": "monthly"}, headers=headers) | |
| assert r.json()["url"] == "https://invoice.stripe.test/i_1" | |
| assert plan_of(db_session, ws) == "writer" # switches only once the payment succeeds | |
| def videos_used(db, ws) -> int: | |
| db.expire_all() | |
| return db.scalar(select(Subscription).where(Subscription.workspace_id == ws.id)).videos_used | |
| def set_videos_used(db, ws, n: int) -> None: | |
| db.scalar(select(Subscription).where(Subscription.workspace_id == ws.id)).videos_used = n | |
| db.commit() | |
| def test_invoice_paid_resets_videos(client, stripe_on, owner, db_session, reason, resets): | |
| _, ws = owner | |
| stripe_on.add_sub("sub_v", ws.id, plan="writer") | |
| set_videos_used(db_session, ws, 7) | |
| paid = event("invoice.paid", {"subscription": "sub_v", "customer": "cus_1", "billing_reason": reason}) | |
| send(client, paid) | |
| assert videos_used(db_session, ws) == (0 if resets else 7) | |
| # A retried delivery of the same event must not wipe videos made since. | |
| set_videos_used(db_session, ws, 3) | |
| send(client, paid) | |
| assert videos_used(db_session, ws) == 3 | |
| def test_subscription_sync_sets_video_limit(client, stripe_on, owner, db_session): | |
| _, ws = owner | |
| stripe_on.add_sub("sub_l", ws.id, plan="studio") | |
| send(client, event("customer.subscription.created", stripe_on.subs["sub_l"])) | |
| db_session.expire_all() | |
| sub = db_session.scalar(select(Subscription).where(Subscription.workspace_id == ws.id)) | |
| assert (sub.video_plan, sub.video_limit) == ("studio", 20) | |