tests/test_module_layering.py
Rama master · réplica de lectura del repositorio original.
"""模块分层门槛。
Peach 是模块化单体，分层靠约定而不是包结构，所以约定必须有人守。这里守一条：
**web 层之外的模块不许 import web 层。**
这条规则不是凭空立的。`catalog_rules.py` 原名 `web_logic.py`，而 `repository.py`
（数据层）要用它的 `is_jav_code`，`taste_history.py` 要用 `LENGTH_TAGS`——数据层
import 一个叫 web 的模块，读代码的人只会得出「分层反了」的结论。实际反的是名字，
但名字骗人和结构错了一样难查。改名之后加这条断言，让下一次真反了的时候直接报错。
"""
import ast
import importlib
import pathlib
import tempfile
import typing
import unittest
import peach
from peach.web_contract import WebContract
SOURCE_ROOT = pathlib.Path(peach.__file__).parent
#: web 层自己。层内互相依赖是正常的。
#:
#: 按文件名推导而不是手写清单。写死一份清单的话，拆分出 `web_review`、
#: `web_resource_sync`、`web_settings` 之后没人回来更新，那三个模块就一直不设防。
#: 守规则的东西自己先漂了，是这条门槛最容易出的故障。
WEB_MODULES = {path.stem for path in SOURCE_ROOT.glob("web_*.py")}
#: FastAPI 适配层：`api` 是组装点，`routes_*` 是它挂上去的路由表。把 web 层接到
#: HTTP 上是这一层的职责，方向是对的。
#:
#: 和 `WEB_MODULES` 同样按文件名推导。只写着 `api` 的话，`create_app` 拆成
#: `routes_auth`/`routes_pages`/`routes_media`/`routes_api` 之后再拆一个模块就会被判成
#: 「非 web 模块 import 了 web 层」——门槛报的是假警，改的人只会来把它加进清单。
ROUTE_MODULES = {path.stem for path in SOURCE_ROOT.glob("routes_*.py")}
COMPOSITION_ROOTS = {"api"} | ROUTE_MODULES
def _local_imports(path: pathlib.Path) -> set[str]:
"""取出一个模块 import 的本包模块名（`from .x import y` 与 `from . import x`）。"""
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
names: set[str] = set()
for node in ast.walk(tree):
if isinstance(node, ast.ImportFrom) and node.level:
if node.module:
names.add(node.module.split(".")[0])
else:
names.update(alias.name for alias in node.names)
return names
#: 项目根：`src/peach` 的上两级。脚本和源码要一起扫。
PROJECT_ROOT = SOURCE_ROOT.parent.parent
def _python_sources() -> list[pathlib.Path]:
return sorted(SOURCE_ROOT.glob("*.py")) + sorted((PROJECT_ROOT / "scripts").glob("*.py"))
class SharedRuleTests(unittest.TestCase):
"""领域规则只许有一份实现。
番号归一化最容易长出第二份：`catalog_rules.normalise_code_key`、
`scripts/fetch_jav_covers.normalise_code`、`scripts/scrape_codes.normalise`
三处逐字相同过，靠一句「与 scrape_codes 同口径」的注释维持一致——注释不是门槛。
番号既是身份判定也是封面缓存键：三份一旦漂移，同一部片会解析出两个键，
封面缓存静默失配，而两边的数都「对」，只是口径不同。
"""
#: 归一化实现的指纹。`FC2-PPV-` 的拼装和番号形态的正则，凑齐就是又抄了一份。
FINGERPRINTS = ('f"FC2-PPV-', r'([A-Z]+)-?(\d+)$')
def test_release_code_normalisation_exists_once(self):
offenders = []
for path in _python_sources():
if path.name == "catalog_rules.py":
continue
text = path.read_text(encoding="utf-8")
if all(mark in text for mark in self.FINGERPRINTS):
offenders.append(path.name)
self.assertEqual(
offenders, [],
"番号归一化又被抄了一份，请改 import catalog_rules.normalise_code_key",
)
def test_the_review_csv_encoding_is_declared_once(self):
"""复核 CSV 的编码只许在一处写死。
AGENTS.md 把 CSV 定为「复核产物」：可机读、可重放。它的编码因此不是风格问题——
`utf-8-sig` 的 BOM 决定 Excel 打开时中文是不是乱码，`newline=""` 决定 Windows
下行间会不会多出空行。两条都属于「写的时候一切正常、几天后有人真去开那份表才
发现坏了」，靠人眼复查拦不住。
这两条此前在 46 个读写点各写一遍，而且已经开始藏：有两处把参数顺序写成
`"w", newline="", encoding="utf-8-sig"`，按规范顺序 grep 根本扫不到。所以这里
断言的是字面量本身只出现一次，而不是某个固定写法。
"""
offenders = []
for path in _python_sources():
if path.name == "review_csv.py":
continue
if '"utf-8-sig"' in path.read_text(encoding="utf-8"):
offenders.append(path.name)
self.assertEqual(
offenders, [],
"编码不要再写一遍：用 peach.review_csv 的 read_rows/write_rows，"
"确实只能自己开文件的（流式写）就用它的 ENCODING 常量",
)
def test_batch_scripts_share_one_cli_tail(self):
"""批处理脚本的 main() 收尾只许有一份。
它此前抄了四份。共用的不只是形状，还有「pid 锁必须包住整个 run」和「策略
异常交出自己的退出码而不是抛栈」这两条约定；散成四份，下一个脚本照抄时漏掉
锁或吞掉退出码都不会有人发现。
"""
offenders = [path.name for path in sorted((PROJECT_ROOT / "scripts").glob("*.py"))
if "with PidFileLock(args.lock):" in path.read_text(encoding="utf-8")]
self.assertEqual(offenders, [],
"又有脚本自己写了 main() 收尾，请改调 peach.jobs.job_main")
def test_pure_rules_are_imported_from_the_policy_module_not_through_web(self):
"""纯规则要直接从 `catalog_rules` 取，不要借道 web 层的再导出。
`web_contract` 把 `is_jav_code` 再导出了一遍，于是脚本写成
`from peach.web_contract import is_jav_code`——为一条纯规则拉起整个 web 层，
读代码的人还会以为这条规则属于 web。
"""
offenders = []
for path in sorted((PROJECT_ROOT / "scripts").glob("*.py")):
text = path.read_text(encoding="utf-8")
for rule in ("is_jav_code", "normalise_code_key", "LENGTH_TAGS"):
if "from peach.web_contract import" in text and rule in text.split(
"from peach.web_contract import", 1)[1].split("\n", 1)[0]:
offenders.append(f"{path.name} → web_contract.{rule}")
self.assertEqual(offenders, [],
"纯规则请直接从 peach.catalog_rules 取")
class ContractConformanceTests(unittest.TestCase):
"""域模块声明的窄契约，真的 `WebContract` 必须全部满足。
`Protocol` 在运行期什么都不检查，域模块又只照着自己那份 Protocol 写调用，
于是「契约上写了、实现里没有」不会有任何东西报错，一路要等到用户点那个按钮。
实测事故：`web_links.LinkContract` 声明了 `write_connection`，`WebContract` 只有
`write_transaction`，`/api/links/prune` 在线上直接 500；而 `tests/test_web_links.py`
的 FakeContract 恰好把两个名字都给了，测试因此全绿。这条门槛就是补那次的窗口。
"""
@staticmethod
def _declared(protocol) -> set[str]:
"""一份 Protocol 声明的能力：带注解的属性 + 协议体里定义的方法。"""
names = set(typing.get_type_hints(protocol))
names.update(name for name, value in vars(protocol).items()
if callable(value) and not name.startswith("_"))
return names
@staticmethod
def _web_protocols() -> list[type]:
"""web 层各模块自己定义的 Protocol，按文件名发现而不是手写清单。"""
found: list[type] = []
for path in sorted(SOURCE_ROOT.glob("web_*.py")):
module = importlib.import_module(f"peach.{path.stem}")
for obj in vars(module).values():
if (isinstance(obj, type) and getattr(obj, "_is_protocol", False)
and obj.__module__ == module.__name__ and obj not in found):
found.append(obj)
return found
def test_every_web_protocol_is_satisfied_by_the_real_contract(self):
protocols = self._web_protocols()
self.assertGreaterEqual(len(protocols), 5, "没发现域契约，门槛已空转")
with tempfile.TemporaryDirectory() as tmp:
# 构造只是赋值路径和建锁，不碰磁盘也不连库；临时目录仅用于坐实这一点。
contract = WebContract(pathlib.Path(tmp).resolve() / "ledger.db")
missing = [f"{protocol.__name__}.{name}"
for protocol in protocols
for name in sorted(self._declared(protocol))
if not hasattr(contract, name)]
self.assertEqual(
missing, [],
"域契约声明了 WebContract 没有的能力：要么补上实现，要么改契约声明",
)
class LayeringTests(unittest.TestCase):
def test_the_web_layer_is_actually_discovered(self):
"""推导出来的 web 层不能是空的。
改成按文件名推导之后，清单不会再漂；但万一命名约定变了，glob 会安静地返回
空集，上面那条断言就永远成立——门槛还在，只是不再拦任何东西。
"""
self.assertGreaterEqual(len(WEB_MODULES), 4, "没发现 web 层，门槛已空转")
self.assertIn("web_contract", WEB_MODULES)
def test_the_route_layer_is_actually_discovered(self):
"""`routes_*` 也不能是空的，理由同上：它现在参与放行判定。
`COMPOSITION_ROOTS` 一旦被 glob 推成只有 `api`，上面那条 web 层门槛会立刻
对四个路由模块报假警；反过来，如果有人改名让 glob 全落空，放行集合缩小，
门槛会变严而不是变松——但报出来的偏移点是错的，所以这里直接钉住。
"""
self.assertGreaterEqual(len(ROUTE_MODULES), 4, "没发现路由层，放行集合已漂")
self.assertIn("routes_api", ROUTE_MODULES)
def test_nothing_below_the_route_layer_imports_it(self):
"""路由层是最上面一层：只有 `api` 和它自己的同层模块可以 import `routes_*`。
方向必须单向。`routes_media` 里放着「取哪个文件、报哪个缓存头」这类 HTTP
决策，一旦被 `media`、`previews` 之类的下层 import，FastAPI 就渗进了不需要
知道 HTTP 的地方，测试也从此必须先建一个 app 才能跑。
"""
offenders = []
for path in sorted(SOURCE_ROOT.glob("*.py")):
module = path.stem
if module in COMPOSITION_ROOTS:
continue
for imported in sorted(_local_imports(path)):
if imported in ROUTE_MODULES:
offenders.append(f"{module} → {imported}")
self.assertEqual(
offenders, [],
"只有 api 与 routes_* 自己可以依赖路由层：需要共享的东西下沉到领域模块",
)
def test_only_web_modules_and_the_composition_root_import_the_web_layer(self):
offenders = []
for path in sorted(SOURCE_ROOT.glob("*.py")):
module = path.stem
if module in WEB_MODULES or module in COMPOSITION_ROOTS:
continue
for imported in sorted(_local_imports(path)):
if imported in WEB_MODULES:
offenders.append(f"{module} → {imported}")
self.assertEqual(
offenders, [],
"非 web 模块不得依赖 web 层：把共享的纯规则下沉到 catalog_rules 一类的策略模块",
)
def test_route_modules_only_depend_on_the_auth_routes(self):
"""路由层内部也是单向的：`routes_auth` 在最下面，别的 `routes_*` 只许 import 它。
登录、首启与错误三页共用的薄壳 `page_shell` 和配置页要的运行信息住在 web 层（`web_entry`），
`same_origin` 住在 `routes_auth`。共享件放进任何一个 `routes_*`，别的路由模块就得反向去拿，
成一个靠函数内 import 撑住的环——import 图上看不出来，读代码的人也说不清谁在谁上面。
函数内的 import 一样算：`_local_imports` 扫整棵语法树，不只看文件顶部。
"""
offenders = []
for path in sorted(SOURCE_ROOT.glob("routes_*.py")):
module = path.stem
allowed = set() if module == "routes_auth" else {"routes_auth"}
for imported in sorted(_local_imports(path) & ROUTE_MODULES - allowed):
offenders.append(f"{module} → {imported}")
self.assertEqual(
offenders, [],
"routes_* 之间只许依赖 routes_auth：两个路由都要的东西下沉到 web_* 模块",
)
def test_catalog_rules_depends_on_nothing_inside_peach(self):
"""最底下那层必须是纯的：只要它开始 import 别的 Peach 模块，就不再是策略了。"""
self.assertEqual(
_local_imports(SOURCE_ROOT / "catalog_rules.py"), set(),
"catalog_rules 是最底层纯规则，不能依赖任何 Peach 模块",
)
def test_the_old_web_logic_name_is_gone(self):
self.assertFalse(
(SOURCE_ROOT / "web_logic.py").exists(),
"web_logic.py 已改名为 catalog_rules.py；两个同时存在意味着改名没做干净",
)
# 只看 import，不看全文：`catalog_rules` 的 docstring 要讲清楚它改过名，
# 那句话里出现旧名字是应该的。
for path in sorted(SOURCE_ROOT.glob("*.py")):
self.assertNotIn(
"web_logic", _local_imports(path),
f"{path.name} 仍在 import 旧模块名 web_logic",
)
if __name__ == "__main__":
unittest.main()