muse-agent-example/工具/用例身份.py

"""pytest 源码用例身份；清单只是收集结果的导出，不参与执行选择。"""

from __future__ import annotations

import json
import os
import subprocess
import sys
import tempfile
from pathlib import Path
from types import SimpleNamespace
from typing import Any

import pytest

_位置字段 = {"case_id", "file", "symbol", "parameter_ids", "node_ids", "fixtures", "markers"}


def 原样参数ID(显示名: str) -> Any:
    """按 pytest 的命名对象协议保留历史中文 ID，避免普通字符串被转义。

    pytest 运行时支持 __name__ 对象，但 ids 的类型声明未覆盖此分支。
    """
    return SimpleNamespace(__name__=显示名)


def 读取用例身份(项: pytest.Item) -> dict[str, Any] | None:
    """从函数或 pytest.param 的唯一 marker 读取原作者元信息。"""
    标记 = list(项.iter_markers("case_id"))
    if not 标记:
        return None
    if len(标记) != 1:
        raise pytest.UsageError(f"用例身份重叠：{项.nodeid}")
    标 = 标记[0]
    if len(标.args) != 1 or not isinstance(标.args[0], str):
        raise pytest.UsageError(f"case_id 必须是一个完整字符串：{项.nodeid}")
    身份 = 标.args[0]
    if (
        not 身份.startswith(("TC-", "NC-"))
        or len(身份) <= 3
        or any(字符.isspace() or 字符 in "[]" for 字符 in 身份)
    ):
        raise pytest.UsageError(f"case_id 无效：{身份!r}（{项.nodeid}）")
    if _位置字段.intersection(标.kwargs):
        raise pytest.UsageError(f"case_id 元信息不能登记派生位置：{项.nodeid}")
    try:
        元信息 = json.loads(json.dumps(标.kwargs, ensure_ascii=False, allow_nan=False))
    except (ValueError, TypeError) as exc:
        raise pytest.UsageError(f"case_id 元信息必须可导出 JSON：{项.nodeid}") from exc
    return {"case_id": 身份, **元信息}


def 导出用例(项目: list[pytest.Item], 根: Path) -> list[dict[str, Any]]:
    """聚合同一函数的参数节点；不填造未提供的 given/when/then。"""
    按身份: dict[str, dict[str, Any]] = {}
    绑定: dict[str, str] = {}
    for 项 in 项目:
        元信息 = 读取用例身份(项)
        if 元信息 is None:
            raise pytest.UsageError(f"未标记 case_id：{项.nodeid}")
        身份 = 元信息["case_id"]
        函数节点 = 项.nodeid.split("[", 1)[0]
        if 绑定.setdefault(身份, 函数节点) != 函数节点:
            raise pytest.UsageError(f"同一 case_id 绑定不同函数：{身份}")
        位置 = {
            "file": 项.path.relative_to(根).as_posix(),
            "symbol": getattr(项, "originalname", None) or 项.name.split("[", 1)[0],
        }
        条 = 按身份.setdefault(
            身份,
            {**元信息, **位置, "parameter_ids": [], "node_ids": [], "fixtures": [], "markers": []},
        )
        if any(条[键] != 值 for 键, 值 in {**元信息, **位置}.items()) or set(元信息) != (
            set(条) - (_位置字段 - {"case_id"})
        ):
            raise pytest.UsageError(f"同一 case_id 绑定不同函数或元信息：{身份}")
        参数 = getattr(getattr(项, "callspec", None), "id", None)
        if 参数 is not None:
            条["parameter_ids"].append(参数)
        条["node_ids"].append(项.nodeid)
        条["fixtures"].extend(getattr(项, "fixturenames", []))
        条["markers"].extend(标.name for 标 in 项.iter_markers() if 标.name != "case_id")
    for 条 in 按身份.values():
        for 键 in ("parameter_ids", "node_ids", "fixtures", "markers"):
            条[键] = sorted(set(条[键]))
    return sorted(按身份.values(), key=lambda 条: 条["case_id"])


def 选择用例(配置: pytest.Config, 项目: list[pytest.Item], 选择: list[str] | None = None) -> None:
    """按本次收集选择完整 ID 或 ID[参数 ID]，并保留 JUnit 身份。"""
    # 即使未指定 --case 也拒绝跨函数重号，不靠导出文件才发现身份冲突。
    导出用例(项目, 配置.rootpath)
    请求 = set(选择 if 选择 is not None else 配置.getoption("--case") or [])
    命中: set[str] = set()
    保留, 排除 = [], []
    for 项 in 项目:
        元信息 = 读取用例身份(项)
        对应: set[str] = set()
        if 元信息 is not None:
            身份 = 元信息["case_id"]
            参数 = getattr(getattr(项, "callspec", None), "id", None)
            项.user_properties[:] = [
                (键, 值) for 键, 值 in 项.user_properties if 键 not in {"case_id", "parameter_id"}
            ]
            项.user_properties.append(("case_id", 身份))
            对应.add(身份)
            if 参数 is not None:
                项.user_properties.append(("parameter_id", 参数))
                对应.add(f"{身份}[{参数}]")
        该项命中 = 请求 & 对应
        命中.update(该项命中)
        (保留 if not 请求 or 该项命中 else 排除).append(项)
    if 请求 - 命中:
        raise pytest.UsageError(f"--case 未收集到完整用例 ID 或参数：{sorted(请求 - 命中)}")
    if 排除:
        配置.hook.pytest_deselected(items=排除)
        项目[:] = 保留


def 收集用例(根: Path, 路径: list[str] | None = None) -> list[dict[str, Any]]:
    """独立 pytest 进程仅收集；临时导出不成为收集前置或仓内缓存。"""
    解释器 = 根 / ".venv/bin/python"
    if not 解释器.is_file():
        解释器 = Path(sys.executable)
    with tempfile.TemporaryDirectory(prefix="muse-case-collection-") as 目录:
        导出 = Path(目录) / "cases.json"
        命令 = [
            str(解释器),
            "-m",
            "pytest",
            "--collect-only",
            "-q",
            "-p",
            "no:cacheprovider",
            f"--导出用例={导出}",
            *(路径 or []),
        ]
        try:
            结果 = subprocess.run(
                命令,
                cwd=根,
                capture_output=True,
                text=True,
                timeout=180,
                env={**os.environ, "PYTHONDONTWRITEBYTECODE": "1"},
            )
        except (OSError, subprocess.TimeoutExpired) as exc:
            raise ValueError(f"pytest 收集未完成：{exc}") from exc
        if 结果.returncode not in (0, 5) or not 导出.is_file():
            raise ValueError(
                f"pytest 收集失败（exit {结果.returncode}）：\n{结果.stdout}\n{结果.stderr}"
            )
        try:
            内容 = json.loads(导出.read_text(encoding="utf-8"))
            if not isinstance(内容, dict) or not isinstance(内容.get("cases"), list):
                raise ValueError("导出缺少 cases 列表")
            return 内容["cases"]
        except (OSError, ValueError) as exc:
            raise ValueError(f"pytest 用例导出无效：{exc}") from exc