tests/test_billing.py (12003 bytes)
1 """Tiers, credits, Stripe fulfilment. The rules under test: 2 3 free a job in the visitor's browser: no price, no limit, each output 4 cloud a job on this server: one credit (or a recurring plan, if one is sold) 5 """ 6 7 from __future__ import annotations 8 9 import json 10 import time 11 from dataclasses import asdict 12 13 import pytest 14 import stripe 15 16 from backend import pricing, runner 17 from backend.db import JobStatus 18 19 from .conftest import ( 20 FakeStripeObject, account_id, checkout_event, checkout_object, 21 get_account_row, get_job_row, make_job, post_webhook, 22 ) 23 24 25 def start(client, job_id): 26 """Start a job on the server: a cloud job.""" 27 return client.post(f"/api/jobs/{job_id}/start", json={"language": "en"}) 28 29 30 @pytest.fixture 31 def monthly_plan(monkeypatch): 32 """The default catalogue sells no recurring plan. An operator can add one.""" 33 plans = [asdict(p) for p in pricing.DEFAULT_PLANS] 34 plans.append({"id": "monthly", "name": "Monthly", "credits": None, "price_cents": 1500, "recurring": True}) 35 monkeypatch.setenv("SUBPLZ_WEB_PLANS_JSON", json.dumps(plans)) 36 37 38 def buy(client, plan="pack5", session_id=None, **kw): 39 session_id = session_id or f"cs_{time.time_ns()}" 40 r = post_webhook(client, checkout_event(session_id, account_id(client), plan, **kw)) 41 assert r.status_code == 200, r.text 42 return session_id 43 44 45 # --- the cloud tier ---------------------------------------------------------- 46 47 def test_new_visitor_is_anonymous_and_has_no_credits(client): 48 a = client.get("/api/account").json() 49 assert a["signed_in"] is False and a["email"] is None 50 assert a["credits"] == 0 and a["subscribed"] is False 51 assert a["cloud_allowed"] is False 52 assert "free in your browser" in a["free_tier_summary"] 53 54 55 def test_nothing_is_for_sale_until_the_cloud_is_connected(client, monkeypatch): 56 """Billing can be on while fast conversion is not yet on offer. The page 57 reads this flag, and shows no way to buy credits that would buy nothing.""" 58 from backend.settings import settings 59 assert client.get("/api/account").json()["cloud_available"] is False 60 monkeypatch.setattr(settings, "cloud_enabled", True) 61 assert client.get("/api/account").json()["cloud_available"] is True 62 63 64 def test_a_server_job_needs_a_credit(client): 65 r = start(client, make_job(client)) 66 assert r.status_code == 402 67 assert "one credit" in r.json()["detail"] and "free" in r.json()["detail"] 68 69 70 def test_each_output_of_a_job_can_be_downloaded(client): 71 job_id = make_job(client, JobStatus.succeeded, with_files=True) 72 arts = client.get(f"/api/jobs/{job_id}").json()["artifacts"] 73 assert {a["kind"] for a in arts} >= {"srt", "video", "video_embedded"} 74 assert all("locked" not in a for a in arts) 75 for kind in ("srt", "video_embedded", "video"): 76 assert client.get(f"/api/jobs/{job_id}/files/{kind}").status_code == 200 77 78 79 # --- credits ----------------------------------------------------------------- 80 81 def test_purchase_credits_the_account_and_signs_it_in(client): 82 buy(client, "pack5", email="Reader@Example.com") 83 a = client.get("/api/account").json() 84 assert a["credits"] == 5 85 assert a["signed_in"] is True and a["email"] == "reader@example.com" 86 assert a["cloud_allowed"] is True 87 88 89 def test_webhook_redelivery_does_not_credit_twice(client): 90 sid = buy(client, "pack5", email="twice@example.com") 91 buy(client, "pack5", session_id=sid, email="twice@example.com") 92 assert client.get("/api/account").json()["credits"] == 5 93 94 95 def test_unpaid_checkout_credits_nothing(client): 96 buy(client, "pack5", email="pending@example.com", payment_status="unpaid") 97 assert client.get("/api/account").json()["credits"] == 0 98 99 100 def test_webhook_rejects_a_bad_signature(client): 101 event = checkout_event("cs_forged", account_id(client), "pack20") 102 assert post_webhook(client, event, secret="whsec_wrong").status_code == 400 103 assert client.get("/api/account").json()["credits"] == 0 104 105 106 def test_a_server_job_spends_a_credit(client): 107 buy(client, "single", email="cloud@example.com") 108 job_id = make_job(client, with_files=True) 109 assert start(client, job_id).status_code == 200 110 assert client.get("/api/account").json()["credits"] == 0 111 assert client.get(f"/api/jobs/{job_id}").json()["tier"] == "cloud" 112 assert get_job_row(job_id).credit_spent == 1 113 114 # And with the credit gone, the next one is refused again. 115 assert start(client, make_job(client)).status_code == 402 116 117 118 def test_failed_server_job_returns_the_credit(client): 119 buy(client, "single", email="refund@example.com") 120 job_id = make_job(client) 121 assert start(client, job_id).status_code == 200 122 assert client.get("/api/account").json()["credits"] == 0 123 124 runner.run_job(job_id) # fails: nothing was staged 125 126 assert client.get("/api/account").json()["credits"] == 1 127 assert get_job_row(job_id).credit_spent == 0 128 129 130 def test_cancelled_server_job_returns_the_credit(client): 131 buy(client, "single", email="cancel@example.com") 132 job_id = make_job(client) 133 assert start(client, job_id).status_code == 200 134 assert client.post(f"/api/jobs/{job_id}/cancel").status_code == 200 135 assert client.get("/api/account").json()["credits"] == 1 136 137 138 def test_cannot_touch_someone_elses_job(client, second_client): 139 job_id = make_job(client, JobStatus.succeeded, with_files=True) 140 assert second_client.get(f"/api/jobs/{job_id}/files/srt").status_code == 404 141 assert second_client.post(f"/api/jobs/{job_id}/cancel").status_code == 404 142 143 144 # --- checkout ---------------------------------------------------------------- 145 146 def test_checkout_sends_stripe_the_right_order(client, monkeypatch): 147 seen = {} 148 149 def create(**params): 150 seen.update(params) 151 return FakeStripeObject({"id": "cs_new", "url": "https://stripe.test/pay"}) 152 153 monkeypatch.setattr(stripe.checkout.Session, "create", create) 154 r = client.post("/api/billing/checkout", json={"plan_id": "pack5"}) 155 assert r.status_code == 200 and r.json()["url"] == "https://stripe.test/pay" 156 157 assert seen["mode"] == "payment" 158 assert seen["client_reference_id"] == account_id(client) 159 assert seen["metadata"]["plan_id"] == "pack5" 160 item = seen["line_items"][0]["price_data"] 161 assert item["unit_amount"] == 1699 and item["currency"] == "usd" 162 assert "recurring" not in item 163 assert seen["success_url"].startswith( 164 "https://example.test/api/billing/return?session_id={CHECKOUT_SESSION_ID}") 165 assert seen["customer_creation"] == "always" 166 167 168 def test_no_unlimited_plan_is_sold_by_default(client): 169 plans = client.get("/api/pricing").json()["plans"] 170 assert [p["id"] for p in plans] == ["single", "pack5", "pack20"] 171 assert [p["price_cents"] for p in plans] == [499, 1699, 3900] 172 assert not any(p["recurring"] for p in plans) 173 174 175 def test_subscription_checkout_is_recurring(client, monkeypatch, monthly_plan): 176 seen = {} 177 monkeypatch.setattr( 178 stripe.checkout.Session, "create", 179 lambda **p: seen.update(p) or FakeStripeObject({"url": "https://stripe.test/sub"}), 180 ) 181 assert client.post("/api/billing/checkout", 182 json={"plan_id": "monthly"}).status_code == 200 183 assert seen["mode"] == "subscription" 184 assert seen["line_items"][0]["price_data"]["recurring"] == {"interval": "month"} 185 assert seen["subscription_data"]["metadata"]["account_id"] == account_id(client) 186 assert "customer_creation" not in seen # not allowed in subscription mode 187 188 189 def test_unknown_plan_is_refused(client): 190 assert client.post("/api/billing/checkout", 191 json={"plan_id": "nope"}).status_code == 404 192 193 194 def test_return_trip_fulfils_and_the_webhook_then_adds_nothing(client, monkeypatch): 195 acct = account_id(client) 196 paid = checkout_object("cs_return", acct, "pack20", email="return@example.com") 197 monkeypatch.setattr(stripe.checkout.Session, "retrieve", 198 lambda sid: FakeStripeObject(paid)) 199 200 r = client.get("/api/billing/return?session_id=cs_return", follow_redirects=False) 201 assert r.status_code == 303 and r.headers["location"] == "/?checkout=paid" 202 assert client.get("/api/account").json()["credits"] == 20 203 204 assert post_webhook(client, checkout_event( 205 "cs_return", acct, "pack20", email="return@example.com")).status_code == 200 206 assert client.get("/api/account").json()["credits"] == 20 207 208 209 def test_return_trip_does_not_trust_an_unpaid_session(client, monkeypatch): 210 unpaid = checkout_object("cs_unpaid", account_id(client), "pack20", 211 payment_status="unpaid") 212 monkeypatch.setattr(stripe.checkout.Session, "retrieve", 213 lambda sid: FakeStripeObject(unpaid)) 214 r = client.get("/api/billing/return?session_id=cs_unpaid", follow_redirects=False) 215 assert r.headers["location"] == "/?checkout=pending" 216 assert client.get("/api/account").json()["credits"] == 0 217 218 219 # --- subscription ------------------------------------------------------------ 220 221 def subscription_event(kind, account, status, ends_in=30 * 86400, sub_id="sub_1"): 222 return { 223 "id": f"evt_{kind}_{time.time_ns()}", 224 "type": f"customer.subscription.{kind}", 225 "data": {"object": { 226 "id": sub_id, "object": "subscription", "status": status, 227 "customer": "cus_sub", "metadata": {"account_id": account}, 228 # Where newer API versions put it; _period_end reads both places. 229 "items": {"data": [{"current_period_end": int(time.time()) + ends_in}]}, 230 }}, 231 } 232 233 234 def test_subscription_lifts_every_limit_then_lapses(client): 235 acct = account_id(client) 236 assert post_webhook(client, subscription_event("created", acct, "active")).status_code == 200 237 238 a = client.get("/api/account").json() 239 assert a["subscribed"] is True and a["cloud_allowed"] is True 240 assert a["subscription_ends"] 241 242 # Server jobs start, and no credits are spent. 243 for _ in range(3): 244 job_id = make_job(client, with_files=True) 245 assert start(client, job_id).status_code == 200 246 assert get_job_row(job_id).credit_spent == 0 247 248 post_webhook(client, subscription_event("deleted", acct, "canceled")) 249 a = client.get("/api/account").json() 250 assert a["subscribed"] is False 251 assert start(client, make_job(client)).status_code == 402 252 253 254 def test_subscription_past_its_paid_period_does_not_count(client): 255 acct = account_id(client) 256 post_webhook(client, subscription_event("updated", acct, "active", ends_in=-3 * 86400)) 257 assert client.get("/api/account").json()["subscribed"] is False 258 259 260 def test_cannot_subscribe_twice(client, monkeypatch, monthly_plan): 261 post_webhook(client, subscription_event("created", account_id(client), "active")) 262 monkeypatch.setattr(stripe.checkout.Session, "create", 263 lambda **p: FakeStripeObject({"url": "x"})) 264 r = client.post("/api/billing/checkout", json={"plan_id": "monthly"}) 265 assert r.status_code == 400 and "already" in r.json()["detail"] 266 267 268 # --- paying with an email that already has an account ------------------------ 269 270 def test_paying_on_a_new_device_joins_the_existing_account(client, second_client): 271 buy(client, "pack5", email="same@example.com") 272 original = account_id(client) 273 274 # Second device, anonymous, converts a book, then pays with the same email. 275 job_id = make_job(second_client, JobStatus.succeeded, with_files=True) 276 device = account_id(second_client) 277 assert device != original 278 buy(second_client, "single", email="same@example.com") 279 280 # Its cookie now resolves to the original account, which holds everything. 281 a = second_client.get("/api/account").json() 282 assert a["id"] == original and a["email"] == "same@example.com" 283 assert a["credits"] == 6 284 assert get_job_row(job_id).account_id == original 285 assert get_account_row(device).merged_into == original 286 # ...and both browsers see the same jobs. 287 assert job_id in {j["id"] for j in client.get("/api/jobs").json()}