tests/test_copy_database.py (5170 bytes)
1 """tools/copy_database.py: a copy has every row of every table, unchanged.""" 2 3 from __future__ import annotations 4 5 import os 6 import subprocess 7 import sys 8 from datetime import date, datetime, timedelta, timezone 9 from pathlib import Path 10 11 import pytest 12 from hypothesis import HealthCheck, given, settings 13 from hypothesis import strategies as st 14 from sqlalchemy import create_engine, inspect, select 15 from sqlalchemy.orm import Session 16 17 from backend.captions import CaptionBalance 18 from backend.db import Account, Artifact, Base, Job, JobStatus, Purchase 19 from backend.subrep import SubrepLicence, WaitlistEntry 20 from tools.copy_database import copy_database 21 22 T0 = datetime(2026, 9, 22, tzinfo=timezone.utc) 23 text = st.text(st.characters(codec="utf-8", exclude_categories=("Cs", "Cc")), max_size=20) 24 25 26 @st.composite 27 def databases(draw): 28 accounts = [] 29 for a in range(draw(st.integers(1, 4))): 30 account = Account( 31 id=f"acct_{a}", device_token=f"dev_{a}", 32 email=draw(st.none() | st.just(f"reader{a}@example.com")), 33 purchased_credits=draw(st.integers(0, 500)), 34 daily_credit_on=draw(st.none() | st.dates(date(2026, 1, 1), date(2027, 1, 1))), 35 created_at=T0 + timedelta(seconds=draw(st.integers(0, 10**6))), 36 ) 37 accounts.append(account) 38 for j in range(draw(st.integers(0, 3))): 39 job_id = f"job_{a}_{j}" 40 accounts.append(Job( 41 id=job_id, account_id=account.id, status=draw(st.sampled_from(list(JobStatus))), 42 language="ja", splitter="pysbd", model="tiny", 43 audio_filename=draw(text), text_filename=draw(text), 44 audio_duration_seconds=draw(st.none() | st.floats(0, 1e5)), 45 progress=draw(st.floats(0, 1)), error=draw(st.none() | text), 46 )) 47 accounts.append(Artifact(id=f"art_{a}_{j}", job_id=job_id, kind="srt", 48 filename=draw(text), storage_key=f"{job_id}/out.srt")) 49 if draw(st.booleans()): 50 accounts.append(Purchase(id=f"buy_{a}", account_id=account.id, plan_id="pack10", 51 credits=10, amount_cents=499, stripe_session_id=f"cs_{a}")) 52 return accounts 53 54 55 def rows(url: str) -> dict[str, list[dict]]: 56 engine = create_engine(url) 57 with engine.connect() as c: 58 return {t.name: sorted((dict(r._mapping) for r in c.execute(select(t))), key=repr) 59 for t in Base.metadata.sorted_tables} 60 61 62 @settings(max_examples=40, deadline=None, 63 suppress_health_check=[HealthCheck.function_scoped_fixture]) 64 @given(objects=databases()) 65 def test_the_copy_has_every_row_unchanged(tmp_path_factory, objects): 66 tmp = tmp_path_factory.mktemp("copy") 67 source, target = f"sqlite:///{tmp / 'source.db'}", f"sqlite:///{tmp / 'target.db'}" 68 engine = create_engine(source) 69 Base.metadata.create_all(engine) 70 with Session(engine) as s: 71 s.add_all(objects) 72 s.commit() 73 74 counts = copy_database(source, target) 75 76 assert rows(target) == rows(source) 77 assert counts["accounts"] == len([o for o in objects if isinstance(o, Account)]) 78 79 80 def test_the_copy_refuses_a_target_with_rows(tmp_path): 81 source, target = f"sqlite:///{tmp_path / 's.db'}", f"sqlite:///{tmp_path / 't.db'}" 82 for url in (source, target): 83 engine = create_engine(url) 84 Base.metadata.create_all(engine) 85 with Session(engine) as s: 86 s.add(Account(id="acct_x", device_token="dev_x")) 87 s.commit() 88 with pytest.raises(RuntimeError, match="accounts"): 89 copy_database(source, target) 90 91 92 def test_the_command_copies_the_tables_of_every_model_module(tmp_path): 93 """The burst runbook runs the tool in a new interpreter. There, only the 94 modules that the tool imports put their tables in Base.metadata. This test 95 process imported backend.main already (conftest), so only a subprocess 96 shows a model module that the tool does not import.""" 97 source, target = f"sqlite:///{tmp_path / 'source.db'}", f"sqlite:///{tmp_path / 'target.db'}" 98 engine = create_engine(source) 99 Base.metadata.create_all(engine) 100 with Session(engine) as s: 101 s.add(Account(id="acct_1", device_token="dev_1", email="reader@example.com")) 102 s.add(CaptionBalance(account_id="acct_1", seconds_left=20 * 3600, seconds_used=60)) 103 s.add(SubrepLicence(account_id="acct_1", refresh_key="key_1", subscription_id="sub_1", 104 status="active", plan_id="subrep_year", period_end=T0)) 105 s.add(WaitlistEntry(kind="ios", email="reader@example.com", account_id="acct_1")) 106 s.commit() 107 108 env = {**os.environ, "SUBPLZ_WEB_DATA_DIR": str(tmp_path)} 109 root = Path(__file__).resolve().parent.parent 110 out = subprocess.run([sys.executable, "-m", "tools.copy_database", source, target], 111 cwd=root, env=env, capture_output=True, text=True, timeout=120, 112 check=False) 113 assert out.returncode == 0, out.stderr 114 115 assert set(Base.metadata.tables) <= set(inspect(create_engine(target)).get_table_names()) 116 assert rows(target) == rows(source)