tests/test_migration.py (3760 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 job.free_credit_spent, acct.free_credits_used, acct.email_verified, 54 acct.daily_credit_on, job.daily_credit_on) 55 """ 56 57 58 def test_old_database_is_upgraded_in_place(tmp_path): 59 db = tmp_path / "subplz.db" 60 con = sqlite3.connect(db) 61 con.executescript(OLD_SCHEMA) 62 con.commit() 63 con.close() 64 65 env = {**os.environ, "SUBPLZ_WEB_DATA_DIR": str(tmp_path)} 66 root = Path(__file__).resolve().parent.parent 67 out = subprocess.run([sys.executable, "-c", PROBE], cwd=root, env=env, 68 capture_output=True, text=True, timeout=120) 69 assert out.returncode == 0, out.stderr 70 # Old rows read back with the new columns defaulted, nothing lost. 71 assert out.stdout.split() == [ 72 "free", "0", "succeeded", "None", "False", "0", "0", "0", "None", "None", 73 ] 74 75 con = sqlite3.connect(db) 76 cols = lambda t: {r[1] for r in con.execute(f"PRAGMA table_info({t})")} # noqa: E731 77 assert {"tier", "credit_spent", "free_credit_spent", "daily_credit_on", 78 "plan_credit_spent"} <= cols("jobs") 79 assert {"merged_into", "subscription_id", "subscription_status", 80 "subscription_period_end", "free_credits_used", "email_verified", 81 "daily_credit_on", "subscription_plan_id", 82 "subscription_credits_used"} <= cols("accounts") 83 tables = {r[0] for r in con.execute("SELECT name FROM sqlite_master WHERE type='table'")} 84 assert {"login_tokens", "purchases"} <= tables 85 # The unique email index has to exist, or two accounts could share one. 86 indexes = {r[1]: r[2] for r in con.execute("PRAGMA index_list(accounts)")} 87 assert indexes.get("ix_accounts_email") == 1 88 con.close()