test：@test 闭环迁移 PowerShell→Python（run_boot_test/verify_log 纯标准库，文档同步）

2026-09-20 03:07:42 +08:00
parent 2375569f96
commit 8ae7a12016
18 changed files with 395 additions and 323 deletions
@@ -0,0 +1,218 @@
+#!/usr/bin/env python3
+"""run_boot_test.py - 一键开机日志抓取 + 校验（纯标准库，2026-09-20 替代 run_boot_test.ps1）。
+
+调用仓库内置串口助手（tools/serial_tool，常驻 daemon，子进程 CLI + JSON）:
+  1. open 串口 + clear 日志（本次运行独立代际）
+  2. 烧录（默认 MDK-ARM\\download.bat：下载 + Reset & Run）
+  3. wait-idle：RX 静默 >= settle-ms，或 max-wait-ms 超时
+  4. 从日志提取 [RX] 载荷行（剥掉 "[RX] <ISO时间戳> " 前缀）
+  5. verify_log.py 标记校验
+
+采集在烧录**之前**武装（open + clear 先于烧录），daemon 常驻全量日志兜底，
+无采集窗口/首字节等待仪式（与旧 PowerShell 版行为一致）。
+
+相对旧 PowerShell 版的加固（本机 Windows 环境实测教训，2026-09-20）:
+  - 子进程输出按**字节**读取、显式 UTF-8 解码 —— 不受控制台代码页（GBK）影响
+  - serial_config.json 按 UTF-8（容忍 BOM）解析 —— 无 PS 5.1 按 GBK 读 JSON
+    导致 "Unterminated string" 崩溃的问题
+  - 不经 PowerShell —— 无执行策略（scripts disabled）拦截
+  - 烧录带超时（默认 180s；旧版无超时，坏 SWD 连接会永久挂起）
+
+用法:
+  & '.venv\\Scripts\\python.exe' test/run_boot_test.py
+      [--port COM7] [--baud-rate 115200] [--settle-ms 8000] [--max-wait-ms 30000]
+      [--flash-cmd MDK-ARM\\download.bat] [--flash-timeout 180]
+      [--required "App init OK;Network Init;net init: OK"]
+      [--forbidden "FAIL;ERROR;HardFault"]
+      [--out-dir <目录>] [--show-tail-lines 15]
+
+  --port 缺省取 .opencode/serial_config.json 的 portName（单一配置源）
+
+退出码（与旧版一致）:
+  0 = PASS（采集成功 + Required 全中 + Forbidden 零命中）
+  1 = FAIL（标记缺失/命中）
+  2 = 采集或烧录错误
+  3 = 用法/预检错误
+"""
+
+import argparse
+import json
+import subprocess
+import sys
+import tempfile
+import time
+from datetime import datetime
+from pathlib import Path
+
+SCRIPT_DIR = Path(__file__).resolve().parent
+REPO_ROOT = SCRIPT_DIR.parent
+SERIAL_TOOL = REPO_ROOT / "tools" / "serial_tool" / "serial_tool.py"
+CFG_FILE = REPO_ROOT / ".opencode" / "serial_config.json"
+DEFAULT_FLASH = REPO_ROOT / "MDK-ARM" / "download.bat"
+
+REQUIRED_DEFAULT = "App init OK;Network Init;net init: OK"
+FORBIDDEN_DEFAULT = "FAIL;ERROR;HardFault"
+
+
+def _cli(cmd_args, timeout):
+    """调用串口助手 CLI，返回 (rc, stdout_text, parsed_json_or_None)。
+
+    全程字节级处理（capture_output 取 bytes 后显式 UTF-8 解码），
+    任何 Windows 控制台代码页都不参与数据路径。
+    """
+    try:
+        proc = subprocess.run(cmd_args, capture_output=True, timeout=timeout)
+    except subprocess.TimeoutExpired:
+        return 2, "client timeout ({}s)".format(timeout), None
+    text = proc.stdout.decode("utf-8", errors="replace")
+    try:
+        return proc.returncode, text, json.loads(text)
+    except ValueError:
+        return proc.returncode, text, None
+
+
+def _tool_cmd(py, cmd, *rest, timeout=15):
+    rc, out, data = _cli([py, str(SERIAL_TOOL), cmd, *rest], timeout)
+    if rc != 0 or data is None:
+        return None, out
+    return data, out
+
+
+def main(argv=None):
+    # 控制台输出固定 UTF-8：Windows 代码页只影响显示，不会让脚本崩溃
+    for stream in (sys.stdout, sys.stderr):
+        try:
+            stream.reconfigure(encoding="utf-8", errors="replace")
+        except Exception:
+            pass
+
+    ap = argparse.ArgumentParser(description="boot log capture + verify (pure stdlib, no PowerShell)")
+    ap.add_argument("--port", default="",
+                    help="serial port (default: portName from .opencode/serial_config.json)")
+    ap.add_argument("--baud-rate", type=int, default=115200)
+    ap.add_argument("--settle-ms", type=int, default=8000,
+                    help="RX silence that ends the capture")
+    ap.add_argument("--max-wait-ms", type=int, default=30000,
+                    help="overall wait-idle timeout")
+    ap.add_argument("--flash-cmd", default="",
+                    help="flash command (default: MDK-ARM/download.bat; pass reset.bat for reset-only)")
+    ap.add_argument("--flash-timeout", type=int, default=180,
+                    help="flash subprocess timeout in seconds")
+    ap.add_argument("--required", default=REQUIRED_DEFAULT,
+                    help="required markers, ';' separated")
+    ap.add_argument("--forbidden", default=FORBIDDEN_DEFAULT,
+                    help="forbidden patterns, ';' separated")
+    ap.add_argument("--out-dir", default=str(Path(tempfile.gettempdir()) / "opencode"))
+    ap.add_argument("--show-tail-lines", type=int, default=15)
+    args = ap.parse_args(argv)
+
+    # 串口助手 CLI 用当前解释器（AGENTS.md 统一入口 = 仓库 .venv python）
+    py = sys.executable
+
+    # 端口缺省：单一配置源 .opencode/serial_config.json（UTF-8，容忍 BOM）
+    port = args.port
+    if not port and CFG_FILE.is_file():
+        try:
+            cfg = json.loads(CFG_FILE.read_text(encoding="utf-8-sig"))
+            port = str(cfg.get("portName") or "")
+        except (OSError, ValueError) as e:
+            print("RUN_TEST_ERROR: cannot parse {}: {}".format(CFG_FILE, e))
+            return 3
+    if not port:
+        print("RUN_TEST_ERROR: no --port given and no portName in {}".format(CFG_FILE))
+        return 3
+
+    flash_cmd = Path(args.flash_cmd) if args.flash_cmd else DEFAULT_FLASH
+
+    # 预检（对应旧版 missing-file 检查）
+    for p in (SERIAL_TOOL, SCRIPT_DIR / "verify_log.py", flash_cmd):
+        if not p.is_file():
+            print("RUN_TEST_ERROR: missing {}".format(p))
+            return 3
+
+    out_dir = Path(args.out_dir)
+    out_dir.mkdir(parents=True, exist_ok=True)
+
+    stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+    boot_txt = out_dir / "boot_log_{}.txt".format(stamp)
+
+    def tool(cmd, *rest, timeout=15):
+        return _tool_cmd(py, cmd, *rest, timeout=timeout)
+
+    print("=== Boot test {} ===".format(stamp))
+    print("Port: {} @ {}  settle={}ms  max={}ms".format(
+        port, args.baud_rate, args.settle_ms, args.max_wait_ms))
+
+    # 1. open（幂等；未运行则自动拉起常驻 daemon）
+    data, out = tool("open", "--port", port, "--baud", str(args.baud_rate), "--timeout", "15")
+    if data is None:
+        print("RUN_TEST_ERROR: open failed: {}".format(out))
+        return 2
+
+    # 2. clear（新代际；此后数据才属于本次运行）
+    data, out = tool("clear", "--timeout", "15")
+    if data is None:
+        print("RUN_TEST_ERROR: clear failed: {}".format(out))
+        return 2
+
+    # 3. flash（下载 + Reset & Run）。烧录失败不中断：板子可能已复位，
+    #    由采集 + 标记判定结果（与旧版一致）
+    print("FLASH_START: {}".format(flash_cmd))
+    t0 = time.monotonic()
+    try:
+        proc = subprocess.run(["cmd", "/c", str(flash_cmd)],
+                              capture_output=True, timeout=args.flash_timeout)
+        flash_rc = proc.returncode
+        flash_bytes = (proc.stdout or b"") + (proc.stderr or b"")
+    except subprocess.TimeoutExpired as e:
+        flash_rc = -1
+        flash_bytes = (e.stdout or b"") + (e.stderr or b"")
+    elapsed_ms = int((time.monotonic() - t0) * 1000)
+    for line in flash_bytes.decode("utf-8", errors="replace").strip().splitlines()[-12:]:
+        print(line)
+    print("FLASH_RC: {}  elapsed_ms={}".format(flash_rc, elapsed_ms))
+    if flash_rc != 0:
+        print("RUN_TEST_WARN: flash rc={} (continuing: board may still have reset)".format(flash_rc))
+
+    # 4. wait-idle（RX 静默收尾或整体超时）
+    wait_timeout_s = args.max_wait_ms // 1000 + 20
+    data, out = tool("wait-idle",
+                     "--settle-ms", str(args.settle_ms),
+                     "--max-ms", str(args.max_wait_ms),
+                     "--timeout", str(wait_timeout_s),
+                     timeout=wait_timeout_s + 15)
+    if data is None:
+        print("RUN_TEST_ERROR: wait-idle failed: {}".format(out))
+        return 2
+    print("WAIT: reason={}  rx_bytes={}  waited_ms={}".format(
+        data.get("reason"), data.get("rx_bytes"), data.get("waited_ms")))
+    if data.get("reason") == "no_data":
+        print("RUN_TEST_ERROR: no RX data at all (check port / board power / flash)")
+        return 2
+
+    # 5. 提取 [RX] 载荷行（剥掉 "[RX] <ISO时间戳> " 前缀）
+    data, out = tool("log", "--tail", "0", "--timeout", "15")
+    if data is None:
+        print("RUN_TEST_ERROR: log read failed: {}".format(out))
+        return 2
+    rx = []
+    for l in data.get("lines", []):
+        if isinstance(l, str) and l.startswith("[RX] "):
+            rest = l[5:]
+            sp = rest.find(" ")
+            rx.append(rest[sp + 1:] if sp >= 0 else rest)
+    if not rx:
+        print("RUN_TEST_ERROR: no [RX] lines in log (device silent?)")
+        return 2
+    boot_txt.write_bytes(("\r\n".join(rx) + "\r\n").encode("utf-8"))
+    print("Boot log: {}  (lines={}  gen={})".format(boot_txt, len(rx), data.get("gen")))
+
+    # 6. 标记校验
+    from verify_log import verify
+    rc = verify(str(boot_txt), args.required, args.forbidden, args.show_tail_lines)
+    print("RUN_TEST_RC={}".format(rc))
+    return rc
+
+
+if __name__ == "__main__":
+    sys.exit(main())