tests/test_seed_followup.py
Rama master · réplica de lectura del repositorio original.
"""导入实体种子后继：扫描结算时按版本与新实体决定派不派，跑一次把包补进账本并留下复核产物（ADR-0075）。"""
import tempfile
import unittest
from pathlib import Path
from types import SimpleNamespace
from unittest import mock
from peach import seed_followup, seed_pack, seed_validation
from peach.entities import normalize_entity_name
from peach.repository import LedgerDatabase
from peach.review_csv import read_rows
from peach.task_runs import TaskRunStore
from support.ledger import fresh_ledger
VERSION = "2026-09-25"
STAMP = "2026-09-25T00:00:00Z"
PACK = {"format": 1, "version": VERSION, "source": "auto:seed", "counts": {}, "entities": [
{"kind": "performer", "name": "天川そら", "aliases": [{"alias": "春山心愛", "source": "r18:performer"}],
"refs": [], "links": [{"kind": "social", "label": "X", "url": "https://x.com/amakawa_sora_"}],
"agency": {"name": "STARTUP", "source": "minnano-av:所属事務所"}},
{"kind": "studio", "name": "PRESTIGE PREMIUM", "aliases": [], "refs": [], "links": [],
"maker": {"name": "Prestige", "source": "review:label-maker-20260923.csv"}},
]}
KEY = seed_followup.followup_key(VERSION, seed_validation.digest(PACK))
class Case(unittest.TestCase):
def setUp(self):
temporary = tempfile.TemporaryDirectory()
self.addCleanup(temporary.cleanup)
self.root = Path(temporary.name).resolve()
self.database = LedgerDatabase(fresh_ledger(self.root))
self.store = TaskRunStore(self.database, host="test-host")
self.pack = self.root / "entities.json"
self.pack.write_text(seed_pack.dump(PACK), encoding="utf-8")
patcher = mock.patch.object(seed_pack, "DEFAULT_PACK", self.pack)
patcher.start()
self.addCleanup(patcher.stop)
self.candidates = self.root / "candidates"
self.candidates.mkdir()
self.busts = 0
self.contract = SimpleNamespace(database=self.database, candidate_root=self.candidates,
cache_bust=self._bust)
def _bust(self) -> None:
self.busts += 1
def add(self, kind: str, name: str) -> int:
with self.database.write_transaction() as connection:
connection.execute(
"INSERT INTO entity(kind,canonical_name,normalized_name,metadata_json,created_at,updated_at)"
" VALUES(?,?,?,'{}',?,?)", (kind, name, normalize_entity_name(name), STAMP, STAMP))
return int(connection.execute("SELECT last_insert_rowid()").fetchone()[0])
def plan(self, since_entity_id: int):
with self.database.read_connection() as connection:
return seed_followup.plan(connection, since_entity_id=since_entity_id)
def finished_run(self, key: str) -> None:
"""任务记录里留一条这个 key 的成功后继：父任务派它，通道认领，跑完。"""
parent = self.store.start("library-processing", trigger="manual")
self.store.enqueue_followups(parent.id, [(key, seed_followup.TASK_KEY, "x")])
child = self.store.claim_followup([seed_followup.TASK_KEY])
self.store.finish(child.id, "succeeded", summary={})
self.store.finish(parent.id, "succeeded", summary={})
class PlanTests(Case):
def test_content_hash_changes_the_followup_key(self):
first = self.plan(since_entity_id=10 ** 9)[0].key
self.finished_run(first)
modified = seed_pack.load(self.pack)
modified["entities"][0]["links"][0]["url"] = "https://x.com/verified_handle"
self.pack.write_text(seed_pack.dump(modified), encoding="utf-8")
second = self.plan(since_entity_id=10 ** 9)[0].key
self.assertNotEqual(first, second)
def test_an_unchanged_pack_is_parsed_once_while_history_is_checked_every_round(self):
"""升级补修每分钟问一次：包没变就不再整包解析、校验；导入历史照旧每轮核对。"""
with mock.patch.object(seed_validation, "validate", wraps=seed_validation.validate) as validated:
first = self.plan(since_entity_id=10 ** 9)
second = self.plan(since_entity_id=10 ** 9)
self.assertEqual(first, second)
self.assertEqual(validated.call_count, 1)
with self.database.write_transaction() as connection:
connection.execute(
"INSERT INTO seed_pack_history(version,revision,content_sha256,withdrawals_json,imported_at)"
" VALUES(?,0,'another-content','[]',?)", (VERSION, STAMP))
with self.assertRaisesRegex(ValueError, "同版本种子内容与已导入历史不同"):
self.plan(since_entity_id=10 ** 9)
def test_without_a_pack_nothing_is_planned(self):
with self.database.read_connection() as connection:
self.assertEqual(seed_followup.plan(connection, since_entity_id=0,
pack_path=self.root / "missing.json"), [])
def test_a_version_never_landed_is_planned_even_without_new_entities(self):
found = self.plan(since_entity_id=10 ** 9)
self.assertEqual([(item.key, item.task_key, item.label) for item in found],
[(KEY, "seed-import", f"初始资料导入：{VERSION}")])
def test_a_landed_version_is_planned_again_only_when_new_entities_appear(self):
self.finished_run(KEY)
self.assertEqual(self.plan(since_entity_id=10 ** 9), [], "导过、又没有新实体，不再派")
her = self.add("performer", "天川そら")
self.assertEqual(len(self.plan(since_entity_id=her - 1)), 1, "这一轮登记了新女优，再派一次给她补")
self.add("tag", "标签不算")
self.assertEqual(self.plan(since_entity_id=her), [], "只有标签这类包里没有的种类，不派")
class RunTests(Case):
def chinese_pack(self):
pack = {**PACK, "entities": [{"kind": "performer", "name": "天使萌",
"aliases": [{"alias": "天使もえ", "source": "avdb-actor-mapping@test"}],
"refs": [], "links": []}]}
self.pack.write_text(seed_pack.dump(pack), encoding="utf-8")
return seed_followup.followup_key(VERSION, seed_validation.digest(pack))
def test_chinese_names_are_backed_up_and_reported(self):
import json
import sqlite3
from contextlib import closing
key = self.chinese_pack()
person = self.add("performer", "天使もえ")
summary = seed_followup.run(self.contract, key, None)
backups = list(self.root.rglob("ledger.pre-seed-names-*.db"))
self.assertEqual(len(backups), 1)
with closing(sqlite3.connect(backups[0])) as connection:
self.assertEqual(connection.execute("SELECT canonical_name FROM entity WHERE id=?", (person,)).fetchone()[0], "天使もえ")
with self.database.read_connection() as connection:
row = connection.execute("SELECT canonical_name,metadata_json FROM entity WHERE id=?", (person,)).fetchone()
self.assertEqual(row[0], "天使萌")
self.assertEqual(json.loads(row[1])["name_preference"]["previous_name"], "天使もえ")
self.assertEqual(summary["names"], 1)
self.assertIn("中文名 1", summary["outcome"])
def test_failed_name_backup_leaves_the_ledger_unchanged(self):
key = self.chinese_pack()
person = self.add("performer", "天使もえ")
with mock.patch("peach.migrations.sqlite_backup", side_effect=OSError("备份目录不可写")):
with self.assertRaisesRegex(OSError, "备份目录不可写"):
seed_followup.run(self.contract, key, None)
with self.database.read_connection() as connection:
self.assertEqual(connection.execute("SELECT canonical_name FROM entity WHERE id=?", (person,)).fetchone()[0], "天使もえ")
self.assertEqual(connection.execute("SELECT count(*) FROM entity_alias").fetchone()[0], 0)
def test_queued_content_change_is_rejected_without_writing(self):
self.add("performer", "天川そら")
modified = seed_pack.load(self.pack)
modified["entities"][0]["links"][0]["url"] = "https://x.com/verified_handle"
self.pack.write_text(seed_pack.dump(modified), encoding="utf-8")
with self.assertRaises(ValueError):
seed_followup.run(self.contract, KEY, None)
with self.database.read_connection() as connection:
self.assertEqual(connection.execute("SELECT count(*) FROM entity_alias").fetchone()[0], 0)
self.assertEqual(connection.execute("SELECT count(*) FROM seed_pack_history").fetchone()[0], 0)
def test_run_lands_the_pack_and_writes_the_review_file(self):
her, kmp = self.add("performer", "天川そら"), self.add("studio", "K M Produce")
self.add("agency", "STARTUP")
label = self.add("studio", "PRESTIGE PREMIUM")
self.add("studio", "Prestige")
with self.database.write_transaction() as connection:
connection.execute("INSERT INTO label_maker(label_id,maker_id,source,confidence,checked_at)"
" VALUES(?,?,?,1.0,?)", (label, kmp, "review:user", STAMP))
summary = seed_followup.run(self.contract, KEY, None)
self.assertEqual((summary["matched"], summary["aliases"], summary["links"], summary["memberships"],
summary["makers"], summary["conflicts"], summary["duplicates"]), (2, 1, 1, 1, 0, 1, 0))
self.assertEqual(summary["outcome"], "别名 1、链接 1、归属 1、不一致 1")
self.assertEqual(self.busts, 1, "动了数据库就清一次聚合缓存")
with self.database.read_connection() as connection:
self.assertEqual(connection.execute("SELECT maker_id FROM label_maker WHERE label_id=?",
(label,)).fetchone()[0], kmp, "人复核过的片商不被种子改")
self.assertEqual(connection.execute("SELECT count(*) FROM entity_alias WHERE entity_id=?",
(her,)).fetchone()[0], 1)
rows = read_rows(self.candidates / seed_followup.REVIEW_FILE)
self.assertEqual([(row["entity"], row["field"], row["ours"], row["theirs"]) for row in rows],
[("PRESTIGE PREMIUM", "maker", "K M Produce", "Prestige")])
again = seed_followup.run(self.contract, KEY, None)
self.assertEqual((again["aliases"], again["links"], again["memberships"], again["refreshed"]), (0, 0, 0, 0),
"同一包再跑一遍没有东西可补")
self.assertEqual(self.busts, 1, "没动数据库就不清缓存")
self.assertTrue(again["outcome"].startswith("不一致 1"), again["outcome"])
def test_two_local_entities_for_one_pack_item_become_a_duplicate_row(self):
self.add("performer", "天川そら")
self.add("performer", "春山心愛")
summary = seed_followup.run(self.contract, KEY, None)
self.assertEqual((summary["matched"], summary["duplicates"]), (0, 1))
rows = read_rows(self.candidates / seed_followup.REVIEW_FILE)
self.assertEqual([(row["field"], row["entity"], row["theirs"]) for row in rows],
[("identity", "天川そら、春山心愛", "天川そら")])
if __name__ == "__main__":
unittest.main()