tests/test_seed_pack.py
Rama master · réplica de lectura del repositorio original.
"""实体事实种子包：导出只带公开事实且逐字节稳定，导入只给已有实体填空、只换种子自己写过的行，
整批可撤（ADR-0073、ADR-0075）。
全程两本临时账本：一本当导出方，一本当另一台机器。
"""
import contextlib
import importlib.util
import io
import json
import sqlite3
import tempfile
import unittest
from pathlib import Path
from unittest import mock
from peach import seed_pack
from peach.entities import normalize_entity_name
from peach.performer_profiles import read_profile, write_profile
from support.ledger import fresh_ledger
ROOT = Path(__file__).resolve().parents[1]
STAMP = "2026-09-25T00:00:00Z"
VERSION = "2026-09-25"
BATCH = f"auto:seed@{VERSION}"
def load_script(name: str):
spec = importlib.util.spec_from_file_location(f"{name}_under_test", ROOT / "scripts" / f"{name}.py")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
def connect(path: Path) -> sqlite3.Connection:
connection = sqlite3.connect(path)
connection.row_factory = sqlite3.Row
connection.execute("PRAGMA foreign_keys=ON")
return connection
def entity(connection, kind: str, name: str) -> int:
connection.execute(
"INSERT INTO entity(kind,canonical_name,normalized_name,metadata_json,created_at,updated_at)"
" VALUES(?,?,?,?,?,?)",
(kind, name, normalize_entity_name(name), json.dumps({"raw_snapshot": r"C:\somewhere\x.json"}),
STAMP, STAMP))
return int(connection.execute("SELECT last_insert_rowid()").fetchone()[0])
def alias(connection, entity_id: int, name: str, source: str) -> None:
connection.execute("INSERT INTO entity_alias(entity_id,alias,normalized_alias,source,confidence)"
" VALUES(?,?,?,?,1.0)", (entity_id, name, normalize_entity_name(name), source))
def ref(connection, entity_id: int, provider: str, external_id: str, metadata: dict | None = None) -> None:
connection.execute("INSERT INTO entity_external_ref(entity_id,provider,external_kind,external_id,"
"metadata_json,last_synced_at) VALUES(?,?,'performer',?,?,?)",
(entity_id, provider, external_id, json.dumps(metadata or {}), STAMP))
def link(connection, entity_id: int, kind: str, url: str, *, sensitive: int = 0, metadata=None) -> None:
connection.execute("INSERT INTO entity_link(entity_id,link_kind,label,url,hostname,is_sensitive,"
"metadata_json,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?)",
(entity_id, kind, "label", url, "x.com", sensitive, json.dumps(metadata or {}),
STAMP, STAMP))
def membership(connection, member_id: int, agency_id: int, source: str) -> None:
connection.execute("INSERT INTO entity_membership(member_id,agency_id,source,confidence,checked_at)"
" VALUES(?,?,?,1.0,?)", (member_id, agency_id, source, STAMP))
def label_maker(connection, label_id: int, maker_id: int, source: str) -> None:
connection.execute("INSERT INTO label_maker(label_id,maker_id,source,confidence,checked_at)"
" VALUES(?,?,?,1.0,?)", (label_id, maker_id, source, STAMP))
PROFILE = {"kana": "あまかわそら", "birth_date": "1998-10-10", "height_cm": 164, "cup": "G",
"tags": ["巨乳", "美肌"], "raw": {"名前": "天川そら", "サイズ": "T164 / B87"}}
class SeedPackCase(unittest.TestCase):
def setUp(self):
# 目录先登记回收：清理按后进先出执行，两个连接关掉之后才删目录。
self.root = Path(self.enterContext(
tempfile.TemporaryDirectory(prefix="peach-seed-", ignore_cleanup_errors=True)))
self.source = connect(fresh_ledger(self.root, "source.db"))
self.target = connect(fresh_ledger(self.root, "target.db"))
self.addCleanup(self.source.close)
self.addCleanup(self.target.close)
def seeded_source(self) -> dict:
"""导出方账本：一位有全套事实的女优、她的事务所、一家带别名的厂牌、一个归属片商的 label、
一位只有名字的女优。"""
con = self.source
with con:
sora = entity(con, "performer", "天川そら")
agency = entity(con, "agency", "STARTUP")
studio = entity(con, "studio", "S1 NO.1 STYLE")
maker = entity(con, "studio", "Prestige")
label = entity(con, "studio", "PRESTIGE PREMIUM")
entity(con, "performer", "只有名字")
label_maker(con, label, maker, "review:label-maker-20260923.csv")
alias(con, sora, "春山心愛", "r18:performer")
alias(con, sora, "Amakawa Sora", "r18:performer")
alias(con, sora, "そら", "javdb-actor-page@javdb-20260912")
alias(con, sora, "二宮そら", "javdb-actor-page@javdb-20260912")
alias(con, sora, "二宮そら", "kanji-simplification")
alias(con, sora, "种子写的", BATCH)
ref(con, sora, "minnano-av", "295275")
ref(con, sora, "stash", "26")
ref(con, sora, "javdb", "seedref", {"source": "auto:seed", "batch": "auto:seed@old"})
link(con, sora, "social", "https://x.com/amakawa_sora_")
link(con, sora, "official", "https://www.Startup0701.net/")
link(con, sora, "social", "https://onlyfans.example/sora", sensitive=1)
link(con, sora, "catalog", "https://www.minnano-av.com/actress295275.html")
link(con, sora, "social", "https://x.com/seeded", metadata={"source": "auto:seed", "batch": "x"})
write_profile(con, sora, dict(PROFILE), source="auto:performer-profile@7",
source_url="https://www.minnano-av.com/actress295275.html", fetched_at=STAMP)
membership(con, sora, agency, "minnano-av:所属事務所")
alias(con, studio, "S1", "javbus:studio-localization")
link(con, studio, "official", "https://s1s1s1.com/")
return seed_pack.export_pack(con, version=VERSION)
class SeedPerformerNameTests(SeedPackCase):
def name_pack(self, *, revision=None):
with self.source:
person = entity(self.source, "performer", "天使萌")
alias(self.source, person, "天使もえ", "avdb-actor-mapping@test")
return seed_pack.export_pack(self.source, version=VERSION, revision=revision)
def test_verified_chinese_name_updates_the_entity_and_work_projection(self):
pack = self.name_pack()
with self.target:
person = entity(self.target, "performer", "天使もえ")
self.target.execute("INSERT INTO asset(id,location,path,name,medium) VALUES(1,'local','R:\\a.mp4','a','video')")
self.target.execute("INSERT INTO asset_entity(asset_id,entity_id,role,source) VALUES(1,?,'performer','test')", (person,))
self.target.execute("INSERT INTO asset_tag(asset_id,tag,source) VALUES(1,'演员:天使もえ','test')")
report = seed_pack.land(self.target, pack)
self.assertEqual(self.target.execute("SELECT canonical_name FROM entity WHERE id=?", (person,)).fetchone()[0], "天使萌")
self.assertEqual(self.target.execute("SELECT tag FROM asset_tag WHERE asset_id=1").fetchone()[0], "演员:天使萌")
self.assertEqual(report["names"], 1)
self.assertIn("天使もえ", {r[0] for r in self.target.execute("SELECT alias FROM entity_alias WHERE entity_id=?", (person,))})
with self.target:
self.assertEqual(seed_pack.written(seed_pack.land(self.target, pack)), 0)
target_path = self.target.execute("PRAGMA database_list").fetchone()[2]
revert = load_script("revert_auto_landing")
with contextlib.redirect_stdout(io.StringIO()):
self.assertEqual(revert.main(["--db", target_path, "--source", "auto:seed", "--apply",
"--backup", str(self.root / "pre-revert.db"), "--logo-root", str(self.root / "logos")]), 0)
self.assertEqual(self.target.execute("SELECT canonical_name FROM entity WHERE id=?", (person,)).fetchone()[0], "天使もえ")
self.assertEqual(self.target.execute("SELECT tag FROM asset_tag WHERE asset_id=1").fetchone()[0], "演员:天使もえ")
def test_a_user_preferred_name_is_preserved(self):
from peach.entities import prefer_entity_name
pack = self.name_pack()
with self.target:
person = entity(self.target, "performer", "天使萌")
alias(self.target, person, "天使もえ", "user:manual")
prefer_entity_name(self.target, person, "天使もえ")
seed_pack.land(self.target, pack)
self.assertEqual(self.target.execute("SELECT canonical_name FROM entity WHERE id=?", (person,)).fetchone()[0], "天使もえ")
self.assertEqual(seed_pack.planned_names(self.target, "auto:seed"), [])
def test_confirming_the_current_name_protects_it_from_localization(self):
from peach.entities import prefer_entity_name
pack = self.name_pack()
with self.target:
person = entity(self.target, "performer", "天使もえ")
prefer_entity_name(self.target, person, "天使もえ")
seed_pack.land(self.target, pack)
self.assertEqual(self.target.execute("SELECT canonical_name FROM entity WHERE id=?", (person,)).fetchone()[0], "天使もえ")
def test_revision_identity_and_aliases_must_resolve_to_one_person(self):
pack = self.name_pack(revision=1)
with self.target:
person = entity(self.target, "performer", "天使もえ")
report = seed_pack.land(self.target, pack)
self.assertEqual(report["names"], 1)
self.assertEqual(self.target.execute("SELECT canonical_name FROM entity WHERE id=?", (person,)).fetchone()[0], "天使萌")
def test_custom_names_and_ambiguous_identities_are_preserved(self):
pack = self.name_pack()
with self.target:
person = entity(self.target, "performer", "我选的名字")
alias(self.target, person, "天使もえ", "user:manual")
seed_pack.land(self.target, pack)
self.assertEqual(self.target.execute("SELECT canonical_name FROM entity WHERE id=?", (person,)).fetchone()[0], "我选的名字")
with self.target:
entity(self.target, "performer", "天使萌")
report = seed_pack.land(self.target, pack)
self.assertEqual(len(report["duplicates"]), 1)
self.assertEqual(report.get("names", 0), 0)
class ExportTests(SeedPackCase):
def test_pack_carries_public_facts_and_nothing_local(self):
pack = self.seeded_source()
self.assertEqual(pack["version"], VERSION)
self.assertEqual(pack["counts"], {"entities": 3, "aliases": 3, "refs": 1, "links": 3,
"profiles": 1, "memberships": 1, "makers": 1})
by_name = {item["name"]: item for item in pack["entities"]}
self.assertEqual(list(by_name), ["天川そら", "PRESTIGE PREMIUM", "S1 NO.1 STYLE"], "按种类与归一名排序")
self.assertNotIn("STARTUP", by_name, "事务所自己没有可补的事实，只作为她的归属出现")
self.assertNotIn("Prestige", by_name, "片商自己没有可补的事实，只作为 label 的归属出现")
self.assertEqual(by_name["PRESTIGE PREMIUM"]["maker"],
{"name": "Prestige", "source": "review:label-maker-20260923.csv"})
sora = by_name["天川そら"]
self.assertEqual([(a["alias"], a["source"]) for a in sora["aliases"]],
[("二宮そら", "javdb-actor-page@javdb-20260912"), ("春山心愛", "r18:performer")],
"按归一写法排序，同一写法只留一个来源；罗马字、短单名与种子自己写的不导")
self.assertEqual(sora["refs"], [{"provider": "minnano-av", "kind": "performer", "id": "295275"}],
"Stash 的行号与种子写的编号都不导")
self.assertEqual([l["url"] for l in sora["links"]],
["https://www.startup0701.net/", "https://x.com/amakawa_sora_"],
"敏感链接、目录页与种子写的链接都不导；主机按导入同一套规则归一，大小写不进包")
self.assertEqual(sora["profile"]["fields"],
{"kana": "あまかわそら", "birth_date": "1998-10-10", "height_cm": 164, "cup": "G",
"tags": ["巨乳", "美肌"]}, "资料按列导，raw 残片不导")
self.assertEqual(sora["profile"]["source"], "auto:performer-profile@7")
self.assertEqual(sora["agency"], {"name": "STARTUP", "source": "minnano-av:所属事務所"})
self.assertNotIn("只有名字", by_name, "光一个名字对别的数据库没有用处")
text = seed_pack.dump(pack)
self.assertNotIn("raw_snapshot", text)
self.assertNotIn("somewhere", text)
def test_two_exports_of_the_same_ledger_are_byte_identical(self):
first = seed_pack.dump(self.seeded_source())
second = seed_pack.dump(seed_pack.export_pack(self.source, version=VERSION))
self.assertEqual(first, second)
class ImportTests(SeedPackCase):
def seeded_target(self) -> dict[str, int]:
"""另一台机器：账本里她叫「春山心愛」，事务所已登记，厂牌、label 与片商都只有名字；
包里的另一位 `未登记` 本机没有。"""
con = self.target
with con:
ids = {"her": entity(con, "performer", "春山心愛"), "agency": entity(con, "agency", "STARTUP"),
"studio": entity(con, "studio", "S1 NO.1 STYLE"), "other": entity(con, "performer", "别人"),
"label": entity(con, "studio", "PRESTIGE PREMIUM"), "maker": entity(con, "studio", "Prestige")}
return ids
def test_only_existing_entities_get_their_blanks_filled(self):
pack = self.seeded_source()
pack["entities"].append({"kind": "performer", "name": "未登记", "aliases": [], "refs": [],
"links": [{"kind": "social", "label": "X", "url": "https://x.com/nobody"}]})
ids = self.seeded_target()
with self.target:
report = seed_pack.land(self.target, pack)
self.assertEqual(report, {"version": VERSION, "batch": BATCH, "matched": 3, "unmatched": 1,
"aliases": 3, "refs": 1, "links": 3, "profiles": 1, "memberships": 1,
"makers": 1, "refreshed": 0, "conflicts": [], "duplicates": []})
her = ids["her"]
rows = self.target.execute("SELECT alias,source FROM entity_alias WHERE entity_id=? ORDER BY alias",
(her,)).fetchall()
self.assertEqual([tuple(row) for row in rows], [("二宮そら", BATCH), ("天川そら", BATCH)],
"包里的规范名与别名都补成别名，本机规范名不动，来源是批次号")
self.assertEqual(self.target.execute("SELECT canonical_name FROM entity WHERE id=?", (her,)).fetchone()[0],
"春山心愛")
provider, external_id, metadata = self.target.execute(
"SELECT provider,external_id,metadata_json FROM entity_external_ref WHERE entity_id=?", (her,)).fetchone()
self.assertEqual((provider, external_id, json.loads(metadata)),
("minnano-av", "295275", {"source": "auto:seed", "batch": BATCH}))
links = self.target.execute("SELECT url,metadata_json FROM entity_link WHERE entity_id=? ORDER BY url",
(her,)).fetchall()
self.assertEqual([row[0] for row in links], ["https://www.startup0701.net/", "https://x.com/amakawa_sora_"])
self.assertEqual(json.loads(links[0][1]), {"source": "auto:seed", "batch": BATCH})
profile = read_profile(self.target, her)
self.assertEqual((profile["kana"], profile["height_cm"], profile["tags"], profile["raw"], profile["source"]),
("あまかわそら", 164, ["巨乳", "美肌"], {}, BATCH))
self.assertEqual(profile["fetched_at"], STAMP, "取回时间照包里的记，30 天算旧的规则接着管它")
member = self.target.execute("SELECT agency_id,source FROM entity_membership WHERE member_id=?",
(her,)).fetchone()
self.assertEqual(tuple(member), (ids["agency"], BATCH))
studio_alias = self.target.execute("SELECT alias,source FROM entity_alias WHERE entity_id=?",
(ids["studio"],)).fetchone()
self.assertEqual(tuple(studio_alias), ("S1", BATCH))
maker = self.target.execute("SELECT maker_id,source FROM label_maker WHERE label_id=?",
(ids["label"],)).fetchone()
self.assertEqual(tuple(maker), (ids["maker"], BATCH), "label 归哪家片商也补上，来源是批次号")
self.assertEqual(self.target.execute("SELECT count(*) FROM entity").fetchone()[0], 6, "不造实体")
def test_seed_never_overwrites_what_the_ledger_already_holds(self):
pack = self.seeded_source()
ids = self.seeded_target()
with self.target:
write_profile(self.target, ids["her"], {"kana": "本机刮的", "raw": {}},
source="auto:performer-profile@99", source_url="https://example/", fetched_at=STAMP)
membership(self.target, ids["her"], ids["agency"], "review:user")
link(self.target, ids["her"], "social", "https://X.com/amakawa_sora_")
alias(self.target, ids["her"], "二宮そら", "user:manual")
kmp = entity(self.target, "studio", "K M Produce")
label_maker(self.target, ids["label"], kmp, "review:user")
with self.target:
report = seed_pack.land(self.target, pack)
self.assertEqual((report["profiles"], report["refs"], report["memberships"], report["links"],
report["aliases"], report["makers"], report["refreshed"]), (0, 1, 0, 2, 2, 0, 0))
self.assertEqual(report["conflicts"],
[{"kind": "studio", "entity": "PRESTIGE PREMIUM", "field": "maker", "ours": "K M Produce",
"source": "review:user", "theirs": "Prestige"}],
"人复核过的片商与包里不一致：不动，记进明细等人看")
self.assertEqual(self.target.execute("SELECT maker_id FROM label_maker WHERE label_id=?",
(ids["label"],)).fetchone()[0], kmp)
self.assertEqual(self.target.execute("SELECT count(*) FROM entity_link WHERE entity_id=?",
(ids["her"],)).fetchone()[0], 2,
"本机那条主机大写的同一地址不再补一条只差大小写的")
self.assertEqual(read_profile(self.target, ids["her"])["source"], "auto:performer-profile@99",
"本机后继刚刮的资料比种子新，不被盖掉")
self.assertEqual(self.target.execute("SELECT source FROM entity_membership WHERE member_id=?",
(ids["her"],)).fetchone()[0], "review:user")
rows = self.target.execute("SELECT alias,source FROM entity_alias WHERE entity_id=? ORDER BY alias",
(ids["her"],)).fetchall()
self.assertEqual([tuple(row) for row in rows], [("二宮そら", "user:manual"), ("天川そら", BATCH)],
"已有的写法保留原来源，只补缺的那个")
def test_a_newer_seed_replaces_what_an_older_seed_wrote(self):
pack = self.seeded_source()
ids = self.seeded_target()
with self.target:
old_agency = entity(self.target, "agency", "旧事务所")
old_maker = entity(self.target, "studio", "旧片商")
membership(self.target, ids["her"], old_agency, "auto:seed@2026-01-01")
label_maker(self.target, ids["label"], old_maker, "auto:seed@2026-01-01")
write_profile(self.target, ids["her"], {"kana": "旧种子写的", "raw": {}}, source="auto:seed@2026-01-01",
source_url="https://example/", fetched_at="2026-01-01T00:00:00Z")
with self.target:
report = seed_pack.land(self.target, pack)
self.assertEqual((report["memberships"], report["makers"], report["profiles"], report["refreshed"],
report["conflicts"]), (0, 0, 0, 3, []), "旧种子写的三行都换成新批次，不算新补也不算冲突")
self.assertEqual(tuple(self.target.execute("SELECT agency_id,source FROM entity_membership WHERE member_id=?",
(ids["her"],)).fetchone()), (ids["agency"], BATCH))
self.assertEqual(tuple(self.target.execute("SELECT maker_id,source FROM label_maker WHERE label_id=?",
(ids["label"],)).fetchone()), (ids["maker"], BATCH))
profile = read_profile(self.target, ids["her"])
self.assertEqual((profile["kana"], profile["source"]), ("あまかわそら", BATCH))
with self.target:
again = seed_pack.land(self.target, pack)
self.assertEqual(seed_pack.written(again), 0, "同一包再跑一遍不再动任何一行")
def test_a_name_on_one_entity_and_a_number_on_another_matches_nobody(self):
pack = self.seeded_source()
ids = self.seeded_target()
with self.target:
ref(self.target, ids["other"], "minnano-av", "295275")
with self.target:
report = seed_pack.land(self.target, pack)
self.assertEqual((report["matched"], report["unmatched"], report["duplicates"]),
(2, 0, [{"kind": "performer", "name": "天川そら", "entities": ["别人", "春山心愛"]}]),
"两家厂牌对上了；她那条认出两位，记进重复身份明细")
for entity_id in (ids["her"], ids["other"]):
self.assertEqual(self.target.execute("SELECT count(*) FROM entity_alias WHERE entity_id=?",
(entity_id,)).fetchone()[0], 0, "名字与编号各指一位，谁都不动")
owner = self.target.execute("SELECT entity_id FROM entity_external_ref WHERE external_id='295275'").fetchone()
self.assertEqual(owner[0], ids["other"], "别人已登记的编号不改指")
def test_two_local_matches_are_left_alone(self):
pack = self.seeded_source()
with self.target:
first = entity(self.target, "performer", "春山心愛")
second = entity(self.target, "performer", "二宮そら")
with self.target:
report = seed_pack.land(self.target, pack)
self.assertEqual((report["matched"], report["unmatched"], report["duplicates"]),
(0, 2, [{"kind": "performer", "name": "天川そら", "entities": ["二宮そら", "春山心愛"]}]))
for entity_id in (first, second):
self.assertEqual(self.target.execute("SELECT count(*) FROM entity_alias WHERE entity_id=?",
(entity_id,)).fetchone()[0], 0, "两位都对得上就谁都不动")
def test_plan_reports_the_same_counts_without_touching_the_ledger(self):
pack = self.seeded_source()
self.seeded_target()
before = self.target.execute("SELECT count(*) FROM entity_alias").fetchone()[0]
report = seed_pack.plan(self.target, pack)
self.assertEqual((report["matched"], report["aliases"], report["profiles"]), (3, 3, 1))
self.assertEqual(self.target.execute("SELECT count(*) FROM entity_alias").fetchone()[0], before)
def test_a_seed_batch_is_reverted_across_all_six_tables(self):
pack = self.seeded_source()
ids = self.seeded_target()
with self.target:
alias(self.target, ids["her"], "人写的", "user:manual")
seed_pack.land(self.target, pack)
self.target.close()
revert = load_script("revert_auto_landing")
db = self.root / "target.db"
base = ["--db", str(db), "--source", "auto:seed", "--logo-root", str(self.root / "no-logos")]
with contextlib.redirect_stdout(io.StringIO()) as out:
self.assertEqual(revert.main(base), 0)
self.assertIn("'归属': 1", out.getvalue())
self.assertIn("'片商': 1", out.getvalue())
with contextlib.redirect_stdout(io.StringIO()):
self.assertEqual(revert.main([*base, "--apply", "--backup", str(self.root / "backup.db")]), 0)
con = connect(db)
self.addCleanup(con.close)
counts = [con.execute(sql).fetchone()[0] for sql in (
"SELECT count(*) FROM entity_alias", "SELECT count(*) FROM entity_external_ref",
"SELECT count(*) FROM entity_link", "SELECT count(*) FROM performer_profile",
"SELECT count(*) FROM entity_membership", "SELECT count(*) FROM label_maker")]
self.assertEqual(counts, [1, 0, 0, 0, 0, 0], "只剩人写的那条别名")
class ScriptTests(SeedPackCase):
def test_default_pack_is_a_valid_bundled_public_resource(self):
script = load_script("seed_pack")
export = script.build_parser().parse_args(["export"])
self.assertEqual(export.out, seed_pack.DEFAULT_PACK)
self.assertEqual(export.out, ROOT / "resources/seed/entities.json")
pack = seed_pack.load(export.out)
self.assertEqual(pack["format"], 2)
self.assertGreater(pack["counts"]["entities"], 0)
self.assertGreater(pack["revision"], 0)
def test_pack_is_required_in_each_packaging_layout(self):
from peach.bundle_resources import data_files
with self.assertRaises(FileNotFoundError):
data_files(self.root)
for name in ("web", "migrations", "resources", "tools/amane-bridge"):
(self.root / name).mkdir(parents=True, exist_ok=True)
for name in ("bridge.py", "pyproject.toml", "uv.lock"):
(self.root / "tools/amane-bridge" / name).touch()
seed = self.root / "resources/seed/entities.json"
seed.parent.mkdir()
seed.write_text(seed_pack.dump(self.seeded_source()), encoding="utf-8")
entries = data_files(self.root)
# data_files 给的是解析过链接的真实路径；macOS 的临时目录 /var 指向 /private/var。
self.assertIn((self.root.resolve() / "resources", "resources"), entries)
seed.unlink()
with self.assertRaises(FileNotFoundError):
data_files(self.root)
def test_validate_checks_package_without_opening_a_database(self):
pack = self.root / "package.json"
pack.write_text(seed_pack.dump(self.seeded_source()), encoding="utf-8")
script = load_script("seed_pack")
with mock.patch.object(script, "open_readonly", side_effect=AssertionError("database must remain closed")):
with contextlib.redirect_stdout(io.StringIO()) as output:
self.assertEqual(script.main(["validate", "--pack", str(pack)]), 0)
self.assertTrue(json.loads(output.getvalue())["valid"])
pack.write_text('{"format":true}', encoding="utf-8")
with contextlib.redirect_stdout(io.StringIO()) as rejected:
self.assertEqual(script.main(["validate", "--pack", str(pack)]), 2)
self.assertFalse(json.loads(rejected.getvalue())["written"])
def test_export_is_idempotent_and_import_is_dry_run_by_default(self):
self.seeded_source()
self.source.close()
script = load_script("seed_pack")
out = self.root / "seed" / "entities.json"
with contextlib.redirect_stdout(io.StringIO()) as first:
self.assertEqual(script.main(["export", "--db", str(self.root / "source.db"), "--version", VERSION,
"--out", str(out)]), 0)
with contextlib.redirect_stdout(io.StringIO()) as second:
script.main(["export", "--db", str(self.root / "source.db"), "--version", VERSION, "--out", str(out)])
self.assertEqual((json.loads(first.getvalue())["changed"], json.loads(second.getvalue())["changed"]),
(True, False))
self.assertEqual(seed_pack.load(out)["counts"]["entities"], 3)
with self.target:
entity(self.target, "performer", "天川そら")
self.target.close()
target = self.root / "target.db"
with contextlib.redirect_stdout(io.StringIO()) as dry:
self.assertEqual(script.main(["import", "--db", str(target), "--pack", str(out)]), 0)
self.assertIn("dry-run", dry.getvalue())
con = connect(target)
self.addCleanup(con.close)
self.assertEqual(con.execute("SELECT count(*) FROM entity_alias").fetchone()[0], 0)
with contextlib.redirect_stdout(io.StringIO()) as applied:
self.assertEqual(script.main(["import", "--db", str(target), "--pack", str(out), "--apply",
"--backup", str(self.root / "pre-seed.db")]), 0)
report = json.loads(applied.getvalue())
self.assertEqual((report["matched"], report["aliases"], report["integrity_check"]), (1, 2, "ok"))
self.assertTrue((self.root / "pre-seed.db").exists())
def test_a_changed_export_that_keeps_the_version_is_refused(self):
"""版本与内容绑定，导出内容变化须推进版本。"""
self.seeded_source()
script = load_script("seed_pack")
out, db = self.root / "seed" / "entities.json", str(self.root / "source.db")
with contextlib.redirect_stdout(io.StringIO()):
self.assertEqual(script.main(["export", "--db", db, "--version", VERSION, "--out", str(out)]), 0)
before = out.read_text(encoding="utf-8")
sora = self.source.execute("SELECT id FROM entity WHERE canonical_name='天川そら'").fetchone()[0]
with self.source:
alias(self.source, sora, "改名后的艺名", "r18:performer")
self.source.close()
with contextlib.redirect_stdout(io.StringIO()) as refused:
self.assertEqual(script.main(["export", "--db", db, "--version", VERSION, "--out", str(out)]), 2)
report = json.loads(refused.getvalue())
self.assertEqual((report["version"], report["changed"], report["written"]), (VERSION, True, False))
self.assertIn("换一个新版本串", report["error"])
self.assertEqual(out.read_text(encoding="utf-8"), before, "被拒的导出不动文件")
with contextlib.redirect_stdout(io.StringIO()) as renewed:
self.assertEqual(script.main(["export", "--db", db, "--version", f"{VERSION}.1", "--out", str(out)]), 0)
self.assertTrue(json.loads(renewed.getvalue())["changed"])
self.assertEqual((seed_pack.load(out)["version"], seed_pack.load(out)["counts"]["aliases"]),
(f"{VERSION}.1", 4))
# 文件已是 `.1`，再改内容后用缺省那样的当天日期导出：比现有版本串旧，同样拒绝。
source = connect(self.root / "source.db")
with source:
alias(source, sora, "再改一次", "r18:performer")
source.close()
with contextlib.redirect_stdout(io.StringIO()) as older:
self.assertEqual(script.main(["export", "--db", db, "--version", VERSION, "--out", str(out)]), 2)
self.assertIn("需要更新的 version 与 revision", json.loads(older.getvalue())["error"])
self.assertEqual(seed_pack.load(out)["version"], f"{VERSION}.1")
if __name__ == "__main__":
unittest.main()