312 lines
12 KiB
Python
312 lines
12 KiB
Python
"""原生 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(库名)))
|