tests/test_seed_corrections.py
Rama master · réplica de lectura del repositorio original.
"""种子纠错只撤销自己的精确断言，包历史阻止回放。"""
import copy
import sqlite3
from unittest import mock
from peach import seed_pack
from peach.performer_profiles import read_profile, write_profile
from test_seed_pack import SeedPackCase, entity, alias, ref, link, membership, label_maker, STAMP
class CorrectionTests(SeedPackCase):
def setUp(self):
super().setUp()
with self.source:
self.person = entity(self.source, "performer", "天川そら")
ref(self.source, self.person, "minnano-av", "123")
with self.target:
self.recipient = entity(self.target, "performer", "天川そら")
def pack(self, revision, withdrawals=()):
return seed_pack.export_pack(self.source, version=f"2026-10-{revision:02}",
revision=revision, withdrawals=list(withdrawals))
def withdrawal(self, field, value, *, name="天川そら", key="correction-1"):
return {"id": key, "identity": {"kind": "performer", "name": name,
"refs": [{"provider": "minnano-av", "kind": "performer", "id": "123"}]},
"field": field, "value": value, "reason": "来源核对确认指向错误", "evidence": "review:verified.csv"}
def land(self, pack):
with self.target:
return seed_pack.land(self.target, pack)
def test_explicit_alias_ref_and_link_withdrawals_allow_corrected_ref(self):
with self.source:
alias(self.source, self.person, "错误名字", "review:source")
ref(self.source, self.person, "javdb", "wrong")
link(self.source, self.person, "social", "https://x.com/wrong")
self.land(self.pack(1))
withdrawals = [self.withdrawal("alias", "错误名字", key="bad-alias"),
self.withdrawal("ref", {"provider": "javdb", "kind": "performer", "id": "wrong"}, key="bad-ref"),
self.withdrawal("link", {"kind": "social", "url": "https://x.com/wrong"}, key="bad-link")]
with self.source:
self.source.execute("DELETE FROM entity_alias")
self.source.execute("DELETE FROM entity_external_ref WHERE provider='javdb'")
self.source.execute("DELETE FROM entity_link")
ref(self.source, self.person, "javdb", "correct")
fixed = self.pack(2, withdrawals)
report = self.land(fixed)
self.assertEqual(report["withdrawn"], 3)
self.assertIsNone(self.target.execute("SELECT 1 FROM entity_alias WHERE alias='错误名字'").fetchone())
self.assertEqual(self.target.execute("SELECT external_id FROM entity_external_ref WHERE provider='javdb'").fetchone()[0], "correct")
self.assertEqual(self.target.execute("SELECT count(*) FROM entity_link").fetchone()[0], 0)
self.assertEqual(seed_pack.written(self.land(fixed)), 0)
def test_withdrawal_preserves_each_nonseed_owner(self):
for owner in ("user:manual", "review:verified", "auto:performer-profile@4"):
with self.subTest(owner=owner), self.target:
self.target.execute("DELETE FROM entity_alias")
alias(self.target, self.recipient, "错误名字", owner)
report = seed_pack.land(self.target, self.pack(1, [self.withdrawal("alias", "错误名字")]))
self.assertEqual(report["withdrawn"], 0)
self.assertEqual(self.target.execute("SELECT source FROM entity_alias").fetchone()[0], owner)
self.assertEqual(len(report["withdrawal_conflicts"]), 1)
def test_profile_correction_uses_revision_even_with_same_fetched_at(self):
with self.source:
write_profile(self.source, self.person, {"height_cm": 164}, source="review:profile", source_url="https://example.com/profile", fetched_at=STAMP)
self.land(self.pack(1))
with self.source:
self.source.execute("UPDATE performer_profile SET height_cm=165")
report = self.land(self.pack(2))
self.assertEqual(read_profile(self.target, self.recipient)["height_cm"], 165)
self.assertEqual(report["refreshed"], 1)
self.assertEqual(seed_pack.written(self.land(self.pack(2))), 0)
def test_cumulative_withdrawal_supports_skipped_revision_and_rejects_replay(self):
with self.source:
alias(self.source, self.person, "错误名字", "review:source")
initial = self.pack(1)
self.land(initial)
with self.source:
self.source.execute("DELETE FROM entity_alias")
withdrawal = self.withdrawal("alias", "错误名字")
self.land(self.pack(3, [withdrawal]))
with self.assertRaises(ValueError):
self.land(initial)
with self.assertRaises(ValueError):
self.land(seed_pack.export_pack(self.source, version="2099-01-01"))
self.assertIsNone(self.target.execute("SELECT 1 FROM entity_alias WHERE alias='错误名字'").fetchone())
with self.assertRaises(ValueError):
self.land(self.pack(4))
def test_ambiguous_withdrawal_leaves_rows_and_records_review(self):
with self.target:
other = entity(self.target, "performer", "另一位艺人")
ref(self.target, other, "minnano-av", "123")
alias(self.target, self.recipient, "错误名字", "auto:seed@old")
report = self.land(self.pack(1, [self.withdrawal("alias", "错误名字")]))
self.assertEqual(report["withdrawn"], 0)
self.assertEqual(len(report["withdrawal_conflicts"]), 1)
self.assertEqual(self.target.execute("SELECT count(*) FROM entity_alias").fetchone()[0], 1)
def test_absence_is_not_a_withdrawal(self):
with self.source:
alias(self.source, self.person, "已登记别名", "review:source")
self.land(self.pack(1))
with self.source:
self.source.execute("DELETE FROM entity_alias")
self.land(self.pack(2))
self.assertEqual(self.target.execute("SELECT count(*) FROM entity_alias WHERE alias='已登记别名'").fetchone()[0], 1)
def test_modified_content_and_invalid_pack_are_rejected_atomically(self):
pack = self.pack(1)
broken = copy.deepcopy(pack)
broken["entities"][0]["aliases"].append({"alias": "无效包的名字", "source": "review:source"})
before = list(self.target.iterdump())
with self.assertRaises(ValueError):
self.land(broken)
self.assertEqual(list(self.target.iterdump()), before)
self.land(pack)
with self.source:
alias(self.source, self.person, "同版本新内容", "review:source")
with self.assertRaises(ValueError):
self.land(self.pack(1))
def test_schema_rejects_bool_revision_unknown_fields_and_duplicate_ref_identity(self):
for mutate in (
lambda p: p.update(revision=True),
lambda p: p.update(unexpected="value"),
lambda p: p["entities"].append({**copy.deepcopy(p["entities"][0]), "name": "另一位艺人"})):
with self.subTest(mutate=mutate):
pack = self.pack(1)
mutate(pack)
before = list(self.target.iterdump())
with self.assertRaises(ValueError):
self.land(pack)
self.assertEqual(list(self.target.iterdump()), before)
def test_same_package_cannot_assert_its_withdrawn_value(self):
with self.source:
alias(self.source, self.person, "错误名字", "review:source")
with self.assertRaises(ValueError):
self.pack(1, [self.withdrawal("alias", "错误名字")])
def test_export_inherits_withdrawals_from_explicit_previous_package(self):
previous = self.pack(1, [self.withdrawal("alias", "错误名字")])
later = seed_pack.export_pack(self.source, version="2026-10-04", revision=4, previous=previous)
self.assertEqual(later["withdrawals"], previous["withdrawals"])
def test_profile_and_both_pointer_types_are_precisely_withdrawn(self):
with self.source:
agency = entity(self.source, "agency", "事务所甲")
maker = entity(self.source, "studio", "片商甲")
label = entity(self.source, "studio", "厂牌甲")
membership(self.source, self.person, agency, "review:source")
label_maker(self.source, label, maker, "review:source")
write_profile(self.source, self.person, {"height_cm": 164}, source="review:source", source_url="https://example.com/profile", fetched_at=STAMP)
with self.target:
entity(self.target, "agency", "事务所甲")
entity(self.target, "studio", "片商甲")
entity(self.target, "studio", "厂牌甲")
first = self.pack(1)
self.land(first)
person = next(item for item in first["entities"] if item["kind"] == "performer")
studio = next(item for item in first["entities"] if item["name"] == "厂牌甲")
withdrawals = [self.withdrawal("profile", person["profile"], key="bad-profile"),
self.withdrawal("agency", person["agency"], key="bad-agency"),
{"id": "bad-maker", "identity": studio["identity"], "field": "maker", "value": studio["maker"],
"reason": "已核对归属", "evidence": "review:verified.csv"}]
with self.source:
self.source.execute("DELETE FROM performer_profile")
self.source.execute("DELETE FROM entity_membership")
self.source.execute("DELETE FROM label_maker")
self.assertEqual(self.land(self.pack(2, withdrawals))["withdrawn"], 3)
self.assertIsNone(read_profile(self.target, self.recipient))
self.assertEqual(self.target.execute("SELECT count(*) FROM entity_membership").fetchone()[0], 0)
self.assertEqual(self.target.execute("SELECT count(*) FROM label_maker").fetchone()[0], 0)
def test_profile_withdrawal_requires_all_nonempty_columns(self):
with self.target:
write_profile(self.target, self.recipient, {"height_cm": 164, "birthplace": "東京都"}, source="auto:seed@old", source_url="https://example.com/profile", fetched_at=STAMP)
partial = {"fields": {"height_cm": 164}, "source": "review:source", "source_url": "https://example.com/profile", "fetched_at": STAMP}
report = self.land(self.pack(1, [self.withdrawal("profile", partial)]))
self.assertEqual(report["withdrawn"], 0)
self.assertEqual(len(report["withdrawal_conflicts"]), 1)
self.assertEqual(read_profile(self.target, self.recipient)["birthplace"], "東京都")
def test_runtime_error_rolls_back_withdrawals_facts_and_history(self):
with self.source:
alias(self.source, self.person, "错误名字", "review:source")
self.land(self.pack(1))
with self.source:
self.source.execute("DELETE FROM entity_alias")
fixed = self.pack(2, [self.withdrawal("alias", "错误名字")])
before = list(self.target.iterdump())
with mock.patch.object(seed_pack.seed_corrections, "remember", side_effect=sqlite3.OperationalError("simulated failure")):
with self.assertRaises(sqlite3.OperationalError):
seed_pack.land(self.target, fixed)
self.assertEqual(list(self.target.iterdump()), before)
def test_nonseed_profile_is_not_corrected_by_new_revision(self):
with self.source:
write_profile(self.source, self.person, {"height_cm": 164}, source="review:source", source_url="https://example.com/profile", fetched_at=STAMP)
with self.target:
write_profile(self.target, self.recipient, {"height_cm": 180}, source="auto:performer-profile@5", source_url="https://example.com/local", fetched_at=STAMP)
self.land(self.pack(1))
self.assertEqual(read_profile(self.target, self.recipient)["height_cm"], 180)
def test_schema_failure_after_valid_entity_cannot_land_any_rows(self):
pack = seed_pack.export_pack(self.source, version="2026-10-01")
pack["entities"].append({"kind": "performer", "name": "另一位", "aliases": False, "refs": [], "links": []})
before = list(self.target.iterdump())
with self.assertRaises(ValueError):
seed_pack.land(self.target, pack)
self.assertEqual(list(self.target.iterdump()), before)
def test_conflicting_reference_identities_are_rejected_without_hash_gate(self):
pack = seed_pack.export_pack(self.source, version="2026-10-01")
pack["entities"].append({"kind": "performer", "name": "另一位艺人", "aliases": [],
"refs": copy.deepcopy(pack["entities"][0]["refs"]), "links": []})
before = list(self.target.iterdump())
with self.assertRaisesRegex(ValueError, "编号身份冲突"):
seed_pack.land(self.target, pack)
self.assertEqual(list(self.target.iterdump()), before)
def test_alias_confidence_rejects_boolean_and_out_of_range_values(self):
with self.source:
alias(self.source, self.person, "来源别名", "review:source")
for confidence in (True, None, -0.1, 1.1):
with self.subTest(confidence=confidence):
pack = self.pack(1)
pack["entities"][0]["aliases"][0]["confidence"] = confidence
pack["content_sha256"] = seed_pack.seed_validation.digest(pack)
before = list(self.target.iterdump())
with self.assertRaises(ValueError):
seed_pack.land(self.target, pack)
self.assertEqual(list(self.target.iterdump()), before)
def test_caller_rollback_restores_facts_and_version_history(self):
with self.assertRaises(RuntimeError):
with self.target:
seed_pack.land(self.target, self.pack(1))
raise RuntimeError("caller abort")
self.assertEqual(self.target.execute("SELECT count(*) FROM entity_external_ref").fetchone()[0], 0)
self.assertEqual(self.target.execute("SELECT count(*) FROM seed_pack_history").fetchone()[0], 0)
def test_source_confidence_round_trip_preserves_manual_values(self):
with self.source:
alias(self.source, self.person, "来源别名", "r18:performer")
self.source.execute("UPDATE entity_alias SET confidence=0.42")
pack = self.pack(1)
self.assertEqual(pack["entities"][0]["aliases"][0]["confidence"], 0.42)
self.land(pack)
held = self.target.execute("SELECT confidence,source FROM entity_alias WHERE alias='来源别名'").fetchone()
self.assertEqual(tuple(held), (0.42, "auto:seed@2026-10-01"))
with self.target:
self.target.execute("UPDATE entity_alias SET confidence=0.61,source='user:manual' WHERE alias='来源别名'")
with self.source:
self.source.execute("UPDATE entity_alias SET confidence=0.3")
self.land(self.pack(2))
self.assertEqual(tuple(self.target.execute("SELECT confidence,source FROM entity_alias WHERE alias='来源别名'").fetchone()), (0.61, "user:manual"))
def test_source_confidence_corrects_seed_owned_legacy_confidence(self):
with self.source:
alias(self.source, self.person, "来源别名", "r18:performer")
self.source.execute("UPDATE entity_alias SET confidence=0.42")
legacy = seed_pack.export_pack(self.source, version="2026-09-25")
self.land(legacy)
self.assertEqual(self.target.execute("SELECT confidence FROM entity_alias WHERE alias='来源别名'").fetchone()[0], 1)
report = self.land(self.pack(1))
self.assertEqual(report["refreshed"], 1)
self.assertEqual(self.target.execute("SELECT confidence FROM entity_alias WHERE alias='来源别名'").fetchone()[0], 0.42)
self.assertEqual(seed_pack.written(self.land(self.pack(1))), 0)
def test_stash_aliases_are_not_exported_as_public_v2_facts(self):
with self.source:
alias(self.source, self.person, "本机导入别名", "stash:performer")
self.assertTrue(seed_pack.export_pack(self.source, version="2026-09-25")["entities"][0]["aliases"])
pack = self.pack(1)
self.assertEqual(pack["entities"][0]["aliases"], [])
self.assertEqual(pack["withdrawals"], [])
def test_local_import_only_people_remain_local(self):
local_people = (("古河君", "古河X沐沐", "61"),
("小芽芽不乖", "91社区21442644", "63"))
with self.source:
for name, other_name, external_id in local_people:
person = entity(self.source, "performer", name)
alias(self.source, person, other_name, "stash:performer")
ref(self.source, person, "stash", external_id)
snapshot = {
table: [tuple(row) for row in self.source.execute(f"SELECT * FROM {table} ORDER BY 1")]
for table in ("entity", "entity_alias", "entity_external_ref")}
pack = self.pack(1)
self.assertEqual([item["name"] for item in pack["entities"]], ["天川そら"])
self.assertEqual(pack["counts"]["entities"], 1)
self.assertEqual(pack["withdrawals"], [])
self.assertEqual(snapshot, {
table: [tuple(row) for row in self.source.execute(f"SELECT * FROM {table} ORDER BY 1")]
for table in snapshot})