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

tests/conftest.py (5001 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 )
27 
28 import pytest  # noqa: E402
29 from fastapi.testclient import TestClient  # noqa: E402
30 
31 from backend import api  # noqa: E402
32 from backend.db import (  # noqa: E402
33     Account, Artifact, Job, JobStatus, SessionLocal, init_db,
34 )
35 from backend.main import app  # noqa: E402
36 from backend.storage import storage  # noqa: E402
37 
38 WEBHOOK_SECRET = "whsec_test_secret"
39 
40 init_db()
41 
42 
43 @pytest.fixture(autouse=True)
44 def _no_real_work(monkeypatch):
45     """Starting a job must not actually launch an alignment."""
46     monkeypatch.setattr(api.queue, "enqueue", lambda job_id: None)
47 
48 
49 @pytest.fixture
50 def client():
51     """A browser: keeps its cookie, so it stays one account across requests."""
52     with TestClient(app, base_url="http://testserver") as c:
53         yield c
54 
55 
56 @pytest.fixture
57 def second_client():
58     with TestClient(app, base_url="http://testserver") as c:
59         yield c
60 
61 
62 def account_id(client: TestClient) -> str:
63     return client.get("/api/account").json()["id"]
64 
65 
66 def make_job(
67     client: TestClient, status: JobStatus = JobStatus.draft, with_files: bool = False
68 ) -> str:
69     """A job row owned by `client`, skipping the upload (which needs real audio)."""
70     acct = account_id(client)
71     with SessionLocal() as s:
72         job = Job(
73             account_id=acct, status=status, language="en", splitter="pysbd",
74             model="tiny", audio_filename="book.m4b", text_filename="book.epub",
75         )
76         s.add(job)
77         s.commit()
78         job_id = job.id
79 
80         if with_files:
81             for kind, name in [("srt", "book.en.srt"), ("video", "book.en.mp4"),
82                                ("video_embedded", "book.en.mkv")]:
83                 src = Path(_TMP) / f"{job_id}-{name}"
84                 src.write_bytes(b"x" * 10)
85                 key = f"{job_id}/{name}"
86                 storage.put_file(key, src)
87                 s.add(Artifact(job_id=job_id, kind=kind, filename=name,
88                                storage_key=key, size_bytes=10))
89             s.commit()
90     return job_id
91 
92 
93 def get_account_row(account_id_: str) -> Account:
94     with SessionLocal() as s:
95         return s.get(Account, account_id_)
96 
97 
98 def get_job_row(job_id: str) -> Job:
99     with SessionLocal() as s:
100         return s.get(Job, job_id)
101 
102 
103 def signed(payload: dict, secret: str = WEBHOOK_SECRET) -> tuple[bytes, dict]:
104     """A webhook body plus the header Stripe would have sent with it."""
105     body = json.dumps(payload).encode()
106     stamp = int(time.time())
107     digest = hmac.new(
108         secret.encode(), f"{stamp}.".encode() + body, hashlib.sha256
109     ).hexdigest()
110     return body, {
111         "stripe-signature": f"t={stamp},v1={digest}",
112         "content-type": "application/json",
113     }
114 
115 
116 def checkout_event(
117     session_id: str, account: str, plan: str, email: str | None = "buyer@example.com",
118     job_id: str = "", payment_status: str = "paid", subscription: str | None = None,
119     customer: str | None = "cus_test",
120 ) -> dict:
121     return {
122         "id": f"evt_{session_id}",
123         "type": "checkout.session.completed",
124         "data": {"object": checkout_object(
125             session_id, account, plan, email, job_id, payment_status,
126             subscription, customer,
127         )},
128     }
129 
130 
131 def checkout_object(session_id, account, plan, email="buyer@example.com",
132                     job_id="", payment_status="paid", subscription=None,
133                     customer="cus_test") -> dict:
134     return {
135         "id": session_id,
136         "object": "checkout.session",
137         "client_reference_id": account,
138         "customer": customer,
139         "customer_details": {"email": email} if email else None,
140         "payment_status": payment_status,
141         "amount_total": 1299,
142         "currency": "usd",
143         "subscription": subscription,
144         "metadata": {"account_id": account, "plan_id": plan, "job_id": job_id},
145     }
146 
147 
148 def post_webhook(client: TestClient, event: dict, secret: str = WEBHOOK_SECRET):
149     body, headers = signed(event, secret)
150     return client.post("/api/billing/webhook", content=body, headers=headers)
151 
152 
153 class FakeStripeObject:
154     """Stands in for an SDK return value: not a dict, but has to_dict()."""
155 
156     def __init__(self, data: dict):
157         self._data = data
158 
159     def to_dict(self) -> dict:
160         return self._data