78 lines
4.4 KiB
Python
78 lines
4.4 KiB
Python
# ============================================================
|
||
# 黄金测试看板数据源（moduleId: gateway-golden, 可重生 ✅）
|
||
# plan.md §6.10.3 重生中心：读缓存展示最近一次 pytest 结果；
|
||
# ?run=true 时在子进程同步重跑（测试自隔离 tmp_path，不碰生产数据）。
|
||
# 权力等级：读缓存 P0；run=true 触发跑测仍无业务副作用（只写缓存文件）。
|
||
# ============================================================
|
||
from __future__ import annotations # 前向类型引用
|
||
|
||
import json # 缓存序列化
|
||
import os # 路径
|
||
import re # pytest 输出解析
|
||
import subprocess # 子进程跑 pytest
|
||
import sys # 当前解释器路径
|
||
from datetime import datetime # 跑测时间戳
|
||
from typing import Any # 类型标注
|
||
|
||
from server.timeutil import fmt_dt # 时间格式化
|
||
|
||
# 缓存文件路径（与世界状态同目录；进 .gitignore 的数据区）
|
||
_CACHE_PATH = os.environ.get("APS_GOLDEN_CACHE", "server/data/golden_tests.json")
|
||
# pytest 逐用例行：tests/golden/test_x.py::test_name PASSED/FAILED
|
||
_CASE_RE = re.compile(r"(tests[/\\]golden[/\\]\S+?)::(\S+?)\s+(PASSED|FAILED|ERROR)")
|
||
|
||
|
||
def _read_cache() -> dict[str, Any] | None:
|
||
"""读最近一次跑测缓存；缺失/损坏返回 None。"""
|
||
try:
|
||
with open(_CACHE_PATH, "r", encoding="utf-8") as f: # 读缓存
|
||
return json.load(f) # 解析
|
||
except (FileNotFoundError, json.JSONDecodeError): # 无缓存或损坏
|
||
return None
|
||
|
||
|
||
def _write_cache(data: dict[str, Any]) -> None:
|
||
"""落盘缓存（覆盖写；非关键数据无需原子写）。"""
|
||
os.makedirs(os.path.dirname(_CACHE_PATH) or ".", exist_ok=True) # 确保目录
|
||
with open(_CACHE_PATH, "w", encoding="utf-8") as f: # 覆盖写
|
||
json.dump(data, f, ensure_ascii=False, indent=1) # 序列化
|
||
|
||
|
||
def run_golden() -> dict[str, Any]:
|
||
"""子进程同步跑 `pytest tests/golden -v --tb=no` 并解析结果（更新缓存）。"""
|
||
proc = subprocess.run( # 独立进程（测试用 tmp_path 自隔离）
|
||
[sys.executable, "-m", "pytest", "tests/golden", "-v", "--tb=no"], # -v 输出逐用例行（勿加 -q 会压掉）
|
||
capture_output=True, text=True, timeout=120, # 限时防挂
|
||
)
|
||
out = proc.stdout + proc.stderr # 合并输出
|
||
cases = [{ # 逐用例结果
|
||
"file": m.group(1).replace("\\", "/"), # 测试文件
|
||
"name": m.group(2), # 用例名
|
||
"outcome": m.group(3), # PASSED/FAILED/ERROR
|
||
} for m in _CASE_RE.finditer(out)]
|
||
passed = sum(1 for c in cases if c["outcome"] == "PASSED") # 通过数
|
||
failed = len(cases) - passed # 未通过数
|
||
data = { # 看板数据结构
|
||
"ranAt": fmt_dt(datetime.now()), # 跑测时间
|
||
"passed": passed, "failed": failed, "total": len(cases), # 汇总
|
||
"exitCode": proc.returncode, # pytest 退出码（0=全绿）
|
||
"cases": cases, # 用例明细
|
||
}
|
||
_write_cache(data) # 更新缓存
|
||
return data # 返回
|
||
|
||
|
||
def golden_status(run: bool = False) -> dict[str, Any]:
|
||
"""看板入口：run=true 重跑；否则读缓存（无缓存时返回未运行占位）。"""
|
||
if run: # 显式要求重跑
|
||
try:
|
||
return run_golden() # 跑测并返回
|
||
except Exception as exc: # 跑测失败（环境问题）
|
||
return {"ranAt": None, "passed": 0, "failed": 0, "total": 0,
|
||
"exitCode": -1, "cases": [], "error": str(exc)}
|
||
cached = _read_cache() # 读缓存
|
||
if cached: # 有缓存直接用
|
||
return cached
|
||
return {"ranAt": None, "passed": 0, "failed": 0, "total": 0, # 从未跑过的占位
|
||
"exitCode": None, "cases": []}
|