tests/test_script_policy.py
Rama master · réplica de lectura del repositorio original.
"""`scripts/` 的两条硬门槛，按 AST 判而不是按源码字符串判。
写在测试里而不是文档里，是因为这两件事都反复回归过：路径写死过一轮又一轮，
写库脚本也出现过「跑起来就直接改库」的形态。文档里的提醒不会拦住任何一次提交。
判据落在 AST 上而不是 `grep`：注释和文档串里出现 `R:\\peach-data` 是在说明历史，
不是在读它；反过来，`"UPDATE asset SET ..."` 拼在多行字符串里 grep 也照样能漏。
"""
from __future__ import annotations
import ast
import re
import subprocess
import sys
import unittest
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
SCRIPTS = ROOT / "scripts"
SRC = ROOT / "src" / "peach"
#: Peach 数据目录的写死形态。搬到内置盘之后 `R:\\peach-data` 已经不存在，而写死的路径不会报
#: 「配置过时」——它只会安静地建一个空库，或者往一个没人看的目录里写日志。用户主目录
#: 下的具体位置同样算写死：它对任何别的用户都不成立。
DATA_ROOT_LITERAL = re.compile(
r"R:[\\/]peach-data|[\\/](?:Users|home)[\\/][^\\/]+[\\/]Desktop", re.I)
#: 只有这两个模块可以谈论平台与 Peach 数据目录的具体位置，其余一律从它们取。
PATH_AUTHORITIES = {"config.py", "platform.py"}
#: 门槛（b）的例外，逐个写清理由。
DATA_ROOT_ALLOWLIST: set[str] = set()
DML = re.compile(r"\b(INSERT|UPDATE|DELETE|REPLACE)\b", re.I)
DML_CLAUSE = re.compile(r"\b(INTO|FROM|SET)\b", re.I)
#: 门槛（c）的例外，逐个写清理由。
#:
#: 判据是「这个脚本是不是把既有账本行按复核结论改掉」。是的话必须有 `--apply`：
#: 复核产物和真实写入之间必须有一道人为闸门。下面这些不是那种形态。
APPLY_ALLOWLIST = {
# 常驻批处理，由 `peach.jobs.job_main` + PidFileLock 驱动，没有「先出复核表再决定
# 写不写」这一步：它们写的是自己刚生成的探测结果与九宫格产物。停不停由任务控制，
# 不由一个命令行开关控制。
"probe.py",
"sheets.py",
# 摄取入口。它新增行而不是按复核结论改既有行，整个用途就是写：把 115 服务端
# 算好的 SHA1 灌进来。目前还没有备份闸门，是后续该补的，不是本门槛要管的形态。
"sync_sha1_115.py",
}
def _docstring_nodes(tree: ast.AST) -> set[int]:
"""模块／类／函数的文档串节点。它们是叙述，不是被执行的值。"""
found: set[int] = set()
for node in ast.walk(tree):
if not isinstance(node, (ast.Module, ast.ClassDef,
ast.FunctionDef, ast.AsyncFunctionDef)):
continue
body = getattr(node, "body", None)
if (body and isinstance(body[0], ast.Expr)
and isinstance(body[0].value, ast.Constant)
and isinstance(body[0].value.value, str)):
found.add(id(body[0].value))
return found
def _live_strings(path: Path) -> list[tuple[int, str]]:
"""去掉文档串之后，源码里真正会被当值用的字符串常量。"""
tree = ast.parse(path.read_text(encoding="utf-8"))
skip = _docstring_nodes(tree)
return [(node.lineno, node.value) for node in ast.walk(tree)
if isinstance(node, ast.Constant) and isinstance(node.value, str)
and id(node) not in skip]
def _apply_flags(path: Path) -> set[str]:
"""脚本声明过的 `--apply*` 开关，含由 `add_ledger_write_args` 代为声明的。"""
tree = ast.parse(path.read_text(encoding="utf-8"))
flags: set[str] = set()
for node in ast.walk(tree):
if not isinstance(node, ast.Call):
continue
name = (node.func.attr if isinstance(node.func, ast.Attribute)
else node.func.id if isinstance(node.func, ast.Name) else "")
if name == "add_ledger_write_args":
flags.add("--apply")
if name == "add_argument":
flags.update(arg.value for arg in node.args
if isinstance(arg, ast.Constant) and isinstance(arg.value, str))
return {flag for flag in flags if flag.startswith("--apply")}
class ScriptPathPolicyTests(unittest.TestCase):
"""门槛（b）：Peach 数据目录只在 `peach.config` / `peach.platform` 里出现。"""
def test_no_script_writes_a_data_root_path_by_hand(self):
offenders = []
for path in sorted(SCRIPTS.glob("*.py")):
if path.name in DATA_ROOT_ALLOWLIST:
continue
offenders += [f"{path.name}:{line}" for line, text in _live_strings(path)
if DATA_ROOT_LITERAL.search(text)]
self.assertEqual(offenders, [], "改成从 peach.config 取（DATABASE_PATH、"
"GENERATED_DIR、LOG_DIR、STATE_DIR 等）")
def test_only_config_and_platform_know_where_the_data_root_is(self):
offenders = []
for path in sorted(SRC.glob("*.py")):
if path.name in PATH_AUTHORITIES:
continue
offenders += [f"{path.name}:{line}" for line, text in _live_strings(path)
if DATA_ROOT_LITERAL.search(text)]
self.assertEqual(offenders, [])
def test_the_guard_catches_the_shape_it_describes(self):
"""门槛自身也要能被证伪，否则它可能只是一段永远为真的代码。"""
self.assertTrue(DATA_ROOT_LITERAL.search(r"R:\peach-data\database\ledger.db"))
self.assertTrue(DATA_ROOT_LITERAL.search("C:/Users/<user>/Desktop/peach"))
self.assertTrue(DATA_ROOT_LITERAL.search("/Users/<user>/Desktop/peach"))
self.assertIsNone(DATA_ROOT_LITERAL.search("peach-data/generated"))
class ScriptWriteGatePolicyTests(unittest.TestCase):
"""门槛（c）：会改账本的脚本必须自己声明 `--apply`。"""
def test_every_script_with_dml_declares_an_apply_switch(self):
offenders = []
for path in sorted(SCRIPTS.glob("*.py")):
if path.name in APPLY_ALLOWLIST:
continue
has_dml = any(DML.search(text) and DML_CLAUSE.search(text)
for _line, text in _live_strings(path))
if has_dml and not _apply_flags(path):
offenders.append(path.name)
self.assertEqual(offenders, [], "真实写入默认必须是 dry-run："
"用 peach.scripting.add_ledger_write_args")
def test_the_allowlist_has_no_stale_entries(self):
"""例外表本身会过期。脚本删了或者已经补上 `--apply`，条目就该跟着走。"""
for name in sorted(APPLY_ALLOWLIST | DATA_ROOT_ALLOWLIST):
with self.subTest(script=name):
self.assertTrue((SCRIPTS / name).is_file(), f"{name} 已不存在")
for name in sorted(APPLY_ALLOWLIST):
with self.subTest(script=name):
self.assertFalse(_apply_flags(SCRIPTS / name),
f"{name} 已经有 --apply，从例外表里去掉")
def test_the_guard_catches_the_shape_it_describes(self):
self.assertTrue(DML.search("INSERT OR IGNORE INTO asset_tag(asset_id,tag)"))
self.assertTrue(DML.search("UPDATE asset SET path=? WHERE id=?"))
self.assertIsNone(DML.search("SELECT count(*) FROM asset"))
class ReleaseEntryImportTests(unittest.TestCase):
"""门槛（d）：发布入口导入时只靠标准库。
Release 工作流核对标签提交那一步只有 `setup-python` 装的裸解释器，依赖要到后面的
构建步骤才装——次序是故意的：不给一个没过 CI 的提交花十分钟构建。入口只要多引一条
第三方依赖，那一步就 `ModuleNotFoundError`，而此时标签已经推上去了：版本有标签、却没有
任何可下载的制品，而发布流程的全部意义就是「每个 X.Y.Z 对应一份制品」。
"""
def test_the_release_entry_imports_with_a_bare_interpreter(self):
base = Path(sys.base_prefix)
interpreter = next((path for path in (base / "python.exe",
base / "bin" / "python3",
base / "bin" / "python") if path.is_file()), None)
if interpreter is None:
self.skipTest(f"找不到不带 site-packages 的基底解释器：{base}")
# `-S` 不载入 site：基底解释器自己的 site-packages 里也可能装着 filelock，
# 载入了就会让重依赖混过这道门槛。`-E` 不认 PYTHONPATH，理由相同。
done = subprocess.run(
[str(interpreter), "-S", "-E", "-X", "utf8", "-c",
"import scripts.release_tag as entry; print(entry.MASTER_WRITER)"],
cwd=str(ROOT), capture_output=True, text=True, encoding="utf-8",
errors="replace", check=False)
self.assertEqual(done.returncode, 0,
"发布入口得能裸解释器导入：" + (done.stderr or done.stdout).strip())
if __name__ == "__main__":
unittest.main()