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

583 lines
21 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.

"""测试依赖与执行证据的逐例行为验证。
保护:用例选择、结构化报告、离线依赖阻断、超时与失败区分、
空收集与缺环境不静默、登记范围与执行证据的拒绝语义。
环境:离线;受限子进程和临时测试项目(复制仓库真实 conftest 后在
临时项目中运行 pytest,验证同一套隔离与守卫语义,不是旧文件迁移)。
"""
from __future__ import annotations
import ast
import json
import shutil
import subprocess
import sys
import textwrap
import xml.etree.ElementTree as ET
from pathlib import Path
import pytest
仓库根 = Path(__file__).resolve().parents[2]
真实conftest = 仓库根 / "tests" / "conftest.py"
class 临时项目:
"""受限临时 pytest 项目:复制真实 conftest,写入指定用例文件。"""
def __init__(self, 根: Path) -> None:
self.根 = 根
shutil.copyfile(真实conftest, 根 / "conftest.py")
def 写(self, 相对: str, 内容: str) -> Path:
目标 = self.根 / 相对
目标.parent.mkdir(parents=True, exist_ok=True)
目标.write_text(textwrap.dedent(内容), encoding="utf-8")
return 目标
def 运行(self, *参数: str, 环境: dict[str, str] | None = None) -> tuple[int, dict]:
报告 = self.根 / "报告.xml"
完整 = [sys.executable, "-m", "pytest", "-q", f"--junitxml={报告}", *参数]
清单 = self.根 / "用例清单.json"
if not 清单.exists():
目标 = []
for 文件 in self.根.rglob("test_*.py"):
for 节点 in ast.walk(ast.parse(文件.read_text())):
if isinstance(节点, ast.FunctionDef) and 节点.name.startswith("test_"):
后缀 = 节点.name.rsplit("__", 1)[-1]
目标.append(
{
"case_id": f"TC-x-{后缀}000000",
"file": 文件.relative_to(self.根).as_posix(),
"symbol": 节点.name,
}
)
清单.write_text(json.dumps({"cases": [], "new_cases": 目标}, ensure_ascii=False))
完整.extend(["--用例清单", str(清单)])
进程 = subprocess.run(
完整,
cwd=self.根,
capture_output=True,
text=True,
timeout=180,
env={**(_清理环境()), **(环境 or {})},
)
结果: dict = {"退出码": 进程.returncode, "stdout": 进程.stdout, "stderr": 进程.stderr}
if 报告.exists():
结果["报告"] = 解析junit(报告)
return 进程.returncode, 结果
def _清理环境() -> dict[str, str]:
"""子进程环境:剔除数据库变量,避免宿主环境泄漏进隔离语义。"""
import os
环境 = dict(os.environ)
环境.pop("MUSE_TEST_DATABASE_URL", None)
return 环境
def 解析junit(路径: Path) -> dict:
树 = ET.parse(路径)
根 = 树.getroot()
用例 = []
for 节点 in 根.iter("testcase"):
条目: dict = {"名称": 节点.get("name", ""), "类": 节点.get("classname", "")}
for 子 in 节点:
if 子.tag not in {"failure", "error", "skipped"}:
continue
条目["结果"] = 子.tag
条目["消息"] = (子.get("message") or "")[:300]
if 子.text:
条目["正文"] = 子.text[:300]
用例.append(条目)
汇总 = 根.find(".//testsuite")
return {
"用例": 用例,
"数量": int(汇总.get("tests", "0")) if 汇总 is not None else len(用例),
"失败": int(汇总.get("failures", "0")) if 汇总 is not None else 0,
"跳过": int(汇总.get("skipped", "0")) if 汇总 is not None else 0,
}
@pytest.fixture
def 项目(tmp_path: Path) -> 临时项目:
return 临时项目(tmp_path)
# ---------------------------------------------------------------------------
# 用例选择与结构化报告
# ---------------------------------------------------------------------------
def test_selector_alias_and_intersection_are_reported_as_json__a5c7d4(项目: 临时项目) -> None:
"""--case 与标记取交集只选中目标用例;结构化报告精确到单个用例。"""
项目.写(
"test_样本.py",
"""
def test_目标行为__aa0001():
assert 1 + 1 == 2
def test_另一行为__bb0002():
assert 2 + 2 == 4
""",
)
退出码, 结果 = 项目.运行("--case", "TC-x-aa0001000000")
assert 退出码 == 0, 结果["stderr"]
报告 = 结果["报告"]
assert 报告["数量"] == 1, 报告
assert 报告["用例"][0]["名称"] == "test_目标行为__aa0001"
assert "结果" not in 报告["用例"][0] # 通过:无 failure/skipped 节点
def test_offline_dependency_is_blocked_without_running_child__09d22d(项目: 临时项目) -> None:
"""数据库标记用例在缺环境时被阻断:子进程不执行、全跳过不算通过。"""
项目.写(
"test_数据库用例.py",
"""
from pathlib import Path
import pytest
@pytest.mark.数据库
def test_需要数据库__cc0003(隔离数据库URL):
Path("child_ran").write_text("ran", encoding="utf-8")
assert 隔离数据库URL
""",
)
退出码, 结果 = 项目.运行("--case", "TC-x-cc0003000000")
# 全跳过 → 守卫改写为退出码 5,不计通过;跳过原因可见
assert 退出码 == 5, (退出码, 结果["stderr"])
报告 = 结果["报告"]
assert 报告["用例"][0]["结果"] == "skipped"
assert "MUSE_TEST_DATABASE_URL" in 报告["用例"][0]["消息"]
# 子进程没有执行:没有任何副作用文件
assert not (项目.根 / "child_ran").exists()
def test_allow_requires_runs_and_still_reports_dependency__08eb1b(项目: 临时项目) -> None:
"""提供隔离库引用后数据库用例真实执行;依赖来源在结果中可见。"""
项目.写(
"test_数据库用例.py",
"""
import pytest
@pytest.mark.数据库
def test_获得隔离引用__cc0003(隔离数据库URL):
assert 隔离数据库URL == "postgresql://synthetic-test"
""",
)
退出码, 结果 = 项目.运行(
"--case",
"TC-x-cc0003000000",
环境={"MUSE_TEST_DATABASE_URL": "postgresql://synthetic-test"},
)
assert 退出码 == 0, (退出码, 结果["stderr"])
assert 结果["报告"]["数量"] == 1
def test_nonzero_and_timeout_have_distinct_structured_statuses__306973(项目: 临时项目) -> None:
"""断言失败与超时都以非零退出,且失败消息可区分。"""
项目.写(
"test_失败.py",
"""
def test_断言失败__dd0004():
assert 1 == 2, "合成失败"
""",
)
项目.写(
"test_睡眠.py",
"""
import time
def test_超时睡眠__ee0005():
time.sleep(30)
""",
)
码1, 结果1 = 项目.运行("--case", "TC-x-dd0004000000")
assert 码1 == 1
断言消息 = 结果1["报告"]["用例"][0]["消息"]
assert "合成失败" in 断言消息
码2, 结果2 = 项目.运行("--case", "TC-x-ee0005000000", "--timeout=2")
assert 码2 == 1
超时正文 = 结果2["报告"]["用例"][0].get("正文", "") + 结果2["报告"]["用例"][0].get("消息", "")
assert "Timeout" in 超时正文 or "timeout" in 超时正文
# 两者消息可区分:超时消息不含合成失败标记
assert "合成失败" not in 超时正文
def test_no_matches_and_missing_path_are_not_silent__ae03b5(项目: 临时项目) -> None:
"""未知 case 选择与非空选择的空收集都以非零退出并说明原因。"""
项目.写(
"test_存在.py",
"""
def test_存在用例__ff0006():
assert True
""",
)
码1, 结果1 = 项目.运行("--case", "TC-x-0000zz9999")
assert 码1 != 0
assert "--case" in 结果1["stderr"] or "--case" in 结果1["stdout"]
码2, 结果2 = 项目.运行("tests/不存在的路径")
assert 码2 in (4, 5) # 缺失路径或空收集:非零且不静默
assert "no tests" in (结果2["stdout"] + 结果2["stderr"]).lower()
def test_invalid_manifest_and_selector_requirement_are_structured__f7150b(项目: 临时项目) -> None:
"""非法清单与未登记选择分别失败,错误原因可区分。"""
项目.写("test_任意.py", "def test_任意用例__000007():\n assert True\n")
清单 = 项目.根 / "用例清单.json"
清单.write_text("{broken", encoding="utf-8")
码, 结果 = 项目.运行("--case", "TC-x-000007000000")
assert 码 == 4
assert "用例清单" in 结果["stderr"]
清单.write_text('{"cases": [], "new_cases": []}', encoding="utf-8")
码, 结果 = 项目.运行("--case", "TC-x-000007000000")
assert 码 == 4
assert "未登记" in 结果["stderr"]
def test_all_offline_is_explicit_opt_in_and_output_is_summary_only__265060(项目: 临时项目) -> None:
"""默认运行按标记剔除外部环境用例;报告只含离线用例。"""
项目.写(
"test_离线.py",
"""
def test_离线用例__110008():
assert True
""",
)
项目.写(
"test_外部.py",
"""
import pytest
@pytest.mark.数据库
def test_外部用例__220009(隔离数据库URL):
assert 隔离数据库URL
""",
)
码, 结果 = 项目.运行("-m", "not 数据库")
assert 码 == 0, (码, 结果["stderr"])
名称 = [u["名称"] for u in 结果["报告"]["用例"]]
assert 名称 == ["test_离线用例__110008"], 名称
def test_zero_exit_without_execution_evidence_fails_closed__50f80b(项目: 临时项目) -> None:
"""空用例文件与纯跳过都不允许零退出冒充通过。"""
项目.写(
"test_空.py",
"""
说明 = "本文件没有任何测试函数"
""",
)
项目.写(
"test_跳过.py",
"""
import pytest
def test_跳过用例__33000a():
pytest.skip("合成跳过")
""",
)
# 只选空文件:无收集 → 退出码 5
码1, 结果1 = 项目.运行("test_空.py")
assert 码1 == 5
# 纯跳过:守卫改写退出码为 5,不计通过
码2, 结果2 = 项目.运行("test_跳过.py")
assert 码2 == 5, (码2, 结果2["stdout"])
assert "零实际执行" in 结果2["stderr"]
# ---------------------------------------------------------------------------
# 登记范围(生成范围)检查:工具/维护索引.py 的同一语义
# ---------------------------------------------------------------------------
sys.path.insert(0, str(仓库根 / "工具"))
from 维护索引 import 检查用例身份 # noqa: E402
def 写清单(根: Path, 目标: list[dict]) -> Path:
清单 = 根 / "清单.json"
清单.write_text(
json.dumps({"cases": [{"target_cases": 目标}], "new_cases": []}, ensure_ascii=False),
encoding="utf-8",
)
return 清单
def test_独立用例身份可绑定原生参数行__e2000b(tmp_path: Path) -> None:
项目 = 临时项目(tmp_path)
项目.写(
"test_参数.py",
"""
import pytest
@pytest.mark.parametrize("输入", [1, 2], ids=["first", "second"])
def test_参数行为__abcdef(输入):
assert 输入 in (1, 2)
""",
)
项目.写(
"用例清单.json",
json.dumps(
{
"cases": [],
"new_cases": [
{
"case_id": "NC-first",
"file": "test_参数.py",
"symbol": "test_参数行为__abcdef",
"parameter_ids": ["first"],
},
{
"case_id": "NC-second",
"file": "test_参数.py",
"symbol": "test_参数行为__abcdef",
"parameter_ids": ["second"],
},
],
}
),
)
退出, 结果 = 项目.运行("--case", "NC-first")
assert 退出 == 0, 结果
树 = ET.parse(tmp_path / "报告.xml")
行 = 树.findall(".//testcase")
assert len(行) == 1 and 行[0].attrib["name"].endswith("[first]")
assert 行[0].find(".//property[@name='case_id']").attrib["value"] == "NC-first"
def test_generated_scope_rejects_unregistered_disk_asset__69440b(tmp_path: Path) -> None:
"""磁盘上存在但清单未登记的用例被拒绝(未登记资产)。"""
测试文件 = tmp_path / "tests" / "registered" / "test_登记.py"
测试文件.parent.mkdir(parents=True)
测试文件.write_text("def test_已登记__44000b():\n assert True\n", encoding="utf-8")
未登记 = tmp_path / "tests" / "new" / "test_未登记.py"
未登记.parent.mkdir(parents=True)
未登记.write_text("def test_未登记__55000c():\n assert True\n", encoding="utf-8")
清单 = 写清单(
tmp_path,
[
{
"case_id": "TC-x-44000b000000",
"file": "tests/registered/test_登记.py",
"symbol": "test_已登记__44000b",
}
],
)
问题 = 检查用例身份(tmp_path, 清单, 核对缺失目标=True, 全树扫描=True)
assert any("test_未登记__55000c" in p and "未在用例清单登记" in p for p in 问题), 问题
def test_generated_scope_uses_manifest_scan_roots__20f4fa(tmp_path: Path) -> None:
"""自定义扫描根下的已登记用例被接受(无问题)。"""
测试文件 = tmp_path / "custom-tests" / "test_自定义.py"
测试文件.parent.mkdir(parents=True)
测试文件.write_text("def test_自定义用例__66000d():\n assert True\n", encoding="utf-8")
清单 = 写清单(
tmp_path,
[
{
"case_id": "TC-x-66000d000000",
"file": "custom-tests/test_自定义.py",
"symbol": "test_自定义用例__66000d",
}
],
)
assert 检查用例身份(tmp_path, 清单, 核对缺失目标=True, 全树扫描=True) == []
def test_generated_scope_rejects_manifest_extra_asset__b4901f(tmp_path: Path) -> None:
"""清单登记的目标在磁盘上不存在时被拒绝。"""
清单 = 写清单(
tmp_path,
[
{
"case_id": "TC-x-77000e000000",
"file": "tests/removed/test_已删除.py",
"symbol": "test_已删除__77000e",
}
],
)
问题 = 检查用例身份(tmp_path, 清单, 核对缺失目标=True, 全树扫描=True)
assert any("目标不存在" in p for p in 问题), 问题
def test_generated_scope_accepts_existing_behavior_eval_entry__5b8bb2(tmp_path: Path) -> None:
"""已存在的真实模型评测用例登记被接受(含环境标记)。"""
测试文件 = tmp_path / "tests" / "evals" / "test_评测.py"
测试文件.parent.mkdir(parents=True)
测试文件.write_text(
"import pytest\n\n@pytest.mark.真实模型\ndef test_评测用例__88000f():\n assert True\n",
encoding="utf-8",
)
清单 = 写清单(
tmp_path,
[
{
"case_id": "TC-x-88000f000000",
"file": "tests/evals/test_评测.py",
"symbol": "test_评测用例__88000f",
}
],
)
assert 检查用例身份(tmp_path, 清单, 核对缺失目标=True, 全树扫描=True) == []
def test_generated_scope_rejects_missing_behavior_eval_entry__fe3fba(tmp_path: Path) -> None:
"""登记的真实模型评测用例文件缺失时被拒绝。"""
清单 = 写清单(
tmp_path,
[
{
"case_id": "TC-x-990010000000",
"file": "tests/evals/test_缺失评测.py",
"symbol": "test_缺失评测__990010",
}
],
)
问题 = 检查用例身份(tmp_path, 清单, 核对缺失目标=True, 全树扫描=True)
assert any("目标不存在" in p for p in 问题), 问题
def test_generated_scope_ignores_non_test_helpers_under_test_roots__14f941(tmp_path: Path) -> None:
"""测试根下的非 test_ 辅助函数不触发登记检查。"""
测试文件 = tmp_path / "tests" / "helpers" / "test_辅助.py"
测试文件.parent.mkdir(parents=True)
测试文件.write_text(
"def 辅助函数():\n"
" return 1\n"
"\n"
"\n"
"def test_真用例__a10011():\n"
" assert 辅助函数() == 1\n",
encoding="utf-8",
)
清单 = 写清单(
tmp_path,
[
{
"case_id": "TC-x-a10011000000",
"file": "tests/helpers/test_辅助.py",
"symbol": "test_真用例__a10011",
}
],
)
assert 检查用例身份(tmp_path, 清单, 核对缺失目标=True, 全树扫描=True) == []
def test_generated_scope_blocks_empty_and_print_only_scripts_before_child__36f79e(
tmp_path: Path,
) -> None:
"""无可判定语句的 print 型用例在执行前即被登记检查拒绝。"""
空文件 = tmp_path / "tests" / "empty" / "test_空断言.py"
空文件.parent.mkdir(parents=True)
空文件.write_text("def test_空用例__b20012():\n print('只有输出')\n", encoding="utf-8")
清单 = 写清单(
tmp_path,
[
{
"case_id": "TC-x-b20012000000",
"file": "tests/empty/test_空断言.py",
"symbol": "test_空用例__b20012",
}
],
)
问题 = 检查用例身份(tmp_path, 清单, 核对缺失目标=True, 全树扫描=True)
assert any("无可判定语句" in p for p in 问题), 问题
def test_generated_scope_accepts_supported_python_shapes__7c1382(tmp_path: Path) -> None:
"""函数式与类式用例形状都被识别并核对。"""
测试文件 = tmp_path / "tests" / "shapes" / "test_形状.py"
测试文件.parent.mkdir(parents=True)
测试文件.write_text(
"class Test类式:\n"
" def test_类方法__c30013(self):\n"
" assert True\n"
"\n"
"\n"
"def test_函数式__d30014():\n"
" assert True\n",
encoding="utf-8",
)
清单 = 写清单(
tmp_path,
[
{
"case_id": "TC-x-c30013000000",
"file": "tests/shapes/test_形状.py",
"symbol": "test_类方法__c30013",
},
{
"case_id": "TC-x-d30014000000",
"file": "tests/shapes/test_形状.py",
"symbol": "test_函数式__d30014",
},
],
)
assert 检查用例身份(tmp_path, 清单, 核对缺失目标=True, 全树扫描=True) == []
def test_完整身份不接受冒用后缀__a02001(项目: 临时项目) -> None:
"""未登记 ID 与真实符号后缀相同,也不能选择执行。"""
项目.写("test_样本.py", "def test_目标__aa0001():\n assert True\n")
码, 结果 = 项目.运行("--case", "XX-forged-aa0001000000")
assert 码 == 4
assert "未登记" in 结果["stderr"]
def test_完整身份绑定文件与符号__a02002(tmp_path: Path) -> None:
"""另一文件冒用已登记后缀不能冒充目标用例。"""
(tmp_path / "test_冒用.py").write_text("def test_冒用__aa0001():\n assert True\n")
清单 = 写清单(
tmp_path,
[{"case_id": "TC-x-aa0001000000", "file": "test_原例.py", "symbol": "test_原例__aa0001"}],
)
问题 = 检查用例身份(tmp_path, 清单, 全树扫描=True)
assert any("未在用例清单登记" in 条 for 条 in 问题), 问题
@pytest.mark.数据库
@pytest.mark.parametrize("阶段", ["收集", "测试体", "子进程"], ids=["collection", "call", "child"])
def test_离线进程拒绝真实数据库连接__a02005(项目: 临时项目, 隔离数据库URL: str, 阶段: str) -> None:
"""先确认隔离库可达,再由未标记的子测试验证 OS 阻断 libpq。"""
import psycopg
with psycopg.connect(隔离数据库URL) as 连:
assert 连.execute("SELECT 1").fetchone() == (1,)
连接代码 = "import os, psycopg; psycopg.connect(os.environ['MUSE_TEST_DATABASE_URL'])"
if 阶段 == "收集":
内容 = 连接代码 + "\ndef test_试连接__aa0001():\n assert True\n"
elif 阶段 == "子进程":
内容 = (
"import subprocess, sys\n"
"def test_试连接__aa0001():\n"
f" subprocess.run([sys.executable, '-c', {连接代码!r}], check=True)\n"
)
else:
内容 = "def test_试连接__aa0001():\n " + 连接代码 + "\n"
项目.写("test_连接.py", 内容)
码, 结果 = 项目.运行(环境={"MUSE_TEST_DATABASE_URL": 隔离数据库URL})
assert 码 in (1, 2)
assert "Operation not permitted" in 结果["stdout"] + 结果["stderr"]
def test_已执行目标丢失不被过渡期忽略__a02006(tmp_path: Path) -> None:
清单 = 写清单(
tmp_path,
[
{
"case_id": "NC-existing-test",
"file": "tests/架构/test_已执行.py",
"symbol": "test_已执行__aa0001",
"verification_state": "通过",
}
],
)
问题 = 检查用例身份(tmp_path, 清单)
assert any("目标不存在" in x for x in 问题), 问题