8a44364539
- 维护索引按 git 视角枚举（已跟踪 + 未跟踪且未忽略）：被 .gitignore 忽略的运行痕迹与 过程性检查报告不再进入目录派生或影响索引门禁；非 git 场景退回原磁盘枚举 - conftest 验证闸门在 `验证锁 --执行` 启动的会话内不再取共享锁，修复 `make 索引生成` 在独占锁内收集用例时的自锁（原 180s 超时失败 → 现 6s 通过） - .gitignore 登记过程性检查报告与工具会话缓存（.zcode/）：可复跑，不入历史 - 新增两条测试：忽略内容不参与目录派生、独占锁内收集不自锁
295 lines
12 KiB
Python
295 lines
12 KiB
Python
"""测试依赖与执行证据的逐例行为验证。
|
||
|
||
保护：用例选择、结构化报告、离线依赖阻断、超时与失败区分、
|
||
空收集与缺环境不静默、源身份与执行证据的拒绝语义。
|
||
|
||
环境：离线；受限子进程和临时测试项目（复制仓库真实 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"
|
||
sys.path.insert(0, str(仓库根 / "工具"))
|
||
from 用例身份 import 收集用例 # noqa: E402
|
||
from 维护索引 import 检查用例身份 # noqa: E402
|
||
|
||
|
||
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")
|
||
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={报告}", *参数]
|
||
环境变量 = {**_清理环境(), **(环境 or {})}
|
||
工具目录 = str(self.根 / "工具")
|
||
环境变量["PYTHONPATH"] = 工具目录 + (
|
||
(":" + 环境变量["PYTHONPATH"]) if 环境变量.get("PYTHONPATH") else ""
|
||
)
|
||
进程 = subprocess.run(
|
||
完整,
|
||
cwd=self.根,
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=180,
|
||
env=环境变量,
|
||
)
|
||
结果: 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-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-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(
|
||
"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-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(
|
||
"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"]
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"TC-3c9a17d84e60",
|
||
environment="离线，受限子进程和临时测试项目",
|
||
when="收集并执行选定用例，模拟超时、缺依赖或零执行。",
|
||
contract=".agent/rules/测试隔离.md",
|
||
)
|
||
def test_独占锁内启动的收集会话不自锁__3c9a17(项目: 临时项目, tmp_path: Path) -> None:
|
||
"""`验证锁 --执行` 已持独占锁时，子 pytest 会话不再取共享锁（否则自锁）。"""
|
||
项目.写("tests/单元/test_样本.py", "def test_样本__c10001():\n assert True\n")
|
||
环境变量 = {**_清理环境(), "MUSE_验证锁": str(tmp_path / "验证锁.lock")}
|
||
工具目录 = str(项目.根 / "工具")
|
||
环境变量["PYTHONPATH"] = 工具目录 + (
|
||
(":" + 环境变量["PYTHONPATH"]) if 环境变量.get("PYTHONPATH") else ""
|
||
)
|
||
进程 = subprocess.run(
|
||
[
|
||
sys.executable,
|
||
str(项目.根 / "工具/验证锁.py"),
|
||
"--执行",
|
||
sys.executable,
|
||
"-m",
|
||
"pytest",
|
||
"--collect-only",
|
||
"-q",
|
||
"-p",
|
||
"no:cacheprovider",
|
||
],
|
||
cwd=项目.根,
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=120,
|
||
env=环境变量,
|
||
)
|
||
assert 进程.returncode == 0, (进程.stdout, 进程.stderr)
|
||
assert "test_样本" in 进程.stdout
|