sorrow404null
75b2ec4123
Subprocess-based runner with logic/offscreen groups: per-entry temp dirs, explicit timeouts, dependency/platform SKIP gates, UTF-8 safe output, and an exit code that reflects failures without blocking on a single crash or timeout.
266 lines
13 KiB
Python
266 lines
13 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""P2-04 跨平台聚合测试入口（无 pytest / npm / 网络要求）。
|
||
|
||
用法:
|
||
python tests/run_all.py --group logic
|
||
python tests/run_all.py --group offscreen
|
||
python tests/run_all.py --group all
|
||
python tests/run_all.py --group all --list
|
||
python tests/run_all.py --group all --keep-logs
|
||
python tests/run_all.py --group logic --only test_tool_params
|
||
|
||
设计（对应 REPAIR_BACKLOG.md P2-04 硬约束）:
|
||
* 保留所有独立命令：每个条目就是 [解释器, 测试文件] 的普通子进程调用，
|
||
与各测试文件头部的“运行:”命令等价（同一解释器、同一 cwd、同组环境）。
|
||
* 聚合入口不要求 pytest / npm / 网络；缺 node 或第三方依赖的条目的
|
||
呈现方式（SKIP + 理由），不伪装通过。
|
||
* 每个子测试使用独立临时目录（Windows: TEMP/TMP；POSIX: TMPDIR），
|
||
GUI 测试的临时配置/临时数据库由各测试自身在 import MainWindow 前
|
||
通过 tests/_test_env.isolate() 完成（仓库既有惯例）。
|
||
* 默认不运行 diag_* / verify_* / tune_* / 真实 API（smoke_live_*、
|
||
diag_live_*）/ 真实桌面 / 需要凭据的脚本，也不运行使用真实 DB 的
|
||
遗留脚本（smoke_persist、smoke_repro_real）。
|
||
* 平台不适用项以明确 SKIP + 理由呈现；平台共同项失败 → 退出码非零。
|
||
* 单个子测试崩溃或超时不阻止后续测试与最终汇总；退出码反映失败：
|
||
0 = 无 FAIL/TIMEOUT（SKIP 不影响）；1 = 存在 FAIL 或 TIMEOUT。
|
||
* 不读取真实配置、不访问网络、不删除用户运行数据。
|
||
"""
|
||
import argparse
|
||
import importlib.util
|
||
import os
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
import time
|
||
from datetime import datetime
|
||
|
||
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
PY = sys.executable # 用启动本入口的解释器跑所有 Python 子测试（与独立命令一致）
|
||
NODE = shutil.which("node")
|
||
|
||
# 控制台编码安全化（与 main.py 同惯例）：git-bash/GBK/cp1252 终端下
|
||
# 聚合器自身的中文输出不得崩溃（子测试输出经 UTF-8 解码后同样安全打印）。
|
||
for _name in ("stdout", "stderr"):
|
||
_s = getattr(sys, _name, None)
|
||
if _s is not None:
|
||
try:
|
||
_s.reconfigure(encoding="utf-8", errors="replace")
|
||
except Exception:
|
||
pass
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 测试清单 / 分组元数据
|
||
# (id, groups, argv, timeout_sec, needs, note)
|
||
# needs 取值:
|
||
# "node" —— 需要 node 可执行文件
|
||
# "openai" —— 需要 openai 包（当前解释器）
|
||
# "pymupdf" —— 需要 PyMuPDF (fitz)
|
||
# "pyqt6" —— 需要 PyQt6（GUI/offscreen 条目）
|
||
# ---------------------------------------------------------------------------
|
||
JS = lambda f: [NODE, os.path.join(REPO, "tests", f)] if NODE else [None]
|
||
PYT = lambda f: [PY, os.path.join(REPO, "tests", f)]
|
||
|
||
TESTS = [
|
||
# ---- logic：纯逻辑 / 无 GUI（平台共同项，失败不得豁免）----
|
||
("run_tests.py (agent core 41)", ("logic",), PYT("run_tests.py"), 300,
|
||
("openai",), "core.agent 单测套件（既有 run_tests.py 入口，原样保留）"),
|
||
("test_tool_params.py", ("logic",), PYT("test_tool_params.py"), 120,
|
||
("openai",), "工具参数解析（import core.agent.tools → openai）"),
|
||
("test_bash_stream.py", ("logic",), PYT("test_bash_stream.py"), 180,
|
||
("openai",), "bash 流式 / 进程树（P1-02、P2-01）"),
|
||
("test_copy_session.py", ("logic",), PYT("test_copy_session.py"), 180,
|
||
(), "会话复制 DB 逻辑"),
|
||
("test_compaction_persist.py", ("logic",), PYT("test_compaction_persist.py"), 180,
|
||
("openai",), "压缩持久化"),
|
||
("test_file_attach.py", ("logic",), PYT("test_file_attach.py"), 180,
|
||
(), "附件链路（P1-04）"),
|
||
("test_cross_platform_shell.py", ("logic",), PYT("test_cross_platform_shell.py"), 240,
|
||
("openai",), "跨平台 shell 契约 + 进程树终止（P1-02）"),
|
||
("test_global_hotkey_platforms.py", ("logic",), PYT("test_global_hotkey_platforms.py"), 240,
|
||
("pyqt6",), "全局热键平台矩阵 mock（P1-04；含 QShortcut/overlay 构造）"),
|
||
("test_wv2_guard.py", ("logic",), PYT("test_wv2_guard.py"), 240,
|
||
("platform:win32",), "WebView2 守卫（T0 事故锁死；msvcrt 单实例互斥为 Windows 专属，Linux 无 WebView2 链路）"),
|
||
("test_pdf_reader.py", ("logic",), PYT("test_pdf_reader.py"), 240,
|
||
("pymupdf",), "PDF 提取"),
|
||
("test_math_extract.js", ("logic",), JS("test_math_extract.js"), 120,
|
||
("node",), "数学提取 JS 单测（P1-03）"),
|
||
("test_render_window.js", ("logic",), JS("test_render_window.js"), 180,
|
||
("node",), "渲染窗口状态机 JS 单测（P1-01）"),
|
||
|
||
# ---- offscreen：Qt GUI（QT_QPA_PLATFORM=offscreen 子环境）----
|
||
("smoke_offscreen.py", ("offscreen",), PYT("smoke_offscreen.py"), 420,
|
||
("pyqt6",), "主窗口 + 核心链路冒烟"),
|
||
("smoke_mode.py", ("offscreen",), PYT("smoke_mode.py"), 420,
|
||
("pyqt6",), "模式切换（P0-02）"),
|
||
("smoke_copy_session.py", ("offscreen",), PYT("smoke_copy_session.py"), 420,
|
||
("pyqt6",), "会话复制 UI 端到端"),
|
||
("smoke_bash_panel.py", ("offscreen",), PYT("smoke_bash_panel.py"), 480,
|
||
("pyqt6",), "右侧 Bash 面板端到端（P2-01、P2-02）"),
|
||
("smoke_timeline.py", ("offscreen",), PYT("smoke_timeline.py"), 420,
|
||
("pyqt6",), "时间线持久化/还原"),
|
||
("smoke_midswitch.py", ("offscreen",), PYT("smoke_midswitch.py"), 420,
|
||
("pyqt6",), "流式中途切会话（P1-01）"),
|
||
("test_main_window_event_filter.py", ("offscreen",), PYT("test_main_window_event_filter.py"), 300,
|
||
("pyqt6",), "MainWindow.eventFilter（P0-02）"),
|
||
("test_config_isolation.py", ("offscreen",), PYT("test_config_isolation.py"), 300,
|
||
("pyqt6",), "配置/DB 隔离铁律（P0-01）"),
|
||
("test_error_persist.py", ("offscreen",), PYT("test_error_persist.py"), 300,
|
||
("pyqt6",), "错误持久化"),
|
||
("test_think_code_neutral.py", ("offscreen",), PYT("test_think_code_neutral.py"), 420,
|
||
("pyqt6",), "思考/代码块渲染中立性"),
|
||
("test_debug_window.py", ("offscreen",), PYT("test_debug_window.py"), 300,
|
||
("pyqt6",), "调试窗口"),
|
||
("test_renderer_matrix.py", ("offscreen",), PYT("test_renderer_matrix.py"), 600,
|
||
("pyqt6",), "渲染器矩阵（P1-03，含子进程并行）"),
|
||
("test_screen_capture_platforms.py", ("offscreen",), PYT("test_screen_capture_platforms.py"), 300,
|
||
("pyqt6",), "截图平台矩阵 mock + overlay 守卫（P1-04）"),
|
||
]
|
||
|
||
# 默认不聚合（人工/半自动/真实资源）：diag_*、verify_*、tune_*、
|
||
# smoke_live_guard.py、diag_live_*.py、smoke_manual.py、smoke_repro_real.py、
|
||
# smoke_persist.py（真实 DB）、smoke_probe.py、_probe_*.py、
|
||
# debug_inject.py、inject_math_demo.py、check_db_migration.py
|
||
|
||
|
||
def _has_module(name):
|
||
try:
|
||
return importlib.util.find_spec(name) is not None
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
def skip_reason(entry):
|
||
"""返回 SKIP 理由；None 表示可运行。"""
|
||
_id, _groups, _argv, _to, needs, _note = entry
|
||
for need in needs:
|
||
if need.startswith("platform:"):
|
||
want = need.split(":", 1)[1]
|
||
if sys.platform != want:
|
||
return f"平台不适用：该条目仅 {want} 路径适用（当前 {sys.platform}）"
|
||
continue
|
||
if need == "node" and not NODE:
|
||
return "node 不可用（聚合入口不要求 npm/node；安装 node 后自动纳入）"
|
||
if need == "openai" and not _has_module("openai"):
|
||
return "当前解释器无 openai 包（本环境不允许联网安装）"
|
||
if need == "pymupdf" and not _has_module("fitz"):
|
||
return "当前解释器无 PyMuPDF（本环境不允许联网安装）"
|
||
if need == "pyqt6" and not _has_module("PyQt6"):
|
||
return "当前解释器无 PyQt6（GUI 条目需在装有依赖的解释器运行）"
|
||
return None
|
||
|
||
|
||
def _child_env(extra_temp, offscreen):
|
||
env = os.environ.copy()
|
||
env["PYTHONIOENCODING"] = "utf-8"
|
||
env["PYTHONUNBUFFERED"] = "1"
|
||
if extra_temp:
|
||
if os.name == "nt":
|
||
env["TEMP"] = extra_temp
|
||
env["TMP"] = extra_temp
|
||
else:
|
||
env["TMPDIR"] = extra_temp
|
||
if offscreen:
|
||
env.setdefault("QT_QPA_PLATFORM", "offscreen")
|
||
env.setdefault("HAOCODE_RENDER", "software")
|
||
env.setdefault("QTWEBENGINE_CHROMIUM_FLAGS", "--disable-gpu")
|
||
return env
|
||
|
||
|
||
def run_group(group, keep_logs, only):
|
||
entries = [e for e in TESTS if group in ("all",) or e[1][0] == group]
|
||
if only:
|
||
subs = [s.strip() for s in only.split(",") if s.strip()]
|
||
entries = [e for e in entries if any(s in e[0] for s in subs)]
|
||
if not entries:
|
||
print("没有匹配的测试条目（--group/--only 组合为空）")
|
||
return 1
|
||
|
||
run_dir = tempfile.mkdtemp(prefix="haocode_runall_")
|
||
print("=" * 72)
|
||
print("haocode 聚合测试入口 (P2-04)")
|
||
print(f"platform={sys.platform} python={sys.version.split()[0]} ({PY})")
|
||
print(f"group={group} entries={len(entries)} 开始={datetime.now():%H:%M:%S}")
|
||
print("=" * 72)
|
||
|
||
results = [] # (id, status, seconds, note, argv)
|
||
t_all = time.time()
|
||
for entry in entries:
|
||
tid, _groups, argv, budget, needs, note = entry
|
||
off = "offscreen" in entry[1]
|
||
reason = skip_reason(entry)
|
||
if reason:
|
||
results.append((tid, "SKIP", 0.0, reason, argv))
|
||
print(f" SKIP {tid} [{reason}]")
|
||
continue
|
||
t0 = time.time()
|
||
log_path = os.path.join(run_dir, tid.replace(" ", "_").replace("(", "").replace(")", "") + ".log")
|
||
try:
|
||
with tempfile.TemporaryDirectory(prefix="runall_t_") as td:
|
||
proc = subprocess.run(
|
||
argv, cwd=REPO, env=_child_env(td, off),
|
||
stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
|
||
timeout=budget)
|
||
rc, err = proc.returncode, None
|
||
out = proc.stdout
|
||
except subprocess.TimeoutExpired as e:
|
||
rc, out, err = 124, (e.stdout or b""), f"超时（预算 {budget}s）"
|
||
except Exception as e: # 聚合器自身健壮性：单条崩溃不阻止汇总
|
||
rc, out, err = 125, b"", f"聚合器异常: {e!r}"
|
||
dt = max(0.0, time.time() - t0) # WSL/9P 时钟回拨保护
|
||
status = "PASS" if rc == 0 else ("TIMEOUT" if rc == 124 else "FAIL")
|
||
results.append((tid, status, dt, note, argv))
|
||
tag = f" {status:<7} {dt:7.1f}s {tid}"
|
||
print(tag, flush=True)
|
||
# 非 PASS：打印最后 15 行帮助定位
|
||
tail = (out or b"").decode("utf-8", "replace").splitlines()[-15:]
|
||
if status != "PASS" and tail:
|
||
for line in tail:
|
||
print(f" | {line}")
|
||
with open(log_path, "wb") as f:
|
||
f.write(b"# argv: " + " ".join(str(a) for a in argv).encode() + b"\n")
|
||
if err:
|
||
f.write(f"# {err}\n".encode())
|
||
f.write(out or b"")
|
||
|
||
dt_all = time.time() - t_all
|
||
n_pass = sum(1 for r in results if r[1] == "PASS")
|
||
n_fail = sum(1 for r in results if r[1] in ("FAIL", "TIMEOUT"))
|
||
n_skip = sum(1 for r in results if r[1] == "SKIP")
|
||
print("-" * 72)
|
||
print(f"总计 {len(results)}: PASS {n_pass} FAIL {n_fail} SKIP {n_skip}"
|
||
f" 总耗时 {dt_all:.1f}s")
|
||
if keep_logs or n_fail:
|
||
print(f"日志目录（保留）: {run_dir}")
|
||
else:
|
||
shutil.rmtree(run_dir, ignore_errors=True)
|
||
print("日志目录（无失败，已清理）")
|
||
bad = [r for r in results if r[1] in ("FAIL", "TIMEOUT")]
|
||
if bad:
|
||
print("失败命令:")
|
||
for tid, status, dt, _note, argv in bad:
|
||
print(f" [{status}] {' '.join(str(a) for a in argv)}")
|
||
print("=" * 72)
|
||
return 1 if n_fail else 0
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser(description="haocode 跨平台聚合测试入口（P2-04）")
|
||
ap.add_argument("--group", choices=("logic", "offscreen", "all"), default="all",
|
||
help="logic=纯逻辑; offscreen=Qt GUI offscreen; all=两者（默认）")
|
||
ap.add_argument("--list", action="store_true", help="只列出条目，不执行")
|
||
ap.add_argument("--keep-logs", action="store_true", help="保留全部子测试日志")
|
||
ap.add_argument("--only", default="", help="只运行 id 含子串的条目（逗号分隔）")
|
||
args = ap.parse_args()
|
||
|
||
if args.list:
|
||
for tid, groups, argv, budget, needs, note in TESTS:
|
||
mark = "SKIP?" if skip_reason((tid, groups, argv, budget, needs, note)) else "run "
|
||
print(f" [{mark}] {groups[0]:<10} {budget:>4}s {tid} {note}")
|
||
return 0
|
||
sys.exit(run_group(args.group, args.keep_logs, args.only))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|