muse-agent-example/tests/架构/test_模块依赖.py
zizi b86516505f W02 安装工程、启动骨架与验证环境:pyproject/uv 安装工程、启动装配骨架、conftest 用例身份与 OS 离线隔离。
按 R2 串行阶段整理提交;包内文件为该阶段交付(含后续小增量),状态以工作包清单为准。
2026-09-10 19:25:40 +08:00

331 lines
14 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""模块依赖行为验证。
保护:安装产物边界与依赖方向——业务只经公共接口、Skill 不携带实现、
接入不直写表、共享基础不反向依赖业务、装配层不承载创作规则。
已实现子集(W02):
- 安装包在任意 cwd 可导入且导入不产生外部连接(含旧树同名包遮挡守卫);
- 技能目录不注入 sys.path、不携带实现脚本;
- src/muse 活跃运行代码不依赖外部专有 CLI;
- 共享基础类型不导入业务模块;__main__ 只是薄命令入口。
依赖 W06/W07/W22/W24 的用例(宿主适配、协议、角色路由、旧库参数守卫、
技能目录合同)在对应包落地后逐例实现,不以形状检查代替。
"""
from __future__ import annotations
import ast
import json
import re
import subprocess
import sys
from pathlib import Path
import muse
仓库根 = Path(__file__).resolve().parents[2]
源码根 = 仓库根 / "src" / "muse"
def _遍历py(根: Path) -> list[Path]:
return sorted(p for p in 根.rglob("*.py") if "__pycache__" not in p.parts)
def _导入名(文件: Path) -> set[str]:
"""AST 提取文件内全部 import 的顶层模块名。"""
树 = ast.parse(文件.read_text(encoding="utf-8"))
名单: set[str] = set()
for 节点 in ast.walk(树):
if isinstance(节点, ast.Import):
for 别名 in 节点.names:
名单.add(别名.name.split(".")[0])
elif isinstance(节点, ast.ImportFrom) and 节点.module and 节点.level == 0:
名单.add(节点.module.split(".")[0])
return 名单
def test_skills_do_not_syspath_into_shared_implementations__f4e7a4() -> None:
"""技能目录是方法与操作说明,不携带实现脚本,不注入 sys.path。"""
技能根 = 仓库根 / ".agent" / "skills"
脚本 = [p for p in 技能根.rglob("*.py") if "__pycache__" not in p.parts]
assert 脚本 == [], [str(p.relative_to(仓库根)) for p in 脚本]
for md in 技能根.rglob("*.md"):
文本 = md.read_text(encoding="utf-8")
assert "sys.path" not in 文本, f"技能文档出现 sys.path 注入:{md.relative_to(仓库根)}"
def _字符串常量(文件: Path) -> list[str]:
"""AST 提取全部字符串常量,排除模块/类/函数 docstring。"""
树 = ast.parse(文件.read_text(encoding="utf-8"))
文档: set[int] = set()
for 节点 in ast.walk(树):
if isinstance(节点, (ast.Module, ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
主体 = 节点.body
if 主体 and isinstance(主体[0], ast.Expr) and isinstance(主体[0].value, ast.Constant):
文档.add(id(主体[0].value))
return [
节点.value
for 节点 in ast.walk(树)
if isinstance(节点, ast.Constant) and isinstance(节点.value, str) and id(节点) not in 文档
]
def test_dashboard_does_not_import_skills__ba6ad5() -> None:
"""src/mude 不以代码依赖 .agent 技能目录(文档中的合同引用不算依赖)。"""
for 文件 in _遍历py(源码根):
for 值 in _字符串常量(文件):
assert ".agent" not in 值, f"源码以路径方式引用 .agent:{文件.relative_to(仓库根)}"
assert "skills/" not in 值, f"源码引用技能路径:{文件.relative_to(仓库根)}"
def test_active_runtime_has_no_claude_cli_dependency__1ec1b5() -> None:
"""活跃运行代码不依赖专有外部 CLI(claude/dsh 等);子进程只允许工程工具。"""
for 文件 in _遍历py(源码根):
树 = ast.parse(文件.read_text(encoding="utf-8"))
for 节点 in ast.walk(树):
if isinstance(节点, ast.Call) and getattr(节点.func, "attr", "") == "run":
for 参数 in 节点.args:
if isinstance(参数, ast.List):
for 元素 in 参数.elts:
if isinstance(元素, ast.Constant) and isinstance(元素.value, str):
值 = 元素.value
assert not (值.startswith("claude") or 值.startswith("dsh")), (
f"依赖专有 CLI:{文件.relative_to(仓库根)} -> {值}"
)
def test_runtime_does_not_import_muse_business__5d7edd() -> None:
"""共享纯基础不反向导入业务;启动装配可以绑定具体实现。"""
业务模块 = {
"作品规划",
"故事世界",
"资料研究",
"知识方法",
"正文写作",
"审校修订",
"作者经验",
"交付连载",
"上下文",
"效果评测",
}
基础文件 = list((源码根 / "共享").glob("*.py")) + [
源码根 / "配置.py",
源码根 / "资源加载.py",
]
assert 基础文件, "共享与装配文件缺失"
for 文件 in 基础文件:
树 = ast.parse(文件.read_text(encoding="utf-8"))
for 节点 in ast.walk(树):
if isinstance(节点, ast.ImportFrom) and 节点.module:
顶层 = 节点.module.split(".")[0] == "muse"
if 顶层 and len(节点.module.split(".")) > 1:
第二层 = 节点.module.split(".")[1]
assert 第二层 not in 业务模块, (
f"基础层导入业务模块:{文件.relative_to(仓库根)} -> {节点.module}"
)
def test_dispatch_agent_task_is_thin_cli__b141f1() -> None:
"""python -m muse 是薄入口:只做参数解析与转交,无业务逻辑与连接。"""
文件 = 源码根 / "__main__.py"
文本 = 文件.read_text(encoding="utf-8")
导入 = _导入名(文件)
assert 导入 <= {"__future__", "argparse", "sys", "pathlib", "muse"}, 导入
for 禁止 in ("socket", "psycopg", "http", "urllib", "subprocess"):
assert 禁止 not in 文本, f"命令入口包含 {禁止}"
# ---------------------------------------------------------------------------
# 新增用例(W02 补充):安装与导入边界
# ---------------------------------------------------------------------------
def test_import_from_any_cwd_without_external_connections__e0a1b2() -> None:
"""任意 cwd 导入 muse 全部顶层模块且不产生任何 socket 连接(安装验收)。"""
代码 = (
"import socket\n"
"def _拒绝(self, 地址):\n"
" raise AssertionError(f'导入期间尝试连接 {地址}')\n"
"socket.socket.connect = _拒绝\n"
"import muse, muse.启动, muse.配置, muse.资源加载, muse.共享\n"
"print(muse.__version__)\n"
)
进程 = subprocess.run(
[sys.executable, "-c", 代码], cwd="/", capture_output=True, text=True, timeout=60
)
assert 进程.returncode == 0, 进程.stderr
assert 进程.stdout.strip() == muse.__version__
def test_import_resolves_to_new_package_not_old_tree__c1d2e3() -> None:
"""import muse 解析到 src/muse 安装包,不被旧树同名目录遮挡。"""
代码 = "import muse\nprint(muse.__file__)\n"
进程 = subprocess.run(
[sys.executable, "-c", 代码],
cwd=仓库根,
capture_output=True,
text=True,
timeout=60,
)
assert 进程.returncode == 0, 进程.stderr
assert "src/muse" in 进程.stdout, 进程.stdout
# ---------------------------------------------------------------------------
# W13 承接:作者自然语言入口与角色路由合同(原 AuthorInstructionContractTest)
# ---------------------------------------------------------------------------
作者根 = 仓库根 / ".agent" / "作者"
场景根 = 作者根 / "场景"
标准索引表头 = (
"| 名称 | 相对地址 | 内容描述 | 使用场景 | 使用要求 |\n"
"|------|----------|----------|----------|----------|"
)
场景清单 = ("定故事", "排故事", "塑人物", "写下一章", "改正文", "查作品")
技能链接 = re.compile(r"\[([^\]\n]+)\]\(([^)\n]+/SKILL\.md)\)")
frontmatter名称 = re.compile(r"^name:\s*(.+?)\s*$", re.MULTILINE)
# 已有新版实现并登记进运行资源包的技能:场景路由必须指向新版路径,不再回到旧 muse 树。
已替代技能 = {
"完善故事基础设定": ".agent/skills/操作/完善故事基础设定/SKILL.md",
"决定正文候选去留": ".agent/skills/操作/决定正文候选去留/SKILL.md",
"确认规划候选": ".agent/skills/操作/确认规划候选/SKILL.md",
}
退役技能别名 = {
"生成下一章",
"局部重写",
"场景扩写",
"正文润色",
"作品定盘",
"全书规划",
"章级细纲",
"全书解析",
"章后抽取",
"一致性检测",
"质量评分",
}
def _章节(文本: str, 标题: str) -> str:
匹配 = re.search(
rf"^## {re.escape(标题)}\s*$\n(?P<正文>.*?)(?=^## |\Z)",
文本,
flags=re.MULTILINE | re.DOTALL,
)
assert 匹配 is not None, f"缺少章节:## {标题}"
return 匹配.group("正文")
def test_author_layer_has_discoverable_structure__41d4cb() -> None:
"""作者层文件完整且各级索引可发现;路由不依赖技能自动发现。"""
期望 = {
作者根 / "目录.md",
作者根 / "指令.md",
作者根 / "原生中文写作.md",
场景根 / "目录.md",
*(场景根 / f"{名称}.md" for 名称 in 场景清单),
}
缺失 = sorted(str(p.relative_to(仓库根)) for p in 期望 if not p.is_file())
assert 缺失 == [], "作者指令层文件不完整:\n" + "\n".join(缺失)
根索引 = (仓库根 / ".agent" / "目录.md").read_text(encoding="utf-8")
项目入口 = (仓库根 / "AGENTS.md").read_text(encoding="utf-8")
作者索引 = (作者根 / "目录.md").read_text(encoding="utf-8")
场景索引 = (场景根 / "目录.md").read_text(encoding="utf-8")
assert "作者/目录.md" in 根索引
assert ".agent/作者/指令.md" in 项目入口
assert 作者索引.startswith(标准索引表头)
assert 场景索引.startswith(标准索引表头)
assert "[指令.md](指令.md)" in 作者索引
assert "[原生中文写作.md](原生中文写作.md)" in 作者索引
assert "[目录.md](场景/目录.md)" in 作者索引
for 场景 in 场景清单:
assert f"[{场景}.md]({场景}.md)" in 场景索引
def test_scenes_route_only_to_registered_chinese_skills__5e199d() -> None:
"""场景路由只指向真实存在且 frontmatter 名称一致的技能;已替代技能不再路由旧实现。"""
登记能力 = {
Path(条目).name
for 条目 in json.loads((仓库根 / "工具" / "资源登记.json").read_text(encoding="utf-8"))[
"能力文件"
]
}
for 场景 in 场景清单:
源 = 场景根 / f"{场景}.md"
路由章节 = _章节(源.read_text(encoding="utf-8"), "正式技能")
for 标签, 原目标 in 技能链接.findall(路由章节):
目标 = (源.parent / 原目标).resolve()
assert 目标.is_file(), f"{源}: 技能链接不存在:{原目标}"
匹配 = frontmatter名称.search(目标.read_text(encoding="utf-8"))
assert 匹配 is not None, f"{目标}: 缺少 frontmatter name"
assert 标签 == 匹配.group(1).strip(), f"{源}: 链接文字必须使用正式技能名"
相对 = 目标.relative_to(仓库根).as_posix()
if 标签 in 已替代技能:
assert 相对 == 已替代技能[标签], (
f"{源}: {标签} 已有新版实现,应路由 {已替代技能[标签]} 而不是 {相对}"
)
if "muse/" not in 相对:
assert Path(相对).name in 登记能力, f"{源}: 未登记的新版技能:{相对}"
assert 标签 not in 退役技能别名
# 作者层不是技能:不得出现作者入口冒充技能登记。
assert 登记能力.isdisjoint({"目录.md"})
def test_interaction_and_native_chinese_contracts_are_complete__7239c7() -> None:
"""交互边界与原生中文写作合同的关键短语在作者层保持完整。"""
入口 = (作者根 / "指令.md").read_text(encoding="utf-8")
for 必含 in (
"一轮只问一个关键问题",
"二至三个中文选项",
"采用",
"修改",
"丢弃",
"Shadow",
"Canonical",
"只给正文",
"主会话不代替角色",
):
assert 必含 in 入口, f"作者指令缺少交互合同短语:{必含}"
写作 = (作者根 / "原生中文写作.md").read_text(encoding="utf-8")
for 必含 in (
"中文语序",
"具体名词、动作和关系优先",
"作者声音、叙述者声音和角色声音",
"不得凭空增加",
"润色不得偷改情节事实",
"扩写不得新开场景或改变落点",
"重写不得越过作者指定范围",
"精确诊断",
):
assert 必含 in 写作, f"原生中文写作合同缺少:{必含}"
诊断场景 = (场景根 / "查作品.md").read_text(encoding="utf-8")
修改场景 = (场景根 / "改正文.md").read_text(encoding="utf-8")
assert "只查不改" in 诊断场景
assert "作者指定范围" in 修改场景
def test_role_prompts_route_with_registered_skill_names__4bead8() -> None:
"""五个角色文件与角色合同一致并登记进运行资源包;角色层不使用退役技能别名。"""
登记 = json.loads((仓库根 / "工具" / "资源登记.json").read_text(encoding="utf-8"))
能力 = set(登记["能力文件"])
角色合同 = {
"writer": "写手",
"planner": "规划员",
"extractor": "知识抽取员",
"detector": "检测员",
"judge": "评委",
}
for 角色ID, 名称 in 角色合同.items():
相对 = f".agent/角色/{名称}.md"
assert 相对 in 能力, f"角色未登记进运行资源包:{相对}"
文本 = (仓库根 / 相对).read_text(encoding="utf-8")
头部 = 文本.split("---")[1]
assert f"id: {角色ID}" in 头部
assert f"name: {名称}" in 头部
assert "contract_version:" in 头部
for 别名 in 退役技能别名:
assert f"`{别名}`" not in 文本, f"{相对}: 角色文件引用退役技能别名 {别名}"