tests/test_migration.py (3375 bytes)
1 """A database created by the previous release must come up under this one. 2 3 Runs in a subprocess: the engine is bound at import, and the point is to watch 4 a cold start against a pre-existing file. 5 """ 6 7 from __future__ import annotations 8 9 import os 10 import sqlite3 11 import subprocess 12 import sys 13 from pathlib import Path 14 15 # The schema as deployed before accounts and tiers existed. 16 OLD_SCHEMA = """ 17 CREATE TABLE accounts ( 18 id VARCHAR(64) PRIMARY KEY, device_token VARCHAR(64) NOT NULL UNIQUE, 19 email VARCHAR(320), stripe_customer_id VARCHAR(64), 20 purchased_credits INTEGER NOT NULL, created_at DATETIME NOT NULL 21 ); 22 CREATE TABLE jobs ( 23 id VARCHAR(64) PRIMARY KEY, account_id VARCHAR(64) NOT NULL, 24 status VARCHAR(9) NOT NULL, language VARCHAR(16) NOT NULL, 25 splitter VARCHAR(16) NOT NULL, model VARCHAR(32) NOT NULL, 26 audio_filename VARCHAR(512) NOT NULL, text_filename VARCHAR(512) NOT NULL, 27 audio_parts INTEGER NOT NULL, cover_filename VARCHAR(512), 28 audio_bytes BIGINT NOT NULL, audio_duration_seconds FLOAT, 29 progress FLOAT NOT NULL, stage VARCHAR(128) NOT NULL, error TEXT, 30 billed INTEGER NOT NULL, created_at DATETIME NOT NULL, 31 started_at DATETIME, finished_at DATETIME 32 ); 33 CREATE TABLE artifacts ( 34 id VARCHAR(64) PRIMARY KEY, job_id VARCHAR(64) NOT NULL, 35 kind VARCHAR(32) NOT NULL, filename VARCHAR(512) NOT NULL, 36 storage_key VARCHAR(1024) NOT NULL, size_bytes BIGINT NOT NULL, 37 created_at DATETIME NOT NULL 38 ); 39 INSERT INTO accounts VALUES 40 ('acct_old', 'tok_old', NULL, NULL, 0, '2026-09-01 00:00:00'); 41 INSERT INTO jobs VALUES 42 ('job_old', 'acct_old', 'succeeded', 'ru', 'pysbd', 'tiny', 'a.m4b', 'a.epub', 43 1, NULL, 10, 1.0, 1.0, 'Done', NULL, 1, '2026-09-01 00:00:00', NULL, NULL); 44 """ 45 46 PROBE = """ 47 from backend.db import init_db, SessionLocal, Account, Job 48 init_db(); init_db() # twice: a restart must be a no-op 49 with SessionLocal() as s: 50 job = s.get(Job, "job_old") 51 acct = s.get(Account, "acct_old") 52 print(job.tier, job.credit_spent, job.status.value, acct.merged_into, acct.signed_in) 53 """ 54 55 56 def test_old_database_is_upgraded_in_place(tmp_path): 57 db = tmp_path / "subplz.db" 58 con = sqlite3.connect(db) 59 con.executescript(OLD_SCHEMA) 60 con.commit() 61 con.close() 62 63 env = {**os.environ, "SUBPLZ_WEB_DATA_DIR": str(tmp_path)} 64 root = Path(__file__).resolve().parent.parent 65 out = subprocess.run([sys.executable, "-c", PROBE], cwd=root, env=env, 66 capture_output=True, text=True, timeout=120) 67 assert out.returncode == 0, out.stderr 68 # Old rows read back with the new columns defaulted, nothing lost. 69 assert out.stdout.split() == ["free", "0", "succeeded", "None", "False"] 70 71 con = sqlite3.connect(db) 72 cols = lambda t: {r[1] for r in con.execute(f"PRAGMA table_info({t})")} # noqa: E731 73 assert {"tier", "credit_spent"} <= cols("jobs") 74 assert {"merged_into", "subscription_id", "subscription_status", 75 "subscription_period_end"} <= cols("accounts") 76 tables = {r[0] for r in con.execute("SELECT name FROM sqlite_master WHERE type='table'")} 77 assert {"login_tokens", "purchases"} <= tables 78 # The unique email index has to exist, or two accounts could share one. 79 indexes = {r[1]: r[2] for r in con.execute("PRAGMA index_list(accounts)")} 80 assert indexes.get("ix_accounts_email") == 1 81 con.close()