import json import sqlite3 import shutil import tempfile import unittest from contextlib import closing from pathlib import Path from peach.migrations import plan, upgrade MIGRATIONS = Path(__file__).resolve().parents[1] / "migrations" class MigrationTests(unittest.TestCase): def test_reference_index_upgrade_preserves_rows_and_removes_only_legacy_duplicates(self): previous = self.root / 'previous-indexes' previous.mkdir() for path in MIGRATIONS.glob('*.sql'): if path.name < '0052_': shutil.copyfile(path, previous / path.name) upgrade(self.db, previous) with closing(sqlite3.connect(self.db)) as connection: connection.execute("INSERT INTO asset(id,location,path,name,medium,code) " "VALUES(1,'local','sample.mp4','sample.mp4','video','EXM-001')") for name, columns in [('ix_asset_name', 'name'), ('ix_asset_loc', 'location'), ('ix_asset_size', 'size'), ('ix_asset_hash', 'hash_kind,hash')]: connection.execute(f'CREATE INDEX {name} ON asset({columns})') connection.execute('CREATE INDEX ix_asset_unrelated ON asset(duration)') connection.executemany("INSERT INTO task_run(id,task_key,trigger,status,root_run_id,finished_at) " "VALUES(?,'sample','manual','succeeded',?,'now')", [(index, index) for index in range(1, 1501)]) connection.commit() asset = connection.execute('SELECT * FROM asset').fetchall() self.assertEqual([item.version for item in upgrade(self.db, MIGRATIONS)], ['0052', '0053', '0054']) with closing(sqlite3.connect(self.db)) as connection: self.assertEqual(connection.execute('SELECT * FROM asset').fetchall(), asset) self.assertEqual(connection.execute('SELECT count(*) FROM task_run').fetchone()[0], 1500) plan_rows = connection.execute('EXPLAIN QUERY PLAN SELECT id FROM task_run ' 'WHERE root_run_id=1').fetchall() self.assertFalse(any('SCAN task_run' in row[3] for row in plan_rows), plan_rows) names = {row[1] for row in connection.execute('PRAGMA index_list(asset)')} self.assertTrue(names.isdisjoint({'ix_asset_name', 'ix_asset_loc', 'ix_asset_size', 'ix_asset_hash'})) self.assertIn('ix_asset_unrelated', names) self.assertEqual(connection.execute('PRAGMA foreign_key_check').fetchall(), []) self.assertEqual(upgrade(self.db, MIGRATIONS), []) def test_playback_progress_upgrade_keeps_rows_and_stops_bumping_the_asset_revision(self): previous = self.root / 'previous-revision' previous.mkdir() for path in MIGRATIONS.glob('*.sql'): if path.name < '0054_': shutil.copyfile(path, previous / path.name) upgrade(self.db, previous) with closing(sqlite3.connect(self.db)) as connection: connection.execute("INSERT INTO asset(id,location,path,name,medium,play_count) " "VALUES(1,'local','a.mp4','a.mp4','video',2)") connection.commit() asset = connection.execute('SELECT * FROM asset').fetchall() self.assertEqual([item.version for item in upgrade(self.db, MIGRATIONS)], ['0054']) def revision(connection): return connection.execute("SELECT n FROM ledger_revision WHERE tbl='asset'").fetchone()[0] with closing(sqlite3.connect(self.db)) as connection, connection: self.assertEqual(connection.execute('SELECT * FROM asset').fetchall(), asset) before = revision(connection) connection.execute("UPDATE asset SET play_seconds=12,leave_ratio=0.2,max_reached=0.2," "seek_count=1,last_played=1 WHERE id=1") self.assertEqual(revision(connection), before) connection.execute("UPDATE asset SET play_count=3,last_played=2 WHERE id=1") self.assertEqual(revision(connection), before + 1) connection.execute("UPDATE asset SET rating=80 WHERE id=1") self.assertEqual(revision(connection), before + 2) def test_foreign_key_failure_rolls_back_data_schema_and_migration_record(self): directory = self.root / 'boundary-migrations' directory.mkdir() (directory / '9000_boundary.sql').write_text( 'CREATE TABLE parent(id INTEGER PRIMARY KEY);' 'CREATE TABLE child(id INTEGER PRIMARY KEY,parent_id INTEGER REFERENCES parent(id));' 'INSERT INTO parent VALUES(1);INSERT INTO child VALUES(1,1);', encoding='utf-8') upgrade(self.db, directory) (directory / '9001_invalid.sql').write_text( '-- peach:foreign_keys=off\nCREATE TABLE introduced(id INTEGER);DELETE FROM parent;', encoding='utf-8') with self.assertRaisesRegex(RuntimeError, '违约'): upgrade(self.db, directory) with closing(sqlite3.connect(self.db)) as connection: self.assertEqual(connection.execute('SELECT id FROM parent').fetchall(), [(1,)]) self.assertEqual(connection.execute('PRAGMA foreign_key_check').fetchall(), []) self.assertIsNone(connection.execute("SELECT name FROM sqlite_schema WHERE name='introduced'").fetchone()) self.assertEqual([item.version for item in plan(self.db, directory)[1]], ['9001']) def test_review_history_upgrade_preserves_legacy_decisions(self): prior = self.root / 'prior' prior.mkdir() for path in MIGRATIONS.glob('*.sql'): if path.name < '0050_': shutil.copyfile(path, prior / path.name) upgrade(self.db, prior) with closing(sqlite3.connect(self.db)) as connection: connection.executemany('INSERT INTO review_decision(category,item_key,status,reviewer,note,updated_at) VALUES(?,?,?,?,?,?)', [('creator_tags', state, state, '本人', '旧记录', 'old') for state in ('approved', 'rejected', 'skipped')]) connection.commit() before = connection.execute('SELECT * FROM review_decision ORDER BY item_key').fetchall() self.assertEqual([item.version for item in upgrade(self.db, MIGRATIONS)], ['0050', '0051', '0052', '0053', '0054']) self.assertEqual(connection.execute('SELECT * FROM review_decision ORDER BY item_key').fetchall(), before) self.assertEqual(connection.execute('SELECT COUNT(*) FROM review_decision_history').fetchone()[0], 0) def revision(): return connection.execute("SELECT n FROM ledger_revision WHERE tbl='review_decision_history'").fetchone()[0] self.assertEqual(revision(), 0) connection.execute("INSERT INTO review_decision_history(category,item_key,status,reviewer,note,decided_at,restored_at) VALUES('creator_tags','kept','skipped','本人','旧记录','old','now')") self.assertEqual(revision(), 1) connection.execute("UPDATE review_decision_history SET note='保留原文' WHERE item_key='kept'") self.assertEqual(revision(), 2) connection.execute("DELETE FROM review_decision_history WHERE item_key='kept'") self.assertEqual(revision(), 3) connection.commit() self.assertEqual(connection.execute('PRAGMA integrity_check').fetchone()[0], 'ok') self.assertEqual(connection.execute('PRAGMA foreign_key_check').fetchall(), []) self.assertEqual(upgrade(self.db, MIGRATIONS), []) def test_task_reference_migration_preserves_rows_and_existing_links(self): prior = self.root.resolve() / 'prior' prior.mkdir() for path in MIGRATIONS.glob('*.sql'): if path.name < '0049_': shutil.copyfile(path, prior / path.name) upgrade(self.db, prior) with closing(sqlite3.connect(self.db)) as connection: connection.executescript(""" INSERT INTO task_run(id,task_key,trigger,status,finished_at,parent_run_id,root_run_id,result_summary,error) VALUES(10,'parent','manual','succeeded','t',NULL,10,'{}',NULL), (11,'child','manual','succeeded','t',10,10,'{"changed":1}',NULL), (12,'child','manual','failed','t',100,100,'{}','保留错误'), (13,'child','manual','pending',NULL,100,10,'{}',NULL), (14,'child','manual','running',NULL,11,100,'{}',NULL), (15,'standalone','manual','succeeded','t',NULL,NULL,'{}',NULL); INSERT INTO asset(id,location,path,medium) VALUES(1,'local','sample.mp4','video'); INSERT INTO asset_tag(asset_id,tag,source) VALUES(1,'示例','user'); """) columns = [row[1] for row in connection.execute('PRAGMA table_info(task_run)') if row[1] not in ('parent_run_id', 'root_run_id')] query = f"SELECT {','.join(columns)} FROM task_run ORDER BY id" tasks = connection.execute(query).fetchall() assets = connection.execute('SELECT * FROM asset').fetchall() tags = connection.execute('SELECT * FROM asset_tag').fetchall() self.assertEqual(len(connection.execute('PRAGMA foreign_key_check').fetchall()), 4) self.assertEqual([m.version for m in upgrade(self.db, MIGRATIONS)], ['0049', '0050', '0051', '0052', '0053', '0054']) self.assertEqual(connection.execute('PRAGMA foreign_key_check').fetchall(), []) self.assertEqual(connection.execute('PRAGMA integrity_check').fetchone()[0], 'ok') self.assertEqual(connection.execute('SELECT id,parent_run_id,root_run_id FROM task_run ORDER BY id').fetchall(), [(10,None,10),(11,10,10),(12,None,None),(13,None,10),(14,11,None),(15,None,None)]) self.assertEqual(connection.execute(query).fetchall(), tasks) self.assertEqual(connection.execute('SELECT * FROM asset').fetchall(), assets) self.assertEqual(connection.execute('SELECT * FROM asset_tag').fetchall(), tags) self.assertEqual(upgrade(self.db, MIGRATIONS), []) def test_missing_database_status_does_not_create_a_file(self): self.assertTrue(plan(self.db, MIGRATIONS)[1]) self.assertFalse(self.db.exists()) def test_disposal_reason_migration_preserves_assets_tags_and_unknown_reasons(self): prior = self.root.resolve() / 'prior' prior.mkdir() for path in MIGRATIONS.glob('*.sql'): if path.name < '0046_': shutil.copyfile(path, prior / path.name) upgrade(self.db, prior) connection = sqlite3.connect(self.db) try: connection.execute("INSERT INTO asset(id,location,path,medium,disposal) VALUES(1,'local','sample.mp4','video','trash')") connection.execute("INSERT INTO asset_tag(asset_id,tag,source) VALUES(1,'示例','user')") connection.commit() self.assertEqual([migration.version for migration in upgrade(self.db, MIGRATIONS)], ['0046', '0047', '0048', '0049', '0050', '0051', '0052', '0053', '0054']) self.assertEqual(connection.execute("SELECT disposal,disposal_reason FROM asset WHERE id=1").fetchone(), ('trash', None)) self.assertEqual(connection.execute("SELECT tag,source FROM asset_tag WHERE asset_id=1").fetchall(), [('示例', 'user')]) self.assertEqual(connection.execute("SELECT count(*) FROM asset").fetchone()[0], 1) finally: connection.close() def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.root = Path(self.tmp.name) self.db = self.root / "ledger.db" def tearDown(self): self.tmp.cleanup() def test_empty_database_upgrades_and_is_idempotent(self): sqlite3.connect(self.db).close() backup = self.root / "before.db" done = upgrade(self.db, MIGRATIONS, backup) self.assertEqual([m.version for m in done], [f'{number:04d}' for number in range(55)]) self.assertTrue(backup.exists()) con = sqlite3.connect(self.db) tables = {row[0] for row in con.execute( "SELECT name FROM sqlite_schema WHERE type='table'" )} versions = [row[0] for row in con.execute( "SELECT version FROM schema_migration ORDER BY version" )] con.close() self.assertTrue({"asset", "profile", "media_binding", "activity_event", "provider_profile", "entity", "entity_alias", "entity_external_ref", "asset_entity", "entity_link", "entity_search_term", "watch_queue", "asset_preference", "asset_quality_goal", "playlist", "playlist_item", "asset_tag_preference", "asset_search", "follow_playback", "genre_decision", "asset_subtitle", "task_run", "performer_profile", "code_sample_image", "entity_classification", "entity_identity_link", "seed_pack_history", "schema_migration"} <= tables) self.assertEqual(versions, [f'{number:04d}' for number in range(55)]) self.assertEqual(upgrade(self.db, MIGRATIONS), []) self.assertEqual(plan(self.db, MIGRATIONS)[1], []) def test_old_schema_columns_are_reconciled(self): con = sqlite3.connect(self.db) con.executescript(""" CREATE TABLE asset( id INTEGER PRIMARY KEY, location TEXT NOT NULL, path TEXT NOT NULL, name TEXT, medium TEXT, size INTEGER, mtime TEXT, hash_kind TEXT, hash TEXT, creator TEXT, series TEXT, code TEXT, duration REAL, width INTEGER, height INTEGER, vcodec TEXT, fps REAL, has_audio INTEGER, ctx_length TEXT, ctx_orient TEXT, ctx_quality TEXT, ctx_pace TEXT, ctx_people TEXT, play_count INTEGER DEFAULT 0, last_played TEXT, rating INTEGER, o_count INTEGER, watch_ratio REAL, source_id INTEGER, stash_scene_id INTEGER, snapshot_path TEXT, first_seen TEXT, last_seen TEXT, UNIQUE(location,path)); CREATE TABLE asset_tag(asset_id INTEGER,tag TEXT,confidence REAL,source TEXT, UNIQUE(asset_id,tag)); """) con.close() upgrade(self.db, MIGRATIONS) con = sqlite3.connect(self.db) columns = {row[1] for row in con.execute("PRAGMA table_info(asset)")} con.close() self.assertTrue({"studio", "feedback", "disposal", "leave_ratio", "play_seconds", "feedback_at", "seek_count", "max_reached", "release_date", "catalog_title", "original_title"} <= columns) def test_checksum_change_is_rejected(self): sqlite3.connect(self.db).close() upgrade(self.db, MIGRATIONS) con = sqlite3.connect(self.db) con.execute("UPDATE schema_migration SET checksum='changed' WHERE version='0001'") con.commit() con.close() with self.assertRaises(RuntimeError): plan(self.db, MIGRATIONS) def test_missing_migration_directory_is_rejected_instead_of_reported_empty(self): """migration 目录不存在时拒绝，不要把空账本当成迁移到位的账本。 `pip install .`（非 editable）只装 `src/` 下的包，仓库根的 `migrations/` 不进 wheel。目录缺失若被读成「0 个迁移」，`peach init` 会打印成功并留下 0 字节账本， `peach migrate status` 还跟着报 0 pending。 """ absent = self.root / "no-such-migrations" with self.assertRaises(FileNotFoundError) as caught: plan(self.db, absent) self.assertIn(str(absent), str(caught.exception)) with self.assertRaises(FileNotFoundError): upgrade(self.db, absent) self.assertFalse(self.db.exists()) def test_empty_migration_directory_reports_nothing_to_apply(self): empty = self.root / "empty-migrations" empty.mkdir() self.assertEqual(upgrade(self.db, empty), []) def test_entity_migration_backfills_flattened_relations(self): base_migrations = self.root / "base-migrations" base_migrations.mkdir() for name in ("0000_legacy_schema.sql", "0001_core_boundaries.sql"): shutil.copyfile(MIGRATIONS / name, base_migrations / name) sqlite3.connect(self.db).close() upgrade(self.db, base_migrations) con = sqlite3.connect(self.db) con.executescript(""" INSERT INTO asset(id,location,path,name,medium,creator,studio,series) VALUES(1,'local','one.mp4','one.mp4','video','Owner A','Studio A','Series A'); INSERT INTO asset_tag(asset_id,tag,confidence,source) VALUES(1,'演员:Performer A',1.0,'stash:performer'), (1,'Tag A',0.9,'stash:tag'); """) con.close() upgrade(self.db, MIGRATIONS) con = sqlite3.connect(self.db) entities = set(con.execute("SELECT kind,canonical_name FROM entity")) relations = set(con.execute( "SELECT e.kind,ae.role,ae.source FROM asset_entity ae " "JOIN entity e ON e.id=ae.entity_id WHERE ae.asset_id=1" )) con.close() self.assertTrue({('performer','Performer A'),('tag','Tag A'), ('studio','Studio A'),('creator','Owner A'), ('series','Series A')} <= entities) self.assertTrue({('performer','performer','stash:performer'), ('tag','tag','stash:tag'),('studio','studio','legacy:asset'), ('creator','creator','legacy:asset'), ('series','series','legacy:asset')} <= relations) def test_late_legacy_tags_are_backfilled_by_0005(self): base_migrations = self.root / "base-migrations" base_migrations.mkdir() for name in ( "0000_legacy_schema.sql", "0001_core_boundaries.sql", "0002_canonical_entities.sql", "0003_entity_pages_and_watch_queue.sql", "0004_asset_search_fts.sql", ): shutil.copyfile(MIGRATIONS / name, base_migrations / name) sqlite3.connect(self.db).close() upgrade(self.db, base_migrations) con = sqlite3.connect(self.db) con.executescript(""" INSERT INTO asset(id,location,path,name,medium) VALUES(1,'local','late.mp4','late.mp4','video'); INSERT INTO asset_tag(asset_id,tag,confidence,source) VALUES(1,'足交',0.6,'vision_creator'), (1,'演员:Late Performer',1.0,'stash:performer'); """) con.close() upgrade(self.db, MIGRATIONS) con = sqlite3.connect(self.db) relations = set(con.execute( "SELECT e.kind,e.canonical_name,ae.role,ae.source,ae.confidence " "FROM asset_entity ae JOIN entity e ON e.id=ae.entity_id " "WHERE ae.asset_id=1" )) search_entities = con.execute( "SELECT entities FROM asset_search WHERE asset_id=1" ).fetchone()[0] con.close() self.assertIn(('tag', '足交', 'tag', 'vision_creator', 0.6), relations) self.assertIn(('performer', 'Late Performer', 'performer', 'stash:performer', 1.0), relations) self.assertIn('足交', search_entities) self.assertIn('Late Performer', search_entities) def test_structural_creator_cleanup_keeps_real_creator(self): base_migrations = self.root / "base-migrations" base_migrations.mkdir() for path in sorted(MIGRATIONS.glob("*.sql")): if not path.name.startswith(("0007_", "0009_")): shutil.copyfile(path, base_migrations / path.name) sqlite3.connect(self.db).close() upgrade(self.db, base_migrations) connection = sqlite3.connect(self.db) connection.executescript(""" INSERT INTO asset(id,location,path,name,medium,creator) VALUES(1,'115','B:/xxr/门槛/one.mp4','one.mp4','video','门槛'); INSERT INTO entity(id,kind,canonical_name,normalized_name,created_at,updated_at) VALUES(101,'creator','门槛','门槛','now','now'), (102,'creator','Actual Creator','actual creator','now','now'); INSERT INTO asset_entity(asset_id,entity_id,role,source,confidence) VALUES(1,101,'creator','legacy:asset',0.8), (1,102,'creator','reviewed',1.0); """) connection.close() upgrade(self.db, MIGRATIONS) connection = sqlite3.connect(self.db) self.assertEqual(connection.execute( "SELECT creator FROM asset WHERE id=1" ).fetchone()[0], "Actual Creator") self.assertEqual(connection.execute( "SELECT e.canonical_name FROM asset_entity ae JOIN entity e ON e.id=ae.entity_id " "WHERE ae.asset_id=1 AND e.kind='creator'" ).fetchall(), [("Actual Creator",)]) connection.close() def test_asce_cleanup_removes_only_false_creator_board_assertions(self): base_migrations = self.root / "base-migrations" base_migrations.mkdir() for path in sorted(MIGRATIONS.glob("*.sql")): if not path.name.startswith("0009_"): shutil.copyfile(path, base_migrations / path.name) sqlite3.connect(self.db).close() upgrade(self.db, base_migrations) connection = sqlite3.connect(self.db) connection.executescript(""" INSERT INTO asset(id,location,path,name,medium,creator) VALUES(1,'115','B:/kkg/asce/one.mp4','one.mp4','video','asce'); INSERT INTO entity(id,kind,canonical_name,normalized_name,created_at,updated_at) VALUES(101,'creator','asce','asce','now','now'), (102,'tag','制服','制服','now','now'), (103,'tag','原创','原创','now','now'); INSERT INTO asset_entity(asset_id,entity_id,role,source,confidence) VALUES(1,101,'creator','legacy:asset',0.8), (1,102,'tag','vision_creator',0.6), (1,103,'tag','name',0.9); INSERT INTO asset_tag(asset_id,tag,confidence,source) VALUES(1,'制服',0.6,'vision_creator'),(1,'原创',0.9,'name'); """) connection.close() upgrade(self.db, MIGRATIONS) connection = sqlite3.connect(self.db) self.assertIsNone(connection.execute( "SELECT creator FROM asset WHERE id=1" ).fetchone()[0]) self.assertEqual(connection.execute( "SELECT tag,source FROM asset_tag WHERE asset_id=1" ).fetchall(), [("原创", "name")]) self.assertEqual(connection.execute( "SELECT e.canonical_name,ae.source FROM asset_entity ae " "JOIN entity e ON e.id=ae.entity_id WHERE ae.asset_id=1" ).fetchall(), [("原创", "name")]) connection.close() def test_folder_attribution_corrections_are_narrow_and_evidence_backed(self): base_migrations = self.root / "base-migrations" base_migrations.mkdir() for path in sorted(MIGRATIONS.glob("*.sql")): if not path.name.startswith("0010_"): shutil.copyfile(path, base_migrations / path.name) sqlite3.connect(self.db).close() upgrade(self.db, base_migrations) connection = sqlite3.connect(self.db) connection.executescript(""" INSERT INTO asset(id,location,path,name,medium,creator) VALUES(1,'115','B:\\云下载\\足交仙人\\feet of Suzyq (1).mp4','one.mp4','video','足交仙人'), (2,'115','B:\\MVP\\捅主任\\TokyoDolls\\32.mp4','32.mp4','video','捅主任'), (3,'115','B:\\创作者\\捅主任\\real.mp4','real.mp4','video','捅主任'); INSERT INTO entity(id,kind,canonical_name,normalized_name,created_at,updated_at) VALUES(101,'creator','足交仙人','足交仙人','now','now'), (102,'creator','捅主任','捅主任','now','now'); INSERT INTO asset_entity(asset_id,entity_id,role,source,confidence) VALUES(1,101,'creator','legacy:asset',0.8), (2,102,'creator','legacy:asset',0.8), (3,102,'creator','legacy:asset',0.8); """) connection.close() upgrade(self.db, MIGRATIONS) connection = sqlite3.connect(self.db) self.assertEqual(connection.execute( "SELECT id,creator FROM asset ORDER BY id" ).fetchall(), [(1, "suzuq"), (2, None), (3, "捅主任")]) relations = connection.execute( "SELECT ae.asset_id,e.canonical_name,ae.source,ae.confidence " "FROM asset_entity ae JOIN entity e ON e.id=ae.entity_id " "WHERE e.kind='creator' ORDER BY ae.asset_id" ).fetchall() self.assertEqual(relations, [(1, "suzuq", "user:watermark", 1.0), (3, "捅主任", "legacy:asset", 0.8)]) self.assertEqual(connection.execute( "SELECT alias,source,confidence FROM entity_alias " "WHERE entity_id=(SELECT id FROM entity WHERE normalized_name='suzuq')" ).fetchall(), [("Suzyq", "filename", 0.95)]) connection.close() def test_fts_tracks_asset_entity_alias_and_search_term_changes(self): sqlite3.connect(self.db).close() upgrade(self.db, MIGRATIONS) connection = sqlite3.connect(self.db) connection.executescript(""" INSERT INTO asset(id,location,path,name,medium,code) VALUES(1,'local','one.mp4','Prestige sample.mp4','video','ABW-001'); INSERT INTO entity(id,kind,canonical_name,normalized_name,created_at,updated_at) VALUES(1001,'performer','高桥千凛','高桥千凛','now','now'); INSERT INTO asset_entity(asset_id,entity_id,role,source,confidence) VALUES(1,1001,'performer','test',1.0); INSERT INTO entity_alias(entity_id,alias,normalized_alias,source,confidence) VALUES(1001,'Takachi Chisato','takachi chisato','test',1.0); INSERT INTO entity_search_term(entity_id,term,purpose,source,created_at) VALUES(1001,'private release keyword','source_lookup','test','now'); """) for query in ('"Prestige"', '"高桥千"', '"Takachi"', '"release keyword"'): self.assertEqual(connection.execute( "SELECT asset_id FROM asset_search WHERE asset_search MATCH ?", (query,) ).fetchall(), [(1,)]) connection.execute("UPDATE asset SET name='Renamed title.mp4' WHERE id=1") self.assertEqual(connection.execute( "SELECT asset_id FROM asset_search WHERE asset_search MATCH ?", ('"Renamed"',) ).fetchall(), [(1,)]) self.assertEqual(connection.execute( "SELECT asset_id FROM asset_search WHERE asset_search MATCH ?", ('"Prestige"',) ).fetchall(), []) connection.execute( "UPDATE asset SET catalog_title='Official catalog title'," "original_title='Original catalog title' WHERE id=1" ) for query in ('"Official catalog"', '"Original catalog"'): self.assertEqual(connection.execute( "SELECT asset_id FROM asset_search WHERE asset_search MATCH ?", (query,) ).fetchall(), [(1,)]) connection.close() def test_rule34_source_case_duplicates_merge_without_losing_item_state(self): base_migrations = self.root / "base-migrations" base_migrations.mkdir() for path in sorted(MIGRATIONS.glob("*.sql")): # 历史基线停在 0037，follow_playback 建表与重建随 0020/0021 统一应用。 # 0038 起的版本号触发器依赖这些表，留在后续的完整迁移链中。 if path.name < "0038_" and not path.name.startswith(("0020_", "0021_", "0022_", "0024_")): shutil.copyfile(path, base_migrations / path.name) sqlite3.connect(self.db).close() upgrade(self.db, base_migrations) connection = sqlite3.connect(self.db) connection.executescript(""" INSERT INTO asset(id,location,path,name,medium) VALUES(1,'online','https://example.test/1','saved','video'); INSERT INTO follow_source( id,provider,ref,label,url,semantics,metadata_json,created_at,updated_at ) VALUES (10,'rule34xxx','lazyprocrastinator','lazyprocrastinator', 'https://rule34.xxx/?tags=lazyprocrastinator','work','{}','1','1'), (11,'rule34xxx','LazyProcrastinator','LazyProcrastinator', 'https://rule34.xxx/?tags=LazyProcrastinator','work','{}','2','2'); INSERT INTO follow_item( source_id,external_id,title,published_precision,release_key,variant_kind, status,asset_id,metadata_json,first_seen_at,last_seen_at ) VALUES (10,'same','same','exact','same','main','ignored',NULL,'{}','1','2'), (10,'lower-only','lower','exact','lower','main','new',NULL,'{}','1','2'), (11,'same','same','exact','same','main','saved',1,'{}','2','3'), (11,'upper-only','upper','exact','upper','main','seen',NULL,'{}','2','3'); """) connection.close() upgrade(self.db, MIGRATIONS) connection = sqlite3.connect(self.db) sources = connection.execute( "SELECT id,ref,url FROM follow_source WHERE provider='rule34xxx'" ).fetchall() items = connection.execute( "SELECT external_id,status,asset_id FROM follow_item ORDER BY external_id" ).fetchall() connection.close() self.assertEqual(sources, [(10, "lazyprocrastinator", "https://rule34.xxx/?tags=lazyprocrastinator")]) self.assertEqual(items, [ ("lower-only", "new", None), ("same", "saved", 1), ("upper-only", "seen", None), ]) FK_SEED_SQL = r""" INSERT INTO profile(id,user_id,name,is_default,settings_json,created_at,updated_at) VALUES('p2','local','Second',0,'{}','1','1'); INSERT INTO asset(id,location,path,name,medium,disposal) VALUES (1,'local','R:\Media\alpha.mp4','Alpha title.mp4','video',NULL), (2,'local','R:\Media\beta.mp4','Beta title.mp4','video','trash'); INSERT INTO asset_tag(asset_id,tag,confidence,source) VALUES (1,'巨乳',1.0,'name'),(1,'制服',0.8,'vision'),(2,'巨乳',1.0,'r18'); INSERT INTO media_binding( asset_id,backend,external_id,priority,metadata_json,last_synced_at) VALUES(1,'stash','101',100,'{}','1'); INSERT INTO activity_event(id,asset_id,profile_id,kind,occurred_at,value_json,source) VALUES(1,1,'local-default','play','1','{}','web'), (2,2,'p2','play','2','{}','web'); INSERT INTO asset_tag_preference(profile_id,asset_id,normalized_tag,hidden,updated_at) VALUES('local-default',1,'制服',1,'1'),('p2',2,'巨乳',1,'1'); INSERT INTO watch_queue(profile_id,asset_id,added_at,source) VALUES('local-default',1,'1','web'); INSERT INTO asset_preference(profile_id,asset_id,liked,updated_at) VALUES('p2',1,1,'1'); INSERT INTO asset_quality_goal(profile_id,asset_id,wanted,updated_at) VALUES('p2',1,1,'1'); INSERT INTO playlist(id,profile_id,name,source_kind,source_seed_asset_id, current_asset_id,created_at,updated_at) VALUES(1,'local-default','Mix','mix',1,1,'1','1'); INSERT INTO follow_source( id,provider,ref,label,url,semantics,metadata_json,created_at,updated_at) VALUES(10,'rule34xxx','someone','someone', 'https://rule34.xxx/?tags=someone','work','{}','1','1'); INSERT INTO follow_item( id,source_id,external_id,title,published_precision,release_key,variant_kind, status,asset_id,metadata_json,first_seen_at,last_seen_at) VALUES(100,10,'x','X','exact','x','main','new',NULL,'{}','1','1'); INSERT INTO follow_playback( follow_item_id,profile_id,play_count,play_seconds,max_reached,last_played) VALUES(100,'local-default',2,10,0.5,1.0),(100,'p2',1,5,0.2,2.0); """ # 0024 声明的删除行为。profile.user_id 是唯一保留 NO ACTION 的外键：即时外键下 # NO ACTION 已等于 RESTRICT，而重建 profile 会让 DROP TABLE 的隐式 DELETE 真的 # 级联删掉 profile 私有状态。判据写在 0024 的头部注释里。 EXPECTED_DELETE_RULES = { ("asset_tag", "asset_id"): ("asset", "CASCADE"), ("media_binding", "asset_id"): ("asset", "CASCADE"), ("activity_event", "asset_id"): ("asset", "CASCADE"), ("activity_event", "profile_id"): ("profile", "SET NULL"), ("asset_tag_preference", "profile_id"): ("profile", "CASCADE"), ("follow_playback", "profile_id"): ("profile", "CASCADE"), ("asset_subtitle", "asset_id"): ("asset", "CASCADE"), } NO_ACTION_ALLOWED = {("profile", "user_id")} class AssetFieldOwnerColumnTests(unittest.TestCase): """0029：真相字段的归属列与乐观并发计数。""" def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.root = Path(self.tmp.name).resolve() self.db = self.root / "ledger.db" sqlite3.connect(self.db).close() upgrade(self.db, MIGRATIONS) def test_asset_gains_field_owners_and_a_revision_counter(self): connection = sqlite3.connect(self.db) self.addCleanup(connection.close) columns = {row[1]: (row[2], row[3], row[4]) for row in connection.execute("PRAGMA table_info(asset)")} self.assertEqual(columns["field_owners"][0], "TEXT") declaration, not_null, default = columns["mutation_revision"] self.assertEqual(declaration, "INTEGER") self.assertEqual((not_null, default), (1, "0")) def test_existing_rows_start_unowned_at_revision_zero(self): """归属不回填：谁写的答不出来时，留空比猜一个写入者诚实。""" connection = sqlite3.connect(self.db) self.addCleanup(connection.close) connection.execute( "INSERT INTO asset(id,location,path,name,medium,studio) " "VALUES(1,'local','R:\\\\a.mp4','a.mp4','video','某厂牌')") row = connection.execute( "SELECT field_owners,mutation_revision FROM asset WHERE id=1").fetchone() self.assertEqual(row, (None, 0)) def test_sqlite_has_the_json_functions_the_owner_map_needs(self): """`json_extract` / `json_patch` 是归属写入唯一的实现手段，先确认它们在。""" connection = sqlite3.connect(self.db) self.addCleanup(connection.close) merged = connection.execute( "SELECT json_patch('{\"a\":\"scan:filename\"}','{\"b\":\"user:manual\"}')" ).fetchone()[0] self.assertEqual(json.loads(merged), {"a": "scan:filename", "b": "user:manual"}) self.assertEqual(connection.execute( "SELECT json_extract('{\"studio\":\"user:manual\"}','$.\"studio\"')" ).fetchone()[0], "user:manual") class ForeignKeyDeleteRuleTests(unittest.TestCase): """0024：补齐缺失的 ON DELETE，重建时不丢行、不丢索引与触发器。""" def setUp(self): self.tmp = tempfile.TemporaryDirectory() # 用 addCleanup 而不是 tearDown：清理按后进先出跑，连接一定先于临时目录关闭。 self.addCleanup(self.tmp.cleanup) self.root = Path(self.tmp.name).resolve() self.db = self.root / "ledger.db" def _migrations_below(self, upper: str): """只含版本号小于 `upper` 的迁移目录。 按上界取，不按「排除 0024」取补集：补集会把后来新增的每一个版本都算进这份 「到 0023 为止」的目录，而 0025 与 0024 无关，照样装得上。这一组用例盯的是 0024 那次重建，多装一个版本就会把它新增的表和索引算成 0024 的账。 """ directory = self.root / f"below-{upper}" directory.mkdir() for path in sorted(MIGRATIONS.glob("*.sql")): if path.name < upper: shutil.copyfile(path, directory / path.name) return directory def _upgrade_through_0023(self): sqlite3.connect(self.db).close() upgrade(self.db, self._migrations_below("0024")) def _seed(self): connection = sqlite3.connect(self.db) connection.executescript(FK_SEED_SQL) connection.commit() connection.close() def _open(self, foreign_keys=False): connection = sqlite3.connect(self.db) connection.isolation_level = None # 断言失败也要先关连接，否则 Windows 上 tearDown 删不掉临时库。 self.addCleanup(connection.close) if foreign_keys: connection.execute("PRAGMA foreign_keys=ON") return connection @staticmethod def _counts(connection): # asset_search 的影子表另行核对；schema_migration 本次必然 +1。 names = [row[0] for row in connection.execute( "SELECT name FROM sqlite_schema WHERE type='table' " "AND name NOT LIKE 'sqlite_%' AND name NOT LIKE 'asset_search%' " "AND name<>'schema_migration' ORDER BY name" )] return {name: connection.execute( f'SELECT count(*) FROM "{name}"').fetchone()[0] for name in names} @staticmethod def _objects(connection, kind): return {row[0] for row in connection.execute( "SELECT name FROM sqlite_schema WHERE type=? AND sql IS NOT NULL", (kind,) )} @staticmethod def _search_rows(connection): return sorted(row[0] for row in connection.execute( "SELECT asset_id FROM asset_search")) @staticmethod def _delete_rules(connection): rules = {} tables = [row[0] for row in connection.execute( "SELECT name FROM sqlite_schema WHERE type='table' " "AND name NOT LIKE 'sqlite_%' ORDER BY name" )] for table in tables: for row in connection.execute(f'PRAGMA foreign_key_list("{table}")'): rules[(table, row[3])] = (row[2], row[6]) return rules def test_upgrade_keeps_rows_indexes_triggers_and_search(self): self._upgrade_through_0023() self._seed() connection = self._open() before_counts = self._counts(connection) before_indexes = self._objects(connection, "index") before_triggers = self._objects(connection, "trigger") before_views = self._objects(connection, "view") before_search = self._search_rows(connection) connection.close() self.assertEqual(before_counts["asset_tag"], 3) self.assertEqual(before_counts["follow_playback"], 2) self.assertEqual( [m.version for m in upgrade(self.db, self._migrations_below("0025"))], ["0024"]) connection = self._open() self.assertEqual(self._counts(connection), before_counts) self.assertEqual(self._objects(connection, "trigger"), before_triggers) self.assertEqual(self._objects(connection, "view"), before_views) after_indexes = self._objects(connection, "index") self.assertEqual(before_indexes - after_indexes, set()) self.assertEqual(after_indexes - before_indexes, {"idx_asset_disposal_id", "idx_asset_tag_source_asset"}) self.assertEqual(self._search_rows(connection), before_search) self.assertEqual(connection.execute( "SELECT asset_id FROM asset_search WHERE asset_search MATCH ?", ('"Alpha"',) ).fetchall(), [(1,)]) self.assertEqual(connection.execute("PRAGMA foreign_key_check").fetchall(), []) self.assertEqual(connection.execute("PRAGMA integrity_check").fetchone()[0], "ok") connection.close() def test_every_foreign_key_declares_a_delete_rule(self): self._upgrade_through_0023() upgrade(self.db, MIGRATIONS) connection = self._open() rules = self._delete_rules(connection) connection.close() for key, expected in EXPECTED_DELETE_RULES.items(): self.assertEqual(rules.get(key), expected, key) undeclared = {key for key, value in rules.items() if value[1] == "NO ACTION"} self.assertEqual(undeclared, NO_ACTION_ALLOWED) def test_deleting_asset_cascades_to_its_dependent_rows(self): self._upgrade_through_0023() self._seed() upgrade(self.db, MIGRATIONS) connection = self._open(foreign_keys=True) connection.execute("DELETE FROM asset WHERE id=1") self.assertEqual(connection.execute( "SELECT asset_id,tag FROM asset_tag ORDER BY asset_id,tag").fetchall(), [(2, "巨乳")]) for table in ("media_binding", "watch_queue", "asset_preference", "asset_quality_goal"): self.assertEqual(connection.execute( f"SELECT count(*) FROM {table} WHERE asset_id=1").fetchone()[0], 0, table) self.assertEqual(connection.execute( "SELECT id FROM activity_event ORDER BY id").fetchall(), [(2,)]) self.assertEqual(connection.execute( "SELECT profile_id,asset_id FROM asset_tag_preference").fetchall(), [("p2", 2)]) self.assertEqual(connection.execute( "SELECT source_seed_asset_id,current_asset_id FROM playlist").fetchall(), [(None, None)]) self.assertEqual(self._search_rows(connection), [2]) self.assertEqual(connection.execute("PRAGMA foreign_key_check").fetchall(), []) connection.close() def test_deleting_profile_keeps_activity_history_without_owner(self): self._upgrade_through_0023() self._seed() upgrade(self.db, MIGRATIONS) connection = self._open(foreign_keys=True) connection.execute("DELETE FROM profile WHERE id='p2'") self.assertEqual(connection.execute( "SELECT follow_item_id,profile_id FROM follow_playback").fetchall(), [(100, "local-default")]) self.assertEqual(connection.execute( "SELECT profile_id,asset_id FROM asset_tag_preference").fetchall(), [("local-default", 1)]) # 行为历史保留，只丢归属。 self.assertEqual(connection.execute( "SELECT id,asset_id,profile_id FROM activity_event ORDER BY id").fetchall(), [(1, 1, "local-default"), (2, 2, None)]) self.assertEqual(connection.execute("PRAGMA foreign_key_check").fetchall(), []) connection.close() def test_deleting_app_user_with_profiles_is_still_refused(self): self._upgrade_through_0023() self._seed() upgrade(self.db, MIGRATIONS) connection = self._open(foreign_keys=True) with self.assertRaises(sqlite3.IntegrityError): connection.execute("DELETE FROM app_user WHERE id='local'") self.assertEqual(connection.execute( "SELECT count(*) FROM app_user WHERE id='local'").fetchone()[0], 1) connection.close() def test_orphan_rows_abort_the_migration_instead_of_being_dropped(self): self._upgrade_through_0023() self._seed() connection = self._open() connection.execute( "INSERT INTO asset_tag(asset_id,tag,confidence,source) " "VALUES(999,'orphan',1.0,'name')") connection.close() with self.assertRaises(sqlite3.IntegrityError): upgrade(self.db, MIGRATIONS) connection = self._open() self.assertEqual(connection.execute( "SELECT max(version) FROM schema_migration").fetchone()[0], "0023") self.assertEqual(connection.execute( "SELECT count(*) FROM asset_tag WHERE asset_id=999").fetchone()[0], 1) self.assertEqual(connection.execute( "SELECT count(*) FROM asset_tag").fetchone()[0], 4) self.assertEqual(self._delete_rules(connection).get(("asset_tag", "asset_id")), None) connection.close() def test_new_indexes_are_chosen_by_their_justifying_queries(self): self._upgrade_through_0023() self._seed() upgrade(self.db, MIGRATIONS) connection = self._open() trash = " ".join(row[3] for row in connection.execute( "EXPLAIN QUERY PLAN SELECT count(*) FROM asset a WHERE a.disposal='trash'")) self.assertIn("idx_asset_disposal_id", trash) sources = " ".join(row[3] for row in connection.execute( "EXPLAIN QUERY PLAN SELECT source,count(*),count(DISTINCT asset_id) " "FROM asset_tag GROUP BY source ORDER BY count(*) DESC")) self.assertIn("idx_asset_tag_source_asset", sources) connection.close() class AssetSubtitleTableTests(unittest.TestCase): """0027：字幕 sidecar 表的形状由 schema 自己守住。""" def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.root = Path(self.tmp.name).resolve() self.db = self.root / "ledger.db" sqlite3.connect(self.db).close() upgrade(self.db, MIGRATIONS) self.connection = sqlite3.connect(self.db) self.connection.isolation_level = None self.addCleanup(self.connection.close) self.connection.execute( "INSERT INTO asset(id,location,path,name,medium,first_seen,last_seen) " r"VALUES(1,'local','R:\media\x\a.mp4','a.mp4','video','2026-09-01','2026-09-01')") def _insert(self, **overrides): row = {"asset_id": 1, "location": "local", "path": r"R:\media\x\a.srt", "name": "a.srt", "language": "zh-Hans", "format": "srt", "size": 40, "mtime": "2026-09-01", "pairing": "exact", "first_seen": "2026-09-01", "last_seen": "2026-09-01"} row.update(overrides) columns = ",".join(row) self.connection.execute( f"INSERT INTO asset_subtitle({columns}) " f"VALUES({','.join('?' * len(row))})", tuple(row.values())) def test_the_same_path_can_only_be_registered_once_per_location(self): self._insert() with self.assertRaises(sqlite3.IntegrityError): self._insert(name="别名.srt") def test_an_unpaired_row_must_declare_itself_orphan(self): self._insert(asset_id=None, pairing="orphan") with self.assertRaises(sqlite3.IntegrityError): self._insert(asset_id=None, pairing="exact", path=r"R:\media\x\b.srt") def test_an_unknown_pairing_name_is_refused(self): with self.assertRaises(sqlite3.IntegrityError): self._insert(pairing="差不多") def test_deleting_the_asset_takes_its_subtitle_rows(self): self._insert() self.connection.execute("PRAGMA foreign_keys=ON") self.connection.execute("DELETE FROM asset WHERE id=1") self.assertEqual(self.connection.execute( "SELECT count(*) FROM asset_subtitle").fetchone()[0], 0) self.assertEqual( self.connection.execute("PRAGMA foreign_key_check").fetchall(), []) def test_the_index_carries_the_listing_query(self): plan = " ".join(row[3] for row in self.connection.execute( "EXPLAIN QUERY PLAN SELECT id,name FROM asset_subtitle " "WHERE asset_id=1 ORDER BY name COLLATE NOCASE, id")) self.assertIn("idx_asset_subtitle_asset", plan) class TaskRunTableTests(unittest.TestCase): """0028：任务中心那张表的约束由数据库守，不靠服务层自觉。""" def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.db = Path(self.tmp.name).resolve() / "ledger.db" sqlite3.connect(self.db).close() upgrade(self.db, MIGRATIONS) self.connection = sqlite3.connect(self.db) self.addCleanup(self.connection.close) def _insert(self, **fields): fields.setdefault("task_key", "demo") fields.setdefault("trigger", "manual") fields.setdefault("status", "running") columns = ",".join(fields) marks = ",".join("?" * len(fields)) cursor = self.connection.execute( f"INSERT INTO task_run({columns}) VALUES({marks})", list(fields.values())) self.connection.commit() return cursor.lastrowid def test_one_active_run_per_mutex_key(self): self._insert(mutex_key="follow-check", started_at="2026-09-11T00:00:00.000Z") with self.assertRaises(sqlite3.IntegrityError): self._insert(mutex_key="follow-check", started_at="2026-09-11T00:01:00.000Z") def test_finished_runs_release_the_mutex_key_without_losing_it(self): """终态行不占索引位，但 `mutex_key` 留着——「上次是谁挡的」还查得到。""" for _ in range(3): self._insert(mutex_key="follow-check", status="succeeded", finished_at="2026-09-11T00:00:00.000Z") self._insert(mutex_key="follow-check", started_at="2026-09-11T00:02:00.000Z") kept = self.connection.execute( "SELECT count(*) FROM task_run WHERE mutex_key='follow-check'").fetchone() self.assertEqual(kept[0], 4) def test_runs_without_a_mutex_key_never_block_each_other(self): for _ in range(3): self._insert(started_at="2026-09-11T00:00:00.000Z") self.assertEqual(self.connection.execute( "SELECT count(*) FROM task_run WHERE status='running'").fetchone()[0], 3) def test_terminal_status_and_finished_at_agree(self): with self.assertRaises(sqlite3.IntegrityError): self._insert(status="succeeded") with self.assertRaises(sqlite3.IntegrityError): self._insert(status="running", finished_at="2026-09-11T00:00:00.000Z") def test_unknown_status_and_trigger_are_rejected(self): with self.assertRaises(sqlite3.IntegrityError): self._insert(status="done", finished_at="2026-09-11T00:00:00.000Z") with self.assertRaises(sqlite3.IntegrityError): self._insert(trigger="webhook") def test_recent_runs_of_one_task_use_their_index(self): plan_rows = " ".join(row[3] for row in self.connection.execute( "EXPLAIN QUERY PLAN SELECT id FROM task_run WHERE task_key='demo' " "ORDER BY id DESC LIMIT 20")) self.assertIn("idx_task_run_key_recent", plan_rows) class ExternalRefUniquenessTests(unittest.TestCase): """0032：一位实体在一个站点可以有多个页面，一个页面仍只属于一位实体。""" def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.db = Path(self.tmp.name).resolve() / "ledger.db" sqlite3.connect(self.db).close() upgrade(self.db, MIGRATIONS) self.connection = sqlite3.connect(self.db) self.addCleanup(self.connection.close) for entity_id, name in ((10, "釈アリス"), (11, "天川そら")): self.connection.execute( "INSERT INTO entity(id,kind,canonical_name,normalized_name,created_at,updated_at)" " VALUES(?,'performer',?,?,'t','t')", (entity_id, name, name)) self.connection.commit() def _ref(self, entity_id: int, external_id: str): self.connection.execute( "INSERT INTO entity_external_ref(entity_id,provider,external_kind,external_id)" " VALUES(?,'javdb','performer',?)", (entity_id, external_id)) self.connection.commit() def test_one_performer_can_hold_two_pages_on_the_same_site(self): self._ref(10, "d45k9") self._ref(10, "ZX5z7") self.assertEqual(self.connection.execute( "SELECT count(*) FROM entity_external_ref WHERE entity_id=10").fetchone()[0], 2) def test_the_same_page_cannot_be_claimed_twice(self): self._ref(10, "d45k9") with self.assertRaises(sqlite3.IntegrityError): self._ref(11, "d45k9") with self.assertRaises(sqlite3.IntegrityError): self._ref(10, "d45k9") def test_deleting_the_performer_takes_every_page_with_it(self): self._ref(10, "d45k9") self._ref(10, "ZX5z7") self.connection.execute("PRAGMA foreign_keys=ON") self.connection.execute("DELETE FROM entity WHERE id=10") self.connection.commit() self.assertEqual(self.connection.execute( "SELECT count(*) FROM entity_external_ref").fetchone()[0], 0) class OrphanSourceRowsTests(unittest.TestCase): """0039：删来源时没跟着走的子表行被清掉，还有主人的原样留下。""" def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.root = Path(self.tmp.name).resolve() self.db = self.root / "ledger.db" below = self.root / "below-0039" below.mkdir() for path in sorted(MIGRATIONS.glob("*.sql")): if path.name < "0039": shutil.copyfile(path, below / path.name) sqlite3.connect(self.db).close() upgrade(self.db, below) connection = sqlite3.connect(self.db) # 和服务的连接一样不开外键，才造得出生产上那种孤儿行。 connection.executescript(""" INSERT INTO follow_source(id,provider,ref,label,url,created_at,updated_at) VALUES(1,'rule34xxx','kept','kept','https://x.test/kept','t','t'); INSERT INTO follow_item(id,source_id,external_id,title,release_key,first_seen_at,last_seen_at) VALUES(10,1,'a','kept','a','t','t'), (11,106,'b','orphan','b','t','t'); INSERT INTO follow_playback(follow_item_id,play_count) VALUES(10,1),(11,1),(12,1); INSERT INTO feed_source(id,kind,url,created_at) VALUES(1,'javdb_actor','https://j.test/1','t'); INSERT INTO feed_item(id,source_id,item_key,first_seen) VALUES(20,1,'k','t'),(21,9,'k','t'); INSERT INTO feed_discovery(id,code,source_id,discovered_at) VALUES(30,'ABC-001',1,'t'),(31,'ABC-002',9,'t'); """) connection.commit() connection.close() def test_rows_of_removed_sources_are_cleared(self): upgrade(self.db, MIGRATIONS) connection = sqlite3.connect(self.db) self.addCleanup(connection.close) self.assertEqual(connection.execute("PRAGMA foreign_key_check").fetchall(), []) self.assertEqual([row[0] for row in connection.execute("SELECT id FROM follow_item")], [10]) self.assertEqual([row[0] for row in connection.execute( "SELECT follow_item_id FROM follow_playback")], [10]) self.assertEqual([row[0] for row in connection.execute("SELECT id FROM feed_item")], [20]) # 发现过的作品是「这个番号还没入库」，不随订阅消失，只断开来源。 self.assertEqual(connection.execute( "SELECT id,source_id FROM feed_discovery ORDER BY id").fetchall(), [(30, 1), (31, None)]) if __name__ == "__main__": unittest.main()