tests/test_agency_entity.py
Rama master · réplica de lectura del repositorio original.
"""事务所是实体：有资料页、有成员、搜得到。"""
import re
import shutil
import sqlite3
import tempfile
import unittest
from pathlib import Path
from peach import web_contract as rm_web
from peach.entities import merge_entity
from peach.migrations import needs_foreign_keys_off, upgrade
from peach.web_entity import PROFILE_KINDS, scope_predicate
from scripts.dedupe_entity_links import collect as dedupe_plan
from scripts.dedupe_entity_links import destination
from scripts.install_agencies import collect as agency_plan
from scripts.install_agencies import apply_rows as install_agencies
from scripts.install_agencies import split_name
from support.ledger import fresh_ledger
ROOT = Path(__file__).resolve().parents[1]
MIGRATIONS = ROOT / "migrations"
STAMP = "2026-09-05T00:00:00Z"
class ForeignKeysOffMarkerTests(unittest.TestCase):
"""重建父表的迁移要能关掉外键，而且只有它能。"""
def test_the_marker_is_read_from_the_leading_comment_block(self):
self.assertTrue(needs_foreign_keys_off(
"-- peach:foreign_keys=off\n-- 说明\nCREATE TABLE t(id INTEGER);\n"))
def test_a_migration_without_the_marker_keeps_foreign_keys_on(self):
self.assertFalse(needs_foreign_keys_off("-- 说明\nCREATE TABLE t(id INTEGER);\n"))
def test_the_marker_after_the_first_statement_does_not_count(self):
"""pragma 要在整个文件开跑之前决定，写在语句之间等于没写。"""
self.assertFalse(needs_foreign_keys_off(
"CREATE TABLE t(id INTEGER);\n-- peach:foreign_keys=off\n"))
def test_rebuilding_entity_keeps_every_child_row(self):
"""外键开着时 DROP TABLE entity 会 CASCADE 删空子表，这条迁移不能那样。"""
with tempfile.TemporaryDirectory() as tmp:
older = Path(tmp).resolve() / "migrations"
older.mkdir()
for path in sorted(MIGRATIONS.glob("[0-9][0-9][0-9][0-9]_*.sql")):
if path.name < "0025":
shutil.copy2(path, older / path.name)
db = Path(tmp).resolve() / "ledger.db"
upgrade(db, older)
con = sqlite3.connect(db)
con.execute("INSERT INTO entity(id,kind,canonical_name,normalized_name,"
"created_at,updated_at) VALUES(7,'performer','阿花','阿花','t','t')")
con.execute("INSERT INTO entity_alias(entity_id,alias,normalized_alias,source)"
" VALUES(7,'Hana','hana','test')")
con.execute("INSERT INTO entity_link(entity_id,link_kind,label,url,hostname,"
"created_at,updated_at) VALUES(7,'social','X','https://x.com/h','x.com','t','t')")
con.commit()
con.close()
upgrade(db, MIGRATIONS)
con = sqlite3.connect(db)
try:
counts = (con.execute("SELECT count(*) FROM entity_alias").fetchone()[0],
con.execute("SELECT count(*) FROM entity_link").fetchone()[0])
name = con.execute("SELECT canonical_name FROM entity WHERE id=7").fetchone()[0]
broken = len(con.execute("PRAGMA foreign_key_check").fetchall())
finally:
# Windows 上临时目录删不掉带着打开句柄的库文件，断言失败也要先关。
con.close()
self.assertEqual(counts, (1, 1))
self.assertEqual(name, "阿花")
self.assertEqual(broken, 0)
def test_the_search_index_trigger_survives_the_rebuild(self):
"""DROP TABLE 把挂在 entity 上的触发器一起带走，迁移必须补回来。"""
with tempfile.TemporaryDirectory() as tmp:
db = Path(tmp).resolve() / "ledger.db"
upgrade(db, MIGRATIONS)
con = sqlite3.connect(db)
try:
triggers = con.execute(
"SELECT count(*) FROM sqlite_master WHERE type='trigger'"
" AND name='asset_search_entity_update'").fetchone()[0]
finally:
con.close()
self.assertEqual(triggers, 1)
def test_agency_is_an_accepted_entity_kind(self):
with tempfile.TemporaryDirectory() as tmp:
db = Path(tmp).resolve() / "ledger.db"
upgrade(db, MIGRATIONS)
con = sqlite3.connect(db)
try:
con.execute(
"INSERT INTO entity(kind,canonical_name,normalized_name,"
"created_at,updated_at) VALUES('agency','Capsule','capsule','t','t')")
with self.assertRaises(sqlite3.IntegrityError):
con.execute(
"INSERT INTO entity(kind,canonical_name,normalized_name,"
"created_at,updated_at) VALUES('label','X','x','t','t')")
finally:
con.close()
class AgencyNameTests(unittest.TestCase):
"""括号里是读音或旧名，括号外才是现用名。"""
def test_a_plain_name_has_no_aliases(self):
self.assertEqual(split_name("Capsule Agency"), ("Capsule Agency", []))
def test_the_reading_in_parentheses_becomes_an_alias(self):
self.assertEqual(split_name("ACT(アクト)"), ("ACT", ["アクト"]))
def test_a_former_name_loses_its_prefix(self):
self.assertEqual(split_name("Wish(元・GIRFY)"), ("Wish", ["GIRFY"]))
def test_a_former_name_after_the_closing_bracket_is_kept(self):
self.assertEqual(
split_name("SO MODEL AGENT(ソウ モデルエージェント)旧・Eightman Production"),
("SO MODEL AGENT", ["ソウ モデルエージェント", "Eightman Production"]))
class LinkDestinationTests(unittest.TestCase):
"""写法不同、去处相同的链接是一条，不是两条。"""
def test_a_trailing_slash_is_the_same_destination(self):
self.assertEqual(destination("http://kawakami-yuu.livedoor.biz"),
destination("http://kawakami-yuu.livedoor.biz/"))
def test_a_post_belongs_to_the_same_account_as_its_profile(self):
self.assertEqual(destination("https://x.com/yu_kinao"),
destination("https://x.com/yu_kinao/status/1221041995415580672"))
def test_two_accounts_on_one_platform_stay_apart(self):
self.assertNotEqual(destination("https://x.com/one"), destination("https://x.com/two"))
def test_a_query_string_still_distinguishes_pages(self):
self.assertNotEqual(destination("https://a.jp/t?id=3"), destination("https://a.jp/t?id=4"))
def test_percent_escapes_are_compared_by_byte_not_by_case(self):
"""凉森玲梦的官网被两个来源各按自己的大小写写了一遍，指向同一个页面。"""
self.assertEqual(
destination("https://www.t-powers.co.jp/talent/%e6%b6%bc%e6%a3%ae%e3%82%8c%e3%82%80/"),
destination("https://www.t-powers.co.jp/talent/%E6%B6%BC%E6%A3%AE%E3%82%8C%E3%82%80/"))
def test_two_escaped_paths_that_differ_stay_apart(self):
self.assertNotEqual(destination("https://a.jp/t/%E6%B6%BC"),
destination("https://a.jp/t/%E6%A3%AE"))
class AgencyLedgerTests(unittest.TestCase):
"""按真实形状建一份小账本：两家事务所、三位女优、几条链接。"""
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
root = Path(self.tmp.name).resolve()
self.db = fresh_ledger(root)
self.con = sqlite3.connect(self.db)
self.con.row_factory = sqlite3.Row
self.addCleanup(self.con.close)
self.con.executemany(
"INSERT INTO asset(id,location,path,name,medium,size,first_seen)"
" VALUES(?,'local',?,?,'video',100,'2026-01-01')",
[(1, r"R:\Media\a.mp4", "a.mp4"), (2, r"R:\Media\b.mp4", "b.mp4"),
(3, r"R:\Media\c.mp4", "c.mp4")])
self.con.executemany(
"INSERT INTO entity(id,kind,canonical_name,normalized_name,metadata_json,"
"created_at,updated_at) VALUES(?,?,?,?,?,'t','t')",
[(11, "performer", "七泽美亚", "七泽美亚",
'{"agency":{"name":"Capsule Agency","checked_at":"2026-09-04T00:00:00Z"}}'),
(12, "performer", "神雪", "神雪",
'{"agency":{"name":"Capsule Agency","checked_at":"2026-09-04T00:00:00Z"}}'),
(13, "performer", "白石亚子", "白石亚子",
'{"agency":{"name":"ACT(アクト)","checked_at":"2026-09-04T00:00:00Z"}}'),
(20, "tag", "足交", "足交", "{}")])
self.con.executemany(
"INSERT INTO asset_entity(asset_id,entity_id,role,source,confidence)"
" VALUES(?,?,?,'test',1.0)",
[(1, 11, "performer"), (2, 12, "performer"), (3, 13, "performer"),
(1, 20, "tag")])
# 归属域名的共识：三条同标签的链接指向同一个主机。
self.con.executemany(
"INSERT INTO entity_link(entity_id,link_kind,label,url,hostname,"
"created_at,updated_at) VALUES(?,'official','ACT(アクト)',?,"
"'actentertainment.jp','t','t')",
[(11, "https://actentertainment.jp/a"), (12, "https://actentertainment.jp/b"),
(13, "https://actentertainment.jp/c")])
self.con.commit()
rows = agency_plan(self.con)
install_agencies(self.con, rows, STAMP)
self.con.commit()
self.logos = root / "logos"; self.logos.mkdir()
self.avatars = root / "avatars"; self.avatars.mkdir()
self.contract = rm_web.WebContract(
self.db, avatar_root=self.avatars, logo_root=self.logos, cover_root=root / "covers")
def page(self, kind, name):
return rm_web.q_entity(self.contract, {"kind": kind, "name": name})
def photos(self, name, kind="agency"):
return rm_web.q_entity_photos(self.contract, {"kind": kind, "name": name})
def counted_once(self, photos):
"""总数、分页和图集数的是同一张图：各一次，图集字节数也只算一份。"""
self.assertEqual(
(photos["total"], [item["id"] for item in photos["items"]],
[(item["kind"], item["n"], item["bytes"]) for item in photos["sets"]]),
(1, [21], [("dir", 1, 10)]))
def add_pictures(self, rows):
"""(id, 路径, 挂在哪几位名下, disposal)。"""
for asset_id, path, owners, disposal in rows:
self.con.execute(
"INSERT INTO asset(id,location,path,name,medium,size,first_seen,disposal)"
" VALUES(?,'local',?,?,'image',10,'2026-01-01',?)",
(asset_id, path, path.rsplit("\\", 1)[-1], disposal))
self.con.executemany(
"INSERT INTO asset_entity(asset_id,entity_id,role,source,confidence)"
" VALUES(?,?,'performer','test',1.0)", [(asset_id, owner) for owner in owners])
self.con.commit()
def add_samples(self, codes):
"""作品 id → 番号，每个番号配一张官方样张。"""
for asset_id, code in codes.items():
self.con.execute("UPDATE asset SET code=? WHERE id=?", (code, asset_id))
self.con.execute(
"INSERT INTO code_sample_image(code,position,url,site,source,fetched_at)"
" VALUES(?,1,'https://pics.dmm.co.jp/x.jpg','dmm','test',?)", (code, STAMP))
self.con.commit()
def test_every_performer_with_an_agency_gets_exactly_one_membership(self):
self.assertEqual(
self.con.execute("SELECT count(*) FROM entity_membership").fetchone()[0], 3)
def test_the_agency_profile_counts_the_works_of_its_members(self):
page = self.page("agency", "Capsule Agency")
self.assertEqual(page["asset_count"], 2)
self.assertEqual(page["member_count"], 2)
def test_the_agency_profile_lists_its_roster_not_co_stars(self):
names = [person["k"] for person in self.page("agency", "Capsule Agency")
["related_performers"]]
self.assertEqual(sorted(names), ["七泽美亚", "神雪"])
def test_the_agency_profile_carries_the_tags_of_its_members_works(self):
self.assertIn("足交", [tag["k"] for tag in
self.page("agency", "Capsule Agency")["tags"]])
def test_the_agency_representative_is_picked_from_its_members_works(self):
"""ACT 那位的片更大，但不是这家的。"""
self.con.executemany("UPDATE asset SET snapshot_path='s.jpg',size=? WHERE id=?",
[(200, 1), (300, 2), (900, 3)])
self.con.commit()
self.assertEqual(self.page("agency", "Capsule Agency")["representative_asset_id"], 2)
def test_the_agency_photos_and_sample_sets_are_those_of_its_members(self):
"""事务所自己不挂图：图集、总数、分页和番号样张集都从成员名下取。
ACT 那位的图和片不算，回收站里的图也不算。
"""
self.add_pictures([
(21, r"R:\Media\summer\01.jpg", [11], None),
(22, r"R:\Media\summer\02.jpg", [12], None),
(23, r"R:\Media\winter\01.jpg", [12], None),
(24, r"R:\Media\summer\03.jpg", [11], "trash"),
(25, r"R:\Media\act\01.jpg", [13], None)])
self.add_samples({1: "SSIS-057", 2: "SSIS-058", 3: "IPX-001"})
photos = self.photos("Capsule Agency")
self.assertEqual([(item["kind"], item["id"], item["n"]) for item in photos["sets"]],
[("code", "code:SSIS-057", 1), ("code", "code:SSIS-058", 1),
("dir", 21, 2), ("dir", 23, 1)])
self.assertEqual((photos["total"], photos["sample_total"]), (3, 2))
self.assertEqual([item["id"] for item in photos["items"]], [21, 22, 23])
def test_a_picture_of_two_members_is_counted_and_listed_once(self):
self.add_pictures([(21, r"R:\Media\duo\01.jpg", [11, 12], None)])
self.counted_once(self.photos("Capsule Agency"))
def test_a_picture_filed_twice_under_one_entity_is_counted_once(self):
"""主键含 role 与 source：同一位名下，复核再写一行也还是同一张图。"""
self.add_pictures([(21, r"R:\Media\solo\01.jpg", [11], None)])
self.con.execute("INSERT INTO asset_entity(asset_id,entity_id,role,source,confidence)"
" VALUES(21,11,'performer','review',1.0)")
self.con.commit()
self.counted_once(self.photos("七泽美亚", "performer"))
def test_a_performer_profile_points_at_her_agency_entity(self):
home = self.page("performer", "七泽美亚")["agency"]
self.assertEqual(home["canonical_name"], "Capsule Agency")
self.assertEqual(home["checked_at"], "2026-09-04T00:00:00Z")
def test_an_agency_profile_has_no_agency_of_its_own(self):
self.assertIsNone(self.page("agency", "Capsule Agency")["agency"])
def test_the_agency_is_reachable_by_its_former_name(self):
"""`ACT(アクト)` 的读音留成别名，按它也要能打开这一页。"""
self.assertEqual(self.page("agency", "アクト")["canonical_name"], "ACT")
def test_the_agency_keeps_its_own_official_site(self):
page = self.page("agency", "ACT")
self.assertEqual([link["url"] for link in page["links"]],
["https://actentertainment.jp/"])
def test_agency_is_a_profile_kind(self):
self.assertIn("agency", PROFILE_KINDS)
def test_an_unknown_kind_is_still_refused(self):
self.assertEqual(rm_web.q_entity(self.contract, {"kind": "label", "name": "x"}),
{"error": "invalid entity"})
def test_the_catalog_can_be_filtered_by_agency(self):
items = rm_web.q_items(self.contract, {"agency": "Capsule Agency", "limit": "10"})
self.assertEqual(sorted(item["id"] for item in items["items"]), [1, 2])
def test_searching_an_agency_name_finds_the_works_of_its_members(self):
"""用户搜「Capsule」时，账本里确实有它，只是此前没有一条路把它接上作品。"""
items = rm_web.q_items(self.contract, {"q": "Capsule", "limit": "10"})
self.assertEqual(sorted(item["id"] for item in items["items"]), [1, 2])
def test_a_short_query_reaches_the_agency_too(self):
"""短查询走 LIKE 分支，两条分支要覆盖同样的身份写法。"""
items = rm_web.q_items(self.contract, {"q": "AC", "limit": "10"})
self.assertEqual([item["id"] for item in items["items"]], [3])
def test_agency_facets_come_from_the_same_rows_as_its_items(self):
facets = rm_web.q_facets(self.contract, scope_kind="agency",
scope_name="Capsule Agency")
self.assertEqual(sum(row["n"] for row in facets["locations"]), 2)
def index(self, kind, q=""):
return {row["k"]: row for row in
rm_web.q_index(self.contract, kind, q=q, limit=50)["items"]}
def test_the_agency_index_lists_every_agency_by_headcount(self):
"""事务所自己不挂作品，一家有多少艺人才是它的规模。"""
rows = self.index("agencies")
self.assertEqual(sorted(rows), ["ACT", "Capsule Agency"])
self.assertEqual(rows["Capsule Agency"]["members"], 2)
self.assertEqual(rows["ACT"]["members"], 1)
def test_the_agency_index_counts_works_the_same_way_the_profile_does(self):
"""判据只有 `scope_predicate` 一份，两个页面数出来的才是同一个数。"""
rows = self.index("agencies")
for name in rows:
with self.subTest(agency=name):
self.assertEqual(rows[name]["n"], self.page("agency", name)["asset_count"])
def test_the_agency_index_hands_out_its_own_mark_not_a_members_frame(self):
"""索引页那格图取的是官网圆标：作品截图是某位成员某部片的画面。"""
rows = self.index("agencies")
self.assertIsNotNone(rows["ACT"]["mark"])
self.assertFalse(rows["ACT"]["has_avatar"])
def test_the_agency_index_can_be_filtered_by_name(self):
self.assertEqual(sorted(self.index("agencies", q="Capsule")), ["Capsule Agency"])
def test_both_halves_of_the_maker_switch_have_an_index(self):
"""开关的两半都得有索引页，否则那个开关有一边是死的。"""
for kind in ("studios", "agencies"):
with self.subTest(kind=kind):
self.assertEqual(rm_web.q_index(self.contract, kind, limit=5)["kind"], kind)
def test_the_agency_asks_whether_its_logo_is_installed(self):
"""标识按名字落盘，厂牌和事务所同一个仓：两边都得先问过再出图。"""
self.assertIn("has_logo", self.page("agency", "ACT"))
self.assertIn("has_logo", self.index("agencies")["ACT"])
(self.logos / "ACT.img").write_bytes(b"x")
self.contract = rm_web.WebContract(
self.db, avatar_root=self.avatars, logo_root=self.logos)
self.assertTrue(self.page("agency", "ACT")["has_logo"])
def test_the_agency_profile_hands_out_the_link_id_for_its_mark(self):
page = self.page("agency", "ACT")
self.assertEqual(page["mark_link_id"],
next(link["link_id"] for link in page["links"]))
def test_an_archived_official_site_is_listed_but_not_used_as_the_mark(self):
"""关门事务所的官网只剩存档快照：链接照列，门面不取存档站的圆标。"""
agency_id = self.con.execute(
"SELECT id FROM entity WHERE kind='agency' AND canonical_name='Capsule Agency'"
).fetchone()[0]
self.con.execute("DELETE FROM entity_link WHERE entity_id=?", (agency_id,))
self.con.execute(
"INSERT INTO entity_link(entity_id,link_kind,label,url,hostname,created_at,updated_at)"
" VALUES(?,'official','官网存档（2020-01）',"
"'https://web.archive.org/web/20200101000000/https://capsule.bz/','web.archive.org',"
"'t','t')", (agency_id,))
self.con.commit()
page = self.page("agency", "Capsule Agency")
self.assertEqual([link["hostname"] for link in page["links"]], ["web.archive.org"])
self.assertIsNone(page["mark_link_id"])
self.assertIsNone(self.index("agencies")["Capsule Agency"]["mark"])
def test_the_agency_roster_carries_the_focus_the_big_layout_needs(self):
"""名册摆的是竖幅大图，几何居中会切掉脸，取景要随资料一起下发。"""
roster = self.page("agency", "Capsule Agency")["related_performers"]
self.assertTrue(roster)
for person in roster:
with self.subTest(person=person["k"]):
self.assertIn("avatar_focus", person)
def test_the_scope_predicate_takes_one_placeholder_either_way(self):
self.assertEqual(scope_predicate("performer", "ae.entity_id").count("?"), 1)
self.assertEqual(scope_predicate("agency", "ae.entity_id").count("?"), 1)
def test_the_index_scope_predicate_reads_the_agency_from_each_row(self):
"""索引页一句 SQL 数完所有事务所，判据那一侧填的是列名不是占位符。"""
self.assertIn("agency_id=e.id", scope_predicate("agency", "ae.entity_id", "e.id"))
def test_merging_a_performer_carries_her_membership_over(self):
self.con.execute(
"INSERT INTO entity(id,kind,canonical_name,normalized_name,created_at,updated_at)"
" VALUES(99,'performer','Mia Nanasawa','mia nanasawa','t','t')")
moved = merge_entity(self.con, target_id=99, source_id=11,
source_name="七泽美亚", alias_source="merge:test")
# 资料页读的是另一条只读连接，不提交就还看得见合并前的账本。
self.con.commit()
self.assertEqual(moved["memberships"], 1)
self.assertEqual(self.con.execute(
"SELECT count(*) FROM entity_membership WHERE member_id=11").fetchone()[0], 0)
self.assertEqual(self.page("agency", "Capsule Agency")["member_count"], 2)
def test_merging_two_agencies_moves_the_roster(self):
target = self.con.execute(
"SELECT id FROM entity WHERE kind='agency' AND canonical_name='ACT'").fetchone()[0]
source = self.con.execute(
"SELECT id FROM entity WHERE kind='agency' AND canonical_name='Capsule Agency'"
).fetchone()[0]
moved = merge_entity(self.con, target_id=target, source_id=source,
source_name="Capsule Agency", alias_source="merge:test")
self.con.commit()
self.assertEqual(moved["members"], 2)
self.assertEqual(self.page("agency", "ACT")["member_count"], 3)
def test_duplicate_links_on_one_entity_collapse_to_the_https_one(self):
self.con.executemany(
"INSERT INTO entity_link(entity_id,link_kind,label,url,hostname,"
"created_at,updated_at) VALUES(11,'social','X',?,'x.com','t','t')",
[("https://x.com/mia",), ("https://x.com/mia/status/1",)])
self.con.execute(
"INSERT INTO entity_link(entity_id,link_kind,label,url,hostname,"
"created_at,updated_at) VALUES(12,'official','ACT',"
"'http://actentertainment.jp/b','actentertainment.jp','t','t')")
self.con.commit()
verdicts = {row["url"]: row["verdict"] for row in dedupe_plan(self.con)}
self.assertEqual(verdicts["https://x.com/mia"], "keep")
self.assertEqual(verdicts["https://x.com/mia/status/1"], "delete")
self.assertEqual(verdicts["https://actentertainment.jp/b"], "keep")
self.assertEqual(verdicts["http://actentertainment.jp/b"], "delete")
class DeepLinkRouteTests(unittest.TestCase):
"""资料页的地址得能直接打开，而不只是在应用里点得到。"""
def test_every_entity_route_is_served_as_a_page(self):
"""前端认得的每一种实体路由，服务端都要发页面。
少一条不会报错，只会在直接打开地址时回 404：应用内点进去照常，刷新一下就没了。
所以判据取自前端 core 的路由表本身，加一种实体就自动多一条要求。
"""
routes = re.search(r"const ENTITY_ROUTES(?::[^=]*)?=\{([^}]*)\}",
(ROOT / "frontend" / "src" / "core" / "index.ts").read_text(encoding="utf-8"))
self.assertIsNotNone(routes, "frontend/src/core/index.ts 里没有 ENTITY_ROUTES")
wanted = re.findall(r"[:,]\s*'([a-z]+)'", "," + routes.group(1))
self.assertIn("agencies", wanted, "路由表没解析出来")
pages = (ROOT / "src" / "peach" / "routes_pages.py").read_text(encoding="utf-8")
for route in wanted:
with self.subTest(route=route):
self.assertIn(f'"/{route}/{{name:path}}"', pages)
if __name__ == "__main__":
unittest.main()