# -*- 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()