muse-agent-example/tests/架构/test_测试依赖与执行证据.py
zizi d909d1bd1b 后端实现与用例身份:19 包集成落地并修复收尾缺陷
实现侧:
- 上下文:任务范围拆分为 范围校验/范围授权;索引按可发现口径重建、索引新鲜度改对称差;依赖校验统一快照漂移说明。
- 知识方法:方法与材料读取口径统一;超限方法材料按可选省略,核对路径不再二次计费;删除无合同的读时重算。
- 任务运行:新增 context.usage/tool.denied 事件类型;连接池常驻并在装配生命周期内开关;调用结算与核对分列。
- 效果评测/审校修订/交付连载/作者经验/作品规划:凭据冻结、标定消费、导出补证、事实引文核对等收尾修复。
- 资源加载:能力正文不再夹带索引用的导航注记(该注记此前进入角色与技能的模型提示)。
- 元数据:受保护骨架与代码保护属性对齐;字段校验与内置结构口径同步。
- 基础设施:环境预检进入装配生命周期;数据库连接运行期字段不参与相等比较;索引指纹归一化 jsonb 浮点。
- 删除被替代实现:7 份旧提示词模板与空壳 资料来源 读取器。

用例侧:
- 用例身份与导航元信息迁移;夹具补生命周期、同库暴露与模板封存;
- 本轮定向修复:方法材料省略、事实引文、迁移回执、额度与暂停用例、慢用例超时预算等。
2026-09-18 01:15:00 +08:00

661 lines
26 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 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")
shutil.copytree(
仓库根 / "tests/支持", 根 / "支持", ignore=shutil.ignore_patterns("__pycache__")
)
(根 / "工具").mkdir(exist_ok=True)
shutil.copyfile(仓库根 / "工具/用例身份.py", 根 / "工具/用例身份.py")
def 写(self, 相对: str, 内容: str, *, 登记身份: bool = True) -> Path:
"""合成用例使用显式源码 marker;不生产或读取外部位置清单。"""
目标 = self.根 / 相对
目标.parent.mkdir(parents=True, exist_ok=True)
文本 = textwrap.dedent(内容)
if 目标.name.startswith("test_") and 目标.suffix == ".py" and 登记身份:
树 = ast.parse(文本)
行 = 文本.splitlines(keepends=True)
插入 = []
for 节点 in ast.walk(树):
if not isinstance(节点, ast.FunctionDef) or not 节点.name.startswith("test_"):
continue
if any("pytest.mark.case_id" in ast.unparse(标) for 标 in 节点.decorator_list):
continue
后缀 = 节点.name.rsplit("__", 1)[-1]
起始 = min([节点.lineno, *[标.lineno for 标 in 节点.decorator_list]]) - 1
插入.append(
(起始, " " * 节点.col_offset + f"@pytest.mark.case_id('TC-x-{后缀}000000')\n")
)
for 位置, 代码 in sorted(插入, reverse=True):
行.insert(位置, 代码)
文本 = "".join(行)
if 插入 and not any(
isinstance(节, ast.Import) and any(别名.name == "pytest" for 别名 in 节.names)
for 节 in 树.body
):
文本 = "import pytest\n" + 文本
目标.write_text(文本, encoding="utf-8")
return 目标
def 运行(self, *参数: str, 环境: dict[str, str] | None = None) -> tuple[int, dict]:
报告 = self.根 / "报告.xml"
完整 = [sys.executable, "-m", "pytest", "-q", f"--junitxml={报告}", *参数]
进程 = 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 收集(self, *路径: str) -> list[dict]:
return 收集用例(self.根, list(路径) or None)
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)
# ---------------------------------------------------------------------------
# 用例选择与结构化报告
# ---------------------------------------------------------------------------
@pytest.mark.case_id(
"TC-a5c7d4adafe3",
environment="离线,受限子进程和临时测试项目",
when="收集并执行选定用例,模拟超时、缺依赖或零执行。",
contract=".agent/rules/测试隔离.md",
)
def test_selector_alias_and_intersection_are_reported_as_json__a5c7d4(项目: 临时项目) -> None:
"""--case 与标记取交集只选中目标用例;结构化报告精确到单个用例。"""
项目.写(
"test_样本.py",
"""
import pytest
def test_目标行为__aa0001():
assert 1 + 1 == 2
@pytest.mark.skip(reason="标记交集应在执行前排除")
def test_另一行为__bb0002():
assert 2 + 2 == 4
""",
)
退出码, 结果 = 项目.运行(
"--case", "TC-x-aa0001000000", "--case", "TC-x-bb0002000000", "-m", "not skip"
)
assert 退出码 == 0, 结果["stderr"]
报告 = 结果["报告"]
assert 报告["数量"] == 1, 报告
assert 报告["用例"][0]["名称"] == "test_目标行为__aa0001"
assert "结果" not in 报告["用例"][0] # 通过:无 failure/skipped 节点
@pytest.mark.case_id(
"TC-09d22d646494",
environment="离线,受限子进程和临时测试项目",
when="收集并执行选定用例,模拟超时、缺依赖或零执行。",
contract=".agent/rules/测试隔离.md",
)
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")
# 缺配置在setup阶段失败,不以skip计绿;业务体不执行。
assert 退出码 == 1, (退出码, 结果["stderr"])
报告 = 结果["报告"]
assert 报告["用例"][0]["结果"] == "error"
assert "MUSE_TEST_DATABASE_URL" in 报告["用例"][0]["消息"]
# 子进程没有执行:没有任何副作用文件
assert not (项目.根 / "child_ran").exists()
@pytest.mark.case_id(
"TC-08eb1b30bf46",
environment="离线,受限子进程和临时测试项目",
when="收集并执行选定用例,模拟超时、缺依赖或零执行。",
contract=".agent/rules/测试隔离.md",
)
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
@pytest.mark.case_id(
"TC-306973cf24b1",
environment="离线,受限子进程和临时测试项目",
when="收集并执行选定用例,模拟超时、缺依赖或零执行。",
contract=".agent/rules/测试隔离.md",
)
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 超时正文
@pytest.mark.case_id(
"TC-ae03b535241c",
environment="离线,受限子进程和临时测试项目",
when="收集并执行选定用例,模拟超时、缺依赖或零执行。",
contract=".agent/rules/测试隔离.md",
)
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()
@pytest.mark.case_id(
"TC-f7150be204a9",
environment="离线,受限子进程和临时测试项目",
given="损坏的 JSON 用例清单,以及合法空清单中的未登记完整 ID",
when="通过原生 pytest --case 选择",
then=["非法清单与未登记选择分别退出 4,错误原因明确,不执行测试"],
contract=".agent/rules/测试隔离.md",
)
def test_invalid_manifest_and_selector_requirement_are_structured__f7150b(项目: 临时项目) -> None:
"""旧导出损坏不干扰源身份;无效源 marker 给出准确错误。"""
项目.写("test_任意.py", "def test_任意用例__000007():\n assert True\n")
(项目.根 / "用例清单.json").write_text("{broken", encoding="utf-8")
码, 结果 = 项目.运行("--case", "TC-x-000007000000")
assert 码 == 0, 结果
项目.写(
"test_任意.py",
"import pytest\n@pytest.mark.case_id('wrong')\ndef test_任意(): assert True\n",
)
码, 结果 = 项目.运行()
assert 码 == 4
assert "case_id 无效" in 结果["stderr"]
@pytest.mark.case_id(
"TC-265060590b97",
environment="离线,受限子进程和临时测试项目",
when="收集并执行选定用例,模拟超时、缺依赖或零执行。",
contract=".agent/rules/测试隔离.md",
)
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"], 名称
@pytest.mark.case_id(
"TC-50f80b27c79f",
environment="离线,受限子进程和临时测试项目",
when="收集并执行选定用例,模拟超时、缺依赖或零执行。",
contract=".agent/rules/测试隔离.md",
)
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
from 维护索引 import 检查用例身份 # noqa: E402
@pytest.mark.case_id(
"NC-case-parameter-binding",
environment="离线子进程",
given="同一个原生参数化函数的两行分别登记不同 case_id",
when="仅选择第一条完整 ID",
then=["只运行对应参数行,JUnit 记录该 case_id"],
contract=".agent/rules/测试隔离.md",
)
def test_独立用例身份可绑定原生参数行__e2000b(tmp_path: Path) -> None:
项目 = 临时项目(tmp_path)
项目.写(
"test_参数.py",
"""
import pytest
@pytest.mark.parametrize("输入", [
pytest.param(1, id="first",
marks=pytest.mark.case_id("NC-first", given="第一输入", then=["等于1"])),
pytest.param(2, id="second",
marks=pytest.mark.case_id("NC-second", when="读取第二输入")),
])
def test_参数行为__abcdef(输入):
assert 输入 in (1, 2)
""",
)
for 身份, 参数 in [("NC-first", "first"), ("NC-second[second]", "second")]:
退出, 结果 = 项目.运行("--case", 身份)
assert 退出 == 0, 结果
行 = ET.parse(tmp_path / "报告.xml").findall(".//testcase")
assert len(行) == 1 and 行[0].attrib["name"].endswith(f"[{参数}]")
身份属性 = 行[0].find(".//property[@name='case_id']")
参数属性 = 行[0].find(".//property[@name='parameter_id']")
assert 身份属性 is not None and 身份属性.attrib["value"] == 身份.split("[")[0]
assert 参数属性 is not None and 参数属性.attrib["value"] == 参数
导出 = {条["case_id"]: 条 for 条 in 项目.收集()}
assert 导出["NC-first"]["then"] == ["等于1"]
assert "then" not in 导出["NC-second"]
@pytest.mark.case_id(
"TC-69440b7b4439",
environment="离线,受限子进程和临时测试项目",
when="收集并执行选定用例,模拟超时、缺依赖或零执行。",
contract=".agent/rules/测试隔离.md",
)
def test_generated_scope_rejects_unregistered_disk_asset__69440b(tmp_path: Path) -> None:
"""未标记的测试在实际收集时拒绝,不依赖旧清单。"""
项目 = 临时项目(tmp_path)
项目.写("test_登记.py", "def test_已登记__44000b(): assert True\n")
项目.写("test_未登记.py", "def test_未登记__55000c(): assert True\n", 登记身份=False)
with pytest.raises(ValueError, match="未标记 case_id.*test_未登记__55000c"):
项目.收集()
@pytest.mark.case_id(
"TC-20f4fa31f97b",
environment="离线,受限子进程和临时测试项目",
when="收集并执行选定用例,模拟超时、缺依赖或零执行。",
contract=".agent/rules/测试隔离.md",
)
def test_generated_scope_uses_manifest_scan_roots__20f4fa(tmp_path: Path) -> None:
"""自定义收集路径直接来自 pytest,不需登记扫描根。"""
项目 = 临时项目(tmp_path)
项目.写("custom-tests/test_自定义.py", "def test_自定义用例__66000d(): assert True\n")
条 = 项目.收集("custom-tests")[0]
assert 条["file"] == "custom-tests/test_自定义.py"
assert 条["symbol"] == "test_自定义用例__66000d"
@pytest.mark.case_id(
"TC-b4901fe62743",
environment="离线,受限子进程和临时测试项目",
when="收集并执行选定用例,模拟超时、缺依赖或零执行。",
contract=".agent/rules/测试隔离.md",
)
def test_generated_scope_rejects_manifest_extra_asset__b4901f(tmp_path: Path) -> None:
"""旧导出不能令已删除的用例仍被选中。"""
项目 = 临时项目(tmp_path)
项目.写("test_存在.py", "def test_存在__44000b(): assert True\n")
项目.写("用例清单.json", '{"cases":[{"case_id":"TC-removed","file":"test_已删除.py"}]}')
码, 结果 = 项目.运行("--case", "TC-removed")
assert 码 == 4
assert "未收集" in 结果["stderr"]
@pytest.mark.case_id(
"TC-5b8bb297e8c9",
environment="离线,受限子进程和临时测试项目",
when="收集并执行选定用例,模拟超时、缺依赖或零执行。",
contract=".agent/rules/测试隔离.md",
)
def test_generated_scope_accepts_existing_behavior_eval_entry__5b8bb2(tmp_path: Path) -> None:
"""仅收集真实模型层的身份和环境,不执行模型测试体。"""
项目 = 临时项目(tmp_path)
项目.写(
"tests/evals/test_评测.py",
"import pytest\n@pytest.mark.真实模型\ndef test_评测用例__88000f():\n"
" raise AssertionError('不能执行')\n",
)
条 = 项目.收集()[0]
assert 条["case_id"] == "TC-x-88000f000000"
assert "真实模型" in 条["markers"]
assert 条["file"] == "tests/evals/test_评测.py"
@pytest.mark.case_id(
"TC-fe3fba0485ff",
environment="离线,受限子进程和临时测试项目",
when="收集并执行选定用例,模拟超时、缺依赖或零执行。",
contract=".agent/rules/测试隔离.md",
)
def test_generated_scope_rejects_missing_behavior_eval_entry__fe3fba(tmp_path: Path) -> None:
"""实际删除的模型用例从当前收集退出,不能由历史导出复活。"""
项目 = 临时项目(tmp_path)
文件 = 项目.写(
"test_评测.py",
"import pytest\n@pytest.mark.真实模型\ndef test_评测__990010(): assert True\n",
)
assert len(项目.收集()) == 1
文件.unlink()
assert 项目.收集() == []
码, 结果 = 项目.运行("--case", "TC-x-990010000000")
assert 码 == 4
assert "未收集" in 结果["stderr"]
@pytest.mark.case_id(
"TC-14f941957440",
environment="离线,受限子进程和临时测试项目",
when="收集并执行选定用例,模拟超时、缺依赖或零执行。",
contract=".agent/rules/测试隔离.md",
)
def test_generated_scope_ignores_non_test_helpers_under_test_roots__14f941(tmp_path: Path) -> None:
"""普通辅助函数不产生测试身份要求。"""
项目 = 临时项目(tmp_path)
项目.写(
"tests/helpers/test_辅助.py",
"def 辅助函数(): return 1\ndef test_真用例__a10011(): assert 辅助函数() == 1\n",
)
assert [条["symbol"] for 条 in 项目.收集()] == ["test_真用例__a10011"]
@pytest.mark.case_id(
"TC-36f79ee6e164",
environment="离线,受限子进程和临时测试项目",
when="收集并执行选定用例,模拟超时、缺依赖或零执行。",
contract=".agent/rules/测试隔离.md",
)
def test_generated_scope_blocks_empty_and_print_only_scripts_before_child__36f79e(
tmp_path: Path,
) -> None:
"""print 型用例不能靠登记身份取得有断言的验证地位。"""
项目 = 临时项目(tmp_path)
项目.写("tests/单元/test_空断言.py", "def test_空用例__b20012(): print('只有输出')\n")
问题 = 检查用例身份(tmp_path)
assert any("无可判定语句" in p for p in 问题), 问题
@pytest.mark.case_id(
"TC-7c13823a1ef2",
environment="离线,受限子进程和临时测试项目",
when="收集并执行选定用例,模拟超时、缺依赖或零执行。",
contract=".agent/rules/测试隔离.md",
)
def test_generated_scope_accepts_supported_python_shapes__7c1382(tmp_path: Path) -> None:
"""函数式与类式身份及当前位置由真实 pytest 收集导出。"""
项目 = 临时项目(tmp_path)
项目.写(
"test_形状.py",
"class Test类式:\n def test_类方法__c30013(self): assert True\n"
"def test_函数式__d30014(): assert True\n",
)
条目 = 项目.收集()
assert {条["symbol"] for 条 in 条目} == {"test_类方法__c30013", "test_函数式__d30014"}
assert any("::Test类式::" in 节点 for 条 in 条目 for 节点 in 条["node_ids"])
@pytest.mark.case_id(
"NC-case-exact-id",
environment="离线,临时 pytest 项目",
given="临时项目内已登记的完整 ID、路径及测试符号",
when="选择用例或核对执行身份",
then=["未登记 ID 退出 4,不执行已登记用例"],
contract=".agent/rules/测试隔离.md",
)
def test_完整身份不接受冒用后缀__a02001(项目: 临时项目) -> None:
"""未登记 ID 与真实符号后缀相同,也不能选择执行。"""
项目.写("test_样本.py", "def test_目标__aa0001():\n assert True\n")
码, 结果 = 项目.运行("--case", "XX-forged-aa0001000000")
assert 码 == 4
assert "未收集" in 结果["stderr"]
@pytest.mark.case_id(
"NC-case-exact-node",
environment="离线,临时 pytest 项目",
given="临时项目内已登记的完整 ID、路径及测试符号",
when="选择用例或核对执行身份",
then=["文件及符号必须与登记一致,不能冒用后缀"],
contract=".agent/rules/测试隔离.md",
)
def test_完整身份绑定文件与符号__a02002(tmp_path: Path) -> None:
"""稳定身份可随文件/函数移动;同一身份绑定两函数则明确拒绝。"""
项目 = 临时项目(tmp_path)
项目.写(
"test_原例.py",
"import pytest\n@pytest.mark.case_id('NC-stable')\ndef test_原例(): assert True\n",
)
原 = 项目.收集()[0]
(tmp_path / "test_原例.py").unlink()
项目.写(
"test_改名.py",
"import pytest\n@pytest.mark.case_id('NC-stable')\ndef test_改名(): assert True\n",
)
新 = 项目.收集()[0]
assert 新["case_id"] == 原["case_id"]
assert 新["file"] == "test_改名.py" and 新["symbol"] == "test_改名"
项目.写(
"test_冒用.py",
"import pytest\n@pytest.mark.case_id('NC-stable')\ndef test_冒用(): assert True\n",
)
with pytest.raises(ValueError, match="同一 case_id 绑定不同函数"):
项目.收集()
@pytest.mark.case_id(
"NC-offline-libpq-process",
environment="隔离 PostgreSQL,内层离线进程",
given="真实可达的隔离 PostgreSQL 与无外部标记的临时测试",
when="分别从收集、测试体及子进程经 psycopg 连接",
then=["外部父测试 SELECT 1 成功", "三个离线子进程路径均由 OS 拒绝连接并非零退出"],
contract=".agent/rules/测试隔离.md",
)
@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"]
@pytest.mark.case_id(
"NC-case-implemented-target",
environment="离线",
given="已声明执行过的登记目标已消失",
when="执行默认索引检查",
then=["明确报告目标缺失"],
contract=".agent/rules/测试隔离.md",
)
@pytest.mark.parametrize("目录", ["架构", "迁移"])
def test_已执行目标丢失不被过渡期忽略__a02006(tmp_path: Path, 目录: str) -> None:
"""实际节点丢失就不能继续选择;恢复相同身份可继续精确选例。"""
项目 = 临时项目(tmp_path)
路径 = f"tests/{目录}/test_已执行.py"
文件 = 项目.写(
路径,
"import pytest\n@pytest.mark.case_id('NC-existing-test')\n"
"def test_已执行(): assert 1 == 1\n",
)
assert 项目.收集()[0]["case_id"] == "NC-existing-test"
文件.unlink()
码, 结果 = 项目.运行("--case", "NC-existing-test")
assert 码 == 4 and "未收集" in 结果["stderr"]
项目.写(
路径,
"import pytest\n@pytest.mark.case_id('NC-existing-test')\n"
"def test_已执行(): assert 1 == 1\n",
)
码, 结果 = 项目.运行("--case", "NC-existing-test")
assert 码 == 0, 结果