tests/test_content_region.py
Rama master · réplica de lectura del repositorio original.
"""内容产地：韩国与国产的番号不能再算进 JAV。
这一组用例盯的是同一个口径在三处成立：`regions.infer_region` 的推断本身、
`is_jav_asset` 的短路、以及 SQL 那份 `EFFECTIVE_REGION_SQL`。分开写是因为出事的
正是它们分家的那一刻——`is_korean_mib_code` 早就认得出 MIB，可分类口径从来没读过它。
"""
from __future__ import annotations
import sqlite3
import tempfile
import unittest
from pathlib import Path
from peach import web_batch
from peach import web_contract as rm_web
from peach.catalog_rules import is_jav_asset
from peach.field_owners import USER_MANUAL, owner_of
from peach.regions import (
REGIONS,
infer_region,
is_region,
normalize_region,
region_label,
)
from support.ledger import fresh_ledger
class RegionValueTests(unittest.TestCase):
def test_every_region_has_a_label(self):
self.assertEqual(sorted(REGIONS), sorted(r for r in REGIONS if region_label(r)))
def test_unknown_is_not_other(self):
"""「还没判过」和「判过，不属于那四类」必须分得开，否则待办数永远算不出。"""
self.assertTrue(is_region(""))
self.assertEqual(normalize_region(""), "")
self.assertEqual(normalize_region("other"), "other")
self.assertNotEqual(normalize_region("other"), normalize_region(""))
def test_an_unrecognised_value_reads_as_undecided(self):
"""认不出的写法退回未判定而不是抛错：它会从查询参数和候选两个方向进来。"""
self.assertEqual(normalize_region("jpn"), "")
self.assertEqual(normalize_region(None), "")
self.assertFalse(is_region("jpn"))
def test_case_and_padding_do_not_make_a_new_region(self):
self.assertEqual(normalize_region(" JP "), "jp")
class RegionInferenceTests(unittest.TestCase):
def test_korean_mib_codes_are_korean(self):
for code in ("CA-103", "DB-205", "CHU-201", "ERI-102"):
with self.subTest(code=code):
self.assertEqual(infer_region(code), "kr")
def test_mainland_studio_codes_are_domestic(self):
for code in ("MDX-0123", "MD-0140", "TM-0087", "JD-0031", "SWAG-1020"):
with self.subTest(code=code):
self.assertEqual(infer_region(code), "cn")
def test_japanese_codes_stay_undecided(self):
"""日本不靠推断落下来。把「剩下的都算日本」写成规则，等于让每个没收录的
国产厂牌自动变回 JAV——那正是这次要修的毛病。"""
for code in ("MEYD-911", "ABP-762", "SSIS-001", "FC2-PPV-1234567"):
with self.subTest(code=code):
self.assertEqual(infer_region(code), "")
def test_a_real_jav_studio_sharing_a_two_letter_shape_is_not_korean(self):
"""`BF-366` 在 `B:\\番号\\BeFree\\` 下，BeFree 是真实 JAV 厂牌。"""
self.assertEqual(infer_region("BF-366"), "")
def test_western_release_shapes_are_western(self):
self.assertEqual(infer_region(None, "Vixen.26.05.07.Ella.Rose.mp4"), "west")
self.assertEqual(infer_region("", "DorcelClub.2024.12.02.Christy.White.mp4"), "west")
def test_amateur_numeric_prefixes_keep_their_release_system(self):
"""`300MIUM-1239` 的字母段撞得上名单，但它的发行体系本来就是日本素人系。"""
self.assertEqual(infer_region("300MIUM-1239"), "")
class JavPredicateTests(unittest.TestCase):
def test_a_settled_non_japanese_region_is_never_jav(self):
for region in ("kr", "cn", "west", "other"):
with self.subTest(region=region):
self.assertFalse(is_jav_asset(
"CA-103", studio="MIB", release_date="2022-10-28", region=region))
def test_an_undecided_region_keeps_the_old_rule(self):
"""这一列绝大多数行还是空的，拿「没判过」当「不是日本」会清空整个 JAV 页。"""
self.assertTrue(is_jav_asset("MEYD-911", studio="MOODYZ", region=""))
self.assertTrue(is_jav_asset("MEYD-911", studio="MOODYZ"))
def test_japan_still_needs_release_evidence(self):
"""产地是日本也救不了没有发行证据的 creator clip。"""
self.assertFalse(is_jav_asset("JI-103", region="jp"))
self.assertTrue(is_jav_asset("JI-103", studio="Some Studio", region="jp"))
def test_fc2_stays_jav_under_japan(self):
self.assertTrue(is_jav_asset("FC2-PPV-1234567", region="jp"))
class RegionLedgerTests(unittest.TestCase):
"""SQL 那一份口径。产地不写进账本就得每次查询时算，那就只能有一处算法。"""
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.db_path = str(fresh_ledger(self.tmp.name))
con = sqlite3.connect(self.db_path)
con.executemany(
"INSERT INTO asset(id,location,path,name,medium,size,studio,code,"
"release_date,region,first_seen) VALUES(?,?,?,?,'video',10,?,?,?,?,'2026-09-13')",
[
# 韩国 MIB：官网刮回来的厂牌和发行日让它满足了 JAV 的全部旧判据。
(1, "115", r"B:\MVP\MIB\CA-103.mp4", "CA-103 Chaeah.mp4",
"MIB", "CA-103", "2022-10-28", None),
(2, "115", r"B:\番号\MEYD-911.mp4", "MEYD-911.mp4",
"MOODYZ", "MEYD-911", "2024-01-01", None),
(3, "115", r"B:\国产\MDX-0123.mp4", "MDX-0123.mp4",
"麻豆传媒", "MDX-0123", "2023-05-05", None),
# 番号推不出产地，靠创作者实体投影。
(4, "115", r"B:\创作者\sunwall\clip.mp4", "clip.mp4", None, None, None, None),
# 用户判定压过番号推断：这条番号形如韩国 MIB，但用户说是日本。
(5, "115", r"B:\番号\AR-101.mp4", "AR-101.mp4",
"Some Studio", "AR-101", "2021-01-01", "jp"),
],
)
con.execute(
"INSERT INTO entity(id,kind,canonical_name,normalized_name,region,"
"created_at,updated_at) VALUES(50,'creator','sunwall','sunwall','kr','t','t')")
con.execute("INSERT INTO asset_entity(asset_id,entity_id,role,source) "
"VALUES(4,50,'creator','test')")
con.commit()
con.close()
self.contract = rm_web.WebContract(Path(self.db_path))
def _regions(self):
rows = rm_web.q_items(self.contract, {"count": "0", "limit": "50"})["items"]
return {row["id"]: row["region"] for row in rows}
def _jav_ids(self):
rows = rm_web.q_items(self.contract, {"jav": "1", "count": "0", "limit": "50"})["items"]
return sorted(row["id"] for row in rows)
def test_korean_and_domestic_codes_leave_the_jav_page(self):
self.assertEqual(self._jav_ids(), [2, 5])
def test_each_layer_wins_over_the_one_below_it(self):
self.assertEqual(self._regions(), {
1: "kr", # 番号推断
2: "", # 日本不靠推断，等人判
3: "cn", # 番号推断
4: "kr", # 创作者实体投影
5: "jp", # 用户判定压过推断
})
def test_the_settled_flag_separates_a_decision_from_a_guess(self):
rows = {row["id"]: row for row in
rm_web.q_items(self.contract, {"count": "0", "limit": "50"})["items"]}
self.assertTrue(rows[5]["region_settled"])
self.assertFalse(rows[1]["region_settled"])
def test_the_detail_page_agrees_with_the_list(self):
"""详情页那条查询没给 asset 起别名，产地表达式必须跟着换写法。"""
detail = rm_web.q_item(self.contract, 1)
self.assertEqual(detail["region"], "kr")
self.assertFalse(detail["is_jav"])
def test_filtering_by_region(self):
rows = rm_web.q_items(self.contract, {"region": "kr", "count": "0", "limit": "50"})
self.assertEqual(sorted(row["id"] for row in rows["items"]), [1, 4])
def test_filtering_several_regions_at_once(self):
rows = rm_web.q_items(self.contract, {"region": "kr,cn", "count": "0", "limit": "50"})
self.assertEqual(sorted(row["id"] for row in rows["items"]), [1, 3, 4])
def test_filtering_the_undecided_pile(self):
"""判产地是人一条条过的活，看不到还剩多少就没人会去过。"""
rows = rm_web.q_items(self.contract, {"region": "none", "count": "0", "limit": "50"})
self.assertEqual(sorted(row["id"] for row in rows["items"]), [2])
def test_the_facet_counts_each_bucket(self):
facets = {row["k"]: row for row in rm_web.q_facets(self.contract)["regions"]}
self.assertEqual({key: row["n"] for key, row in facets.items()},
{"kr": 2, "cn": 1, "jp": 1, "none": 1})
self.assertEqual(facets["none"]["label"], "未判定")
self.assertEqual(facets["kr"]["label"], "韩国")
def test_the_facet_follows_the_other_filters(self):
"""筛选项必须和作品列表同源，否则右栏会给出点进去是 0 条的项。"""
facets = rm_web.q_facets(self.contract, filters={"region": "kr"})["regions"]
self.assertEqual({row["k"] for row in facets}, {"kr"})
class RegionBatchTests(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.db_path = str(fresh_ledger(self.tmp.name))
con = sqlite3.connect(self.db_path)
con.executemany(
"INSERT INTO asset(id,location,path,name,medium,size,code,first_seen) "
"VALUES(?,?,?,?,'video',10,?,'2026-09-13')",
[(1, "115", r"B:\a.mp4", "a.mp4", "AAA-001"),
(2, "115", r"B:\b.mp4", "b.mp4", "BBB-002")],
)
con.commit()
con.close()
self.contract = rm_web.WebContract(Path(self.db_path))
def _stored(self):
con = sqlite3.connect(self.db_path)
try:
return {row[0]: (row[1], owner_of(row[2], "region")) for row in
con.execute("SELECT id,region,field_owners FROM asset ORDER BY id")}
finally:
con.close()
def test_a_batch_decision_is_owned_by_the_user(self):
web_batch.w_batch(self.contract, {"ids": [1, 2], "operation": "region", "region": "cn"})
self.assertEqual(self._stored(), {1: ("cn", USER_MANUAL), 2: ("cn", USER_MANUAL)})
def test_none_withdraws_the_decision(self):
web_batch.w_batch(self.contract, {"ids": [1], "operation": "region", "region": "kr"})
web_batch.w_batch(self.contract, {"ids": [1], "operation": "region", "region": "none"})
self.assertEqual(self._stored()[1], (None, USER_MANUAL))
def test_a_misspelt_region_is_refused_rather_than_clearing_the_column(self):
"""`normalize_region` 把认不出的写法当未判定，批量这一路不能跟着它——
一次拼错就成了一次静默的清空。"""
web_batch.w_batch(self.contract, {"ids": [1], "operation": "region", "region": "kr"})
with self.assertRaises(ValueError):
web_batch.w_batch(self.contract, {"ids": [1], "operation": "region", "region": "jpn"})
self.assertEqual(self._stored()[1], ("kr", USER_MANUAL))
if __name__ == "__main__":
unittest.main()