Spaces:
Runtime error
Runtime error
File size: 10,759 Bytes
a7da6ce 070c931 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 | """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,
}]},
}
@pytest.fixture()
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
@pytest.fixture()
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()
@pytest.mark.parametrize("reason,resets", [("subscription_cycle", True), ("subscription_create", True),
("subscription_update", False)])
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)
|