"""原生 pytest 的用例身份、进程隔离与共享夹具。 默认使用 OS 防护覆盖收集、驱动及子进程。外部环境必须显式选择； 完整 ID 绑定登记文件及符号，参数行和 Junit 保留可追溯身份。 本文件自包含，临时项目复制后验证相同测试入口。 """ from __future__ import annotations import ctypes import ctypes.util import errno import json import os import socket import sys from pathlib import Path from typing import Any import pytest # --------------------------------------------------------------------------- # 标记与用例选择 # --------------------------------------------------------------------------- 外部标记 = ("数据库", "网络", "真实模型", "浏览器", "宿主") # 执行计数：供会话结束守卫判断“无实际执行” _执行计数 = {"passed": 0, "skipped": 0, "failed": 0} def pytest_addoption(parser: pytest.Parser) -> None: 组 = parser.getgroup("muse", "Muse 用例选择") 组.addoption("--用例清单", default=None, help="显式的完整用例 ID 与目标符号清单") 组.addoption( "--外部环境", action="store_true", help="显式允许外部用例；仍按标记选择并使用隔离配置" ) 组.addoption( "--case", action="append", default=None, metavar="CASE_ID", help="按稳定用例 ID（TC-/NC- 前缀）选择；可多次给出，语义为并集", ) def pytest_configure(config: pytest.Config) -> None: for 名 in 外部标记: config.addinivalue_line("markers", f"{名}: 外部环境用例；见 tests/conftest.py") def _读取用例清单(config: pytest.Config) -> list[dict[str, Any]]: 指定 = config.getoption("--用例清单") 路径 = ( Path(指定) if 指定 else config.rootpath / "docs/系统架构/新版设计/验证设计/测试用例清单.json" ) if not 路径.exists() and not 指定 and not config.getoption("--case"): return [] try: 清单 = json.loads(路径.read_text(encoding="utf-8")) return [目标 for 旧例 in 清单["cases"] for 目标 in 旧例.get("target_cases", [])] + 清单[ "new_cases" ] except (OSError, ValueError, TypeError, KeyError) as exc: raise pytest.UsageError(f"用例清单无效：{路径.name}（{type(exc).__name__}）") from exc def pytest_collection_modifyitems(config: pytest.Config, items: list[pytest.Item]) -> None: 登记 = _读取用例清单(config) 选择 = config.getoption("--case") or [] 按ID = {条["case_id"]: 条 for 条 in 登记} 请求 = {str(值): str(值).split("[", 1)[0] for 值 in 选择} for 值, 身份 in 请求.items(): if 身份 not in 按ID: raise pytest.UsageError(f"--case {值} 未登记，必须使用完整用例 ID") 按目标: dict[tuple[str, str], list[dict]] = {} for 条 in 登记: 按目标.setdefault((条["file"], 条["symbol"]), []).append(条) 保留, 排除, 命中 = [], [], set() for 项 in items: 文件 = 项.path.relative_to(config.rootpath).as_posix() 符号 = getattr(项, "originalname", None) or 项.name.split("[", 1)[0] 参数 = getattr(getattr(项, "callspec", None), "id", None) 对应 = [ 条["case_id"] for 条 in 按目标.get((文件, 符号), []) if not 条.get("parameter_ids") or 参数 in 条["parameter_ids"] ] if len(对应) > 1: raise pytest.UsageError(f"用例身份重叠：{项.nodeid} 对应 {对应}") 身份 = 对应[0] if 对应 else None if 身份: 项.user_properties.append(("case_id", 身份)) if 参数 is not None: 项.user_properties.append(("parameter_id", 参数)) 该项命中 = { 值 for 值, id_ in 请求.items() if id_ == 身份 and ("[" not in 值 or 值 == f"{身份}[{参数}]") } 命中.update(该项命中) if not 选择 or 该项命中: 保留.append(项) else: 排除.append(项) if set(请求) - 命中: raise pytest.UsageError(f"--case 已登记但未收集：{sorted(set(请求) - 命中)}") if 排除: config.hook.pytest_deselected(items=排除) items[:] = 保留 def pytest_sessionstart(session: pytest.Session) -> None: """整个离线进程使用 OS 防护，收集、libpq 与子进程共用相同边界。""" if session.config.getoption("--外部环境"): return if sys.platform == "darwin": 库 = ctypes.CDLL("/usr/lib/libsandbox.dylib") 初始化 = 库.sandbox_init 初始化.argtypes = [ctypes.c_char_p, ctypes.c_uint64, ctypes.POINTER(ctypes.c_char_p)] 初始化.restype = ctypes.c_int 错误 = ctypes.c_char_p() if 初始化(b"(version 1)(allow default)(deny network*)", 0, ctypes.byref(错误)): raise pytest.UsageError("操作系统离线隔离初始化失败") elif sys.platform.startswith("linux"): 路径 = ctypes.util.find_library("seccomp") if not 路径: raise pytest.UsageError("离线测试需要系统 libseccomp；不能无防护运行") 库 = ctypes.CDLL(路径) 库.seccomp_init.argtypes = [ctypes.c_uint32] 库.seccomp_init.restype = ctypes.c_void_p 库.seccomp_syscall_resolve_name.argtypes = [ctypes.c_char_p] 库.seccomp_rule_add.argtypes = [ ctypes.c_void_p, ctypes.c_uint32, ctypes.c_int, ctypes.c_uint, ] 库.seccomp_load.argtypes = [ctypes.c_void_p] 库.seccomp_release.argtypes = [ctypes.c_void_p] 上下文 = 库.seccomp_init(0x7FFF0000) if not 上下文: raise pytest.UsageError("无法创建离线系统调用规则") try: # 禁止建立网络 socket 和所有外部 connect；保留事件循环的内部 socketpair。 class 参数条件(ctypes.Structure): _fields_ = [ ("arg", ctypes.c_uint), ("op", ctypes.c_int), ("datum_a", ctypes.c_uint64), ("datum_b", ctypes.c_uint64), ] 库.seccomp_rule_add_array.argtypes = [ ctypes.c_void_p, ctypes.c_uint32, ctypes.c_int, ctypes.c_uint, ctypes.POINTER(参数条件), ] 拒绝 = 0x00050000 | errno.EPERM connect = 库.seccomp_syscall_resolve_name(b"connect") socket_call = 库.seccomp_syscall_resolve_name(b"socket") if 库.seccomp_rule_add(上下文, 拒绝, connect, 0): raise pytest.UsageError("无法登记离线连接规则") for family in (socket.AF_INET, socket.AF_INET6): 条件 = 参数条件(0, 4, family, 0) # SCMP_CMP_EQ if 库.seccomp_rule_add_array(上下文, 拒绝, socket_call, 1, ctypes.byref(条件)): raise pytest.UsageError("无法登记离线网络规则") if 库.seccomp_load(上下文): raise pytest.UsageError("操作系统离线隔离初始化失败") finally: 库.seccomp_release(上下文) else: raise pytest.UsageError("当前系统尚无离线进程隔离实现；使用受支持的 macOS 或 Linux 环境") # --------------------------------------------------------------------------- # 离线防护：未显式选择外部环境的测试禁止一切 socket 连接 # --------------------------------------------------------------------------- class 离线连接被拒(RuntimeError): """离线测试尝试建立网络或数据库连接。""" 原始connect: Any = socket.socket.connect 原始connect_ex: Any = socket.socket.connect_ex def _拒绝连接(self: socket.socket, address: Any) -> Any: raise 离线连接被拒( f"离线测试尝试连接 {address!r}；需要外部环境时显式标记 {'/'.join(外部标记)} 并使用对应夹具" ) def _安装防护() -> None: socket.socket.connect = _拒绝连接 # type: ignore[method-assign] socket.socket.connect_ex = _拒绝连接 # type: ignore[method-assign] def _卸载防护() -> None: socket.socket.connect = 原始connect # type: ignore[method-assign] socket.socket.connect_ex = 原始connect_ex # type: ignore[method-assign] def _需要外部环境(项: pytest.Item) -> bool: 标记 = {m.name for m in 项.iter_markers()} return any(名 in 标记 for 名 in 外部标记) @pytest.fixture(autouse=True) def 离线防护(request: pytest.FixtureRequest): """默认给所有未选择外部环境的用例安装 socket 防护；用例结束后恢复。""" if _需要外部环境(request.node): yield return _安装防护() try: yield finally: _卸载防护() def pytest_runtest_logreport(report: pytest.TestReport) -> None: if report.when == "call": if report.passed: _执行计数["passed"] += 1 elif report.failed: _执行计数["failed"] += 1 elif report.skipped: _执行计数["skipped"] += 1 elif report.when == "setup" and report.skipped: _执行计数["skipped"] += 1 def pytest_sessionfinish(session: pytest.Session, exitstatus: int) -> None: """全跳过不算通过：没有实际执行时改写退出码为 5（无测试运行）。""" if exitstatus == 0 and _执行计数["passed"] == 0 and _执行计数["skipped"] > 0: session.exitstatus = 5 print( f"\n守卫：{_执行计数['skipped']} 个用例全部跳过、零实际执行；不计通过（退出码 5）", file=sys.stderr, ) # --------------------------------------------------------------------------- # 外部环境夹具 # --------------------------------------------------------------------------- 数据库环境变量 = "MUSE_TEST_DATABASE_URL" @pytest.fixture def 隔离数据库URL() -> str: """隔离 PostgreSQL 连接串；缺失时跳过（跳过不计通过）。""" 值 = os.environ.get(数据库环境变量, "").strip() if not 值: pytest.skip( f"未设置 {数据库环境变量}；隔离库用例未执行（不计通过）。启动本地库并导出该变量后重跑。" ) return 值 @pytest.fixture def 夹具根() -> Path: """tests/夹具 的绝对路径；供用例读取稳定回归输入。""" 根 = Path(__file__).resolve().parent / "夹具" assert 根.is_dir(), f"夹具目录不存在：{根}" return 根 @pytest.fixture def 应用测试库(隔离数据库URL: str, tmp_path: Path): """跨模块接入测试使用全新库；返回受控文件引用，全部数据为合成数据。""" import uuid import psycopg from psycopg import sql from muse.共享.调用身份 import 用途 from muse.基础设施.数据库.迁移 import 执行迁移 from muse.基础设施.数据库.连接 import 数据库工厂 from muse.配置 import 数据库引用 根 = Path(__file__).resolve().parents[1] 库名 = "muse_http_" + uuid.uuid4().hex[:12] with psycopg.connect(隔离数据库URL, autocommit=True) as 管理: 管理.execute((根 / "数据库/初始化/用途角色.sql").read_text()) 管理.execute(sql.SQL("CREATE DATABASE {} OWNER muse_maint").format(sql.Identifier(库名))) 工厂 = {} try: for 名, 角色 in [ (用途.维护, "muse_maint"), (用途.生产, "muse_app"), (用途.评测, "muse_eval"), ]: 参数 = psycopg.conninfo.conninfo_to_dict(隔离数据库URL) 参数.update(dbname=库名, user=角色) 文件 = tmp_path / f"{名.value}.txt" 文件.write_text(psycopg.conninfo.make_conninfo(**参数)) 文件.chmod(0o600) 工厂[名] = 数据库工厂(数据库引用("受控存储", str(文件)), 名) with 工厂[用途.维护].连接() as 连: 执行迁移(连, 根 / "数据库/迁移") yield 工厂 finally: with psycopg.connect(隔离数据库URL, autocommit=True) as 管理: 管理.execute(sql.SQL("DROP DATABASE {} WITH (FORCE)").format(sql.Identifier(库名)))