subplz-web/tests/conftest.py - Recently Written

tests/conftest.py (5409 bytes)
1 """Test harness: a throwaway data dir, billing on, Stripe faked at the SDK edge.
2 
3 Settings are read once at import, so the environment has to be in place before
4 anything under `backend` is imported - hence doing it here, at module load.
5 """
6 
7 from __future__ import annotations
8 
9 import hashlib
10 import hmac
11 import json
12 import os
13 import tempfile
14 import time
15 from pathlib import Path
16 
17 _TMP = tempfile.mkdtemp(prefix="subplz-web-tests-")
18 os.environ.update(
19     SUBPLZ_WEB_DATA_DIR=_TMP,
20     SUBPLZ_WEB_BILLING_ENABLED="true",
21     SUBPLZ_WEB_STRIPE_SECRET_KEY="sk_test_dummy",
22     SUBPLZ_WEB_STRIPE_WEBHOOK_SECRET="whsec_test_secret",
23     SUBPLZ_WEB_PUBLIC_BASE_URL="https://example.test",
24     SUBPLZ_WEB_MATCH_CHECK="false",
25     SUBPLZ_WEB_SMTP_HOST="",
26     # No test may use a real Google Play key from a .env file.
27     SUBPLZ_WEB_PLAY_SERVICE_ACCOUNT_FILE="",
28 )
29 
30 import pytest  # noqa: E402
31 from fastapi.testclient import TestClient  # noqa: E402
32 
33 from backend import api  # noqa: E402
34 from backend.db import (  # noqa: E402
35     Account, Artifact, Job, JobStatus, SessionLocal, init_db,
36 )
37 from backend.main import app  # noqa: E402
38 from backend.storage import storage  # noqa: E402
39 
40 WEBHOOK_SECRET = "whsec_test_secret"
41 
42 init_db()
43 
44 
45 @pytest.fixture(autouse=True)
46 def _no_real_work(monkeypatch):
47     """Starting a job must not actually launch an alignment."""
48     monkeypatch.setattr(api.queue, "enqueue", lambda job_id, paid=True: None)
49 
50 
51 @pytest.fixture(autouse=True)
52 def _no_stripe_prices(monkeypatch):
53     """Stripe has no prices for the plans, unless a test says so. No test may
54     reach the real Stripe API."""
55     import stripe
56     monkeypatch.setattr(stripe.Price, "list", lambda **kw: FakeStripeObject({"data": []}))
57 
58 
59 @pytest.fixture
60 def client():
61     """A browser: keeps its cookie, so it stays one account across requests."""
62     with TestClient(app, base_url="http://testserver") as c:
63         yield c
64 
65 
66 @pytest.fixture
67 def second_client():
68     with TestClient(app, base_url="http://testserver") as c:
69         yield c
70 
71 
72 def account_id(client: TestClient) -> str:
73     return client.get("/api/account").json()["id"]
74 
75 
76 def make_job(
77     client: TestClient, status: JobStatus = JobStatus.draft, with_files: bool = False
78 ) -> str:
79     """A job row owned by `client`, skipping the upload (which needs real audio)."""
80     acct = account_id(client)
81     with SessionLocal() as s:
82         job = Job(
83             account_id=acct, status=status, language="en", splitter="pysbd",
84             model="tiny", audio_filename="book.m4b", text_filename="book.epub",
85         )
86         s.add(job)
87         s.commit()
88         job_id = job.id
89 
90         if with_files:
91             for kind, name in [("srt", "book.en.srt"), ("video", "book.en.mp4"),
92                                ("video_embedded", "book.en.mkv")]:
93                 src = Path(_TMP) / f"{job_id}-{name}"
94                 src.write_bytes(b"x" * 10)
95                 key = f"{job_id}/{name}"
96                 storage.put_file(key, src)
97                 s.add(Artifact(job_id=job_id, kind=kind, filename=name,
98                                storage_key=key, size_bytes=10))
99             s.commit()
100     return job_id
101 
102 
103 def get_account_row(account_id_: str) -> Account:
104     with SessionLocal() as s:
105         return s.get(Account, account_id_)
106 
107 
108 def get_job_row(job_id: str) -> Job:
109     with SessionLocal() as s:
110         return s.get(Job, job_id)
111 
112 
113 def signed(payload: dict, secret: str = WEBHOOK_SECRET) -> tuple[bytes, dict]:
114     """A webhook body plus the header Stripe would have sent with it."""
115     body = json.dumps(payload).encode()
116     stamp = int(time.time())
117     digest = hmac.new(
118         secret.encode(), f"{stamp}.".encode() + body, hashlib.sha256
119     ).hexdigest()
120     return body, {
121         "stripe-signature": f"t={stamp},v1={digest}",
122         "content-type": "application/json",
123     }
124 
125 
126 def checkout_event(
127     session_id: str, account: str, plan: str, email: str | None = "buyer@example.com",
128     job_id: str = "", payment_status: str = "paid", subscription: str | None = None,
129     customer: str | None = "cus_test",
130 ) -> dict:
131     return {
132         "id": f"evt_{session_id}",
133         "type": "checkout.session.completed",
134         "data": {"object": checkout_object(
135             session_id, account, plan, email, job_id, payment_status,
136             subscription, customer,
137         )},
138     }
139 
140 
141 def checkout_object(session_id, account, plan, email="buyer@example.com",
142                     job_id="", payment_status="paid", subscription=None,
143                     customer="cus_test") -> dict:
144     return {
145         "id": session_id,
146         "object": "checkout.session",
147         "client_reference_id": account,
148         "customer": customer,
149         "customer_details": {"email": email} if email else None,
150         "payment_status": payment_status,
151         "amount_total": 1299,
152         "currency": "usd",
153         "subscription": subscription,
154         "metadata": {"account_id": account, "plan_id": plan, "job_id": job_id},
155     }
156 
157 
158 def post_webhook(client: TestClient, event: dict, secret: str = WEBHOOK_SECRET):
159     body, headers = signed(event, secret)
160     return client.post("/api/billing/webhook", content=body, headers=headers)
161 
162 
163 class FakeStripeObject:
164     """Stands in for an SDK return value: not a dict, but has to_dict()."""
165 
166     def __init__(self, data: dict):
167         self._data = data
168 
169     def to_dict(self) -> dict:
170         return self._data