# -*- coding: utf-8 -*- """SQLite 迁移测试（不联网）: uv run python tests/test_migrations.py""" import os import sqlite3 import sys import tempfile from pathlib import Path sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) sys.stdout.reconfigure(encoding="utf-8") from reaction_board.migrations import ( # noqa: E402 MIGRATIONS, Migration, MigrationError, current_version, prune_backups, run_migrations, ) EXPECTED_TABLES = { "video", "snapshot", "folder", "folder_resource", "up_blacklist", "up_stat", "up_profile", "candidate", "seen_aid", "notify", "fav_action", "kv", "visit", } def table_names(conn): rows = conn.execute( "SELECT name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%'" ).fetchall() return {row[0] for row in rows} def test_fresh_and_idempotent(): with tempfile.TemporaryDirectory() as tmp: conn = sqlite3.connect(Path(tmp) / "app.db") try: applied = run_migrations(conn) assert applied == [1, 2, 3, 4, 5, 6, 7, 8] assert current_version(conn) == 8 assert EXPECTED_TABLES <= table_names(conn) assert "schema_version" in table_names(conn) columns = { row[1] for row in conn.execute("PRAGMA table_info(folder_resource)") } assert "fav_time" in columns candidate_columns = { row[1] for row in conn.execute("PRAGMA table_info(candidate)") } assert {"duration", "play", "like", "favorite", "danmaku", "review"} <= candidate_columns video_columns = { row[1] for row in conn.execute("PRAGMA table_info(video)") } assert { "pages", "duration", "up_follower", "up_follower_at", "up_follower_backfilled", "up_face", } <= video_columns profile_columns = { row[1] for row in conn.execute("PRAGMA table_info(up_profile)") } assert { "mid", "follower_first", "following_first", "first_captured_at", "follower_latest", "last_refreshed", } <= profile_columns visit_columns = { row[1] for row in conn.execute("PRAGMA table_info(visit)") } assert { "stat_date", "ip", "path", "hits", "first_at", "last_at", } <= visit_columns snapshot_columns = { row[1] for row in conn.execute("PRAGMA table_info(snapshot)") } assert {"view_open", "open_captured_at"} <= snapshot_columns # 再跑一次不应重复应用 assert run_migrations(conn) == [] row = conn.execute("SELECT COUNT(*) FROM schema_version").fetchone() assert row[0] == 8 finally: conn.close() def test_backup_before_upgrade(): with tempfile.TemporaryDirectory() as tmp: db_path = Path(tmp) / "app.db" backup_dir = Path(tmp) / "backup" conn = sqlite3.connect(db_path) try: run_migrations(conn, backup_dir=backup_dir) conn.execute("INSERT INTO kv(k, v) VALUES ('hello', 'world')") conn.commit() v9 = Migration(9, "add probe", ("CREATE TABLE probe (id INTEGER PRIMARY KEY)",)) applied = run_migrations( conn, backup_dir=backup_dir, now=1_800_000_000, migrations=MIGRATIONS + (v9,), ) assert applied == [9] backups = list(backup_dir.glob("app-pre-v8-*.db")) assert len(backups) == 1 check = sqlite3.connect(backups[0]) try: assert check.execute("SELECT v FROM kv WHERE k='hello'").fetchone()[0] == "world" assert "probe" not in table_names(check) finally: check.close() finally: conn.close() def test_failed_migration_rolls_back(): with tempfile.TemporaryDirectory() as tmp: conn = sqlite3.connect(Path(tmp) / "app.db") bad = Migration( 1, "broken", ( "CREATE TABLE should_not_exist (id INTEGER PRIMARY KEY)", "CREATE TABLE broken (", ), ) try: try: run_migrations(conn, migrations=(bad,)) except MigrationError as exc: assert "已回滚" in str(exc) else: raise AssertionError("坏迁移应抛错") assert "should_not_exist" not in table_names(conn) assert current_version(conn) == 0 finally: conn.close() def test_prune_backups(): with tempfile.TemporaryDirectory() as tmp: backup_dir = Path(tmp) old = backup_dir / "app-pre-v1-old.db" fresh = backup_dir / "app-pre-v1-fresh.db" other = backup_dir / "keep.txt" for path in (old, fresh, other): path.write_bytes(b"x") now = 1_800_000_000 old_time = now - 20 * 86400 os.utime(old, (old_time, old_time)) os.utime(fresh, (now, now)) removed = prune_backups(backup_dir, keep_days=14, now=now) assert removed == [old] assert fresh.exists() and other.exists() def test_validate_migration_order(): with tempfile.TemporaryDirectory() as tmp: conn = sqlite3.connect(Path(tmp) / "app.db") try: try: run_migrations(conn, migrations=( Migration(2, "x", ()), Migration(1, "y", ()), )) except ValueError as exc: assert "升序" in str(exc) else: raise AssertionError("乱序迁移应失败") finally: conn.close() def test_v5_normalizes_unfaved(): with tempfile.TemporaryDirectory() as tmp: conn = sqlite3.connect(Path(tmp) / "app.db") try: run_migrations(conn, migrations=MIGRATIONS[:4]) conn.execute( "INSERT INTO video(aid, title, pubdate, status, first_seen) " "VALUES (1, 't', 1, 'unfaved', 1)" ) conn.execute( "INSERT INTO video(aid, title, pubdate, status, first_seen) " "VALUES (2, 't', 1, 'dropped', 1)" ) conn.commit() applied = run_migrations(conn) assert applied == [5, 6, 7, 8] rows = dict(conn.execute("SELECT aid, status FROM video").fetchall()) assert rows == {1: "active", 2: "dropped"} finally: conn.close() def test_v8_backfills_view_open(): with tempfile.TemporaryDirectory() as tmp: conn = sqlite3.connect(Path(tmp) / "app.db") try: run_migrations(conn, migrations=MIGRATIONS[:7]) conn.execute( "INSERT INTO video(aid, title, pubdate, status, first_seen) " "VALUES (1, 't', 1, 'active', 1)" ) conn.execute( 'INSERT INTO snapshot(stat_date, aid, "view", captured_at, ok) ' "VALUES ('2026-10-31', 1, 500, 100, 1)" ) conn.execute( 'INSERT INTO snapshot(stat_date, aid, "view", captured_at, ok) ' "VALUES ('2026-10-30', 1, NULL, 90, 0)" ) conn.commit() assert run_migrations(conn) == [8] ok_row = conn.execute( "SELECT view_open, open_captured_at FROM snapshot " "WHERE stat_date='2026-10-31'" ).fetchone() assert ok_row[0] == 500 and ok_row[1] == 100 fail_row = conn.execute( "SELECT view_open, open_captured_at FROM snapshot " "WHERE stat_date='2026-10-30'" ).fetchone() assert fail_row[0] is None and fail_row[1] is None finally: conn.close() def main(): test_fresh_and_idempotent() test_backup_before_upgrade() test_failed_migration_rolls_back() test_prune_backups() test_validate_migration_order() test_v5_normalizes_unfaved() test_v8_backfills_view_open() print("OK: SQLite 迁移、备份与回滚测试全部通过") if __name__ == "__main__": main()