291 lines
13 KiB
Python
291 lines
13 KiB
Python
#!/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 标记校验
|
||
6. （仅当 --require-selftest-pass）自检记录联合判定（S7）：
|
||
--selftest-log 提供本地副本则直接用它，否则从设备 FTP 抓取 /log/selftest.log
|
||
（lftpd :21 匿名）；判定：文件缺失 / 0 字节 = 无故障 → PASS，有 64B 记录 = FAIL
|
||
|
||
采集在烧录**之前**武装（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/host/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;[ERR]"]
|
||
[--require-suite-summary "Storage Tests"]
|
||
[--require-selftest-pass] [--selftest-log <本地副本>]
|
||
[--device-ip 192.168.1.100] [--out-dir <目录>] [--show-tail-lines 15]
|
||
|
||
--port 缺省取 .opencode/serial_config.json 的 portName（单一配置源）
|
||
|
||
退出码（与旧版一致）:
|
||
0 = PASS（采集成功 + Required 全中 + Forbidden 零命中）
|
||
1 = FAIL（标记缺失/命中）
|
||
2 = 采集或烧录错误
|
||
3 = 用法/预检错误
|
||
"""
|
||
|
||
import argparse
|
||
import ftplib
|
||
import json
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
import time
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
SCRIPT_DIR = Path(__file__).resolve().parent
|
||
# 本脚本位于 test/host/，仓库根需上溯两级（2026-10 实机窗口修：原 .parent 只到 test/，
|
||
# 导致 .opencode/serial_config.json 与 tools/serial_tool 路径解析错误）
|
||
REPO_ROOT = SCRIPT_DIR.parent.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;[ERR]"
|
||
|
||
|
||
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 _ftp_quit(ftp):
|
||
"""尽力关闭 FTP 控制连接（QUIT 失败不影响判定）。"""
|
||
try:
|
||
ftp.quit()
|
||
except (OSError, ftplib.Error):
|
||
pass
|
||
|
||
|
||
def _fetch_selftest_log(device_ip, out_dir, stamp):
|
||
"""从设备 FTP 抓取 /log/selftest.log（lftpd :21 匿名，根目录 = 卷根）。
|
||
|
||
成功返回本地路径字符串；失败（网络不可达 / FTP 错误 / 设备侧文件缺失）返回 ""
|
||
（按 S7 语义视为无故障记录；此处打 WARN，调用方不判整体 FAIL）。
|
||
|
||
2026-10-02 台架修复：原实现只 `except OSError`，而 ftplib 的 error_temp /
|
||
error_perm 继承自 Exception 而非 OSError —— 台账"仅故障才写"，干净设备上
|
||
/log/selftest.log 根本不存在，RETR 返回 450 直接抛出未捕获异常，令整个
|
||
`@test --require-selftest-pass` 在最后一步崩掉（而它本该是 PASS 路径）。
|
||
现把"设备侧无此文件"单列为正常路径（INFO），其余 FTP 错误保留 WARN。
|
||
"""
|
||
local = Path(out_dir) / "selftest_log_{}.log".format(stamp)
|
||
ftp = ftplib.FTP()
|
||
try:
|
||
ftp.connect(device_ip, 21, timeout=10)
|
||
ftp.login() # 匿名（lftpd 接受 USER anonymous / 空 PASS）
|
||
except (OSError, ftplib.Error) as e:
|
||
print("RUN_TEST_WARN: selftest fetch failed: {} "
|
||
"(按 S7 语义视为无记录；设备网络/FTP 需另行排查)".format(e))
|
||
return ""
|
||
|
||
buf = bytearray()
|
||
try:
|
||
ftp.retrbinary("RETR /log/selftest.log", buf.extend)
|
||
except ftplib.all_errors as e:
|
||
# 450/550 = 设备侧无此文件：干净台账，S7 判 PASS（正常路径，非故障）
|
||
print("SELFTEST_FETCH: {} -> (ledger absent: {}) 按 S7 视为无故障记录"
|
||
.format(device_ip, str(e).strip()))
|
||
_ftp_quit(ftp)
|
||
return ""
|
||
|
||
_ftp_quit(ftp)
|
||
local.write_bytes(bytes(buf))
|
||
print("SELFTEST_FETCH: {} -> {} ({} bytes)".format(device_ip, local, len(buf)))
|
||
return str(local)
|
||
|
||
|
||
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("--require-suite-summary", default="",
|
||
help="suite names that MUST appear as '=== <name>: N/M PASSED (failed=0) ===' "
|
||
"with N>0, ';' separated (T-2). Empty = off")
|
||
ap.add_argument("--require-selftest-pass", action="store_true",
|
||
help="jointly judge /log/selftest.log (S7): missing/empty = PASS, "
|
||
"any 64B record = FAIL")
|
||
ap.add_argument("--selftest-log", default="",
|
||
help="local copy of /log/selftest.log (skips device-side fetch when given)")
|
||
ap.add_argument("--device-ip", default="192.168.1.100",
|
||
help="device IP for fetching /log/selftest.log via FTP (lftpd anonymous)")
|
||
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")))
|
||
|
||
# 5.5 自检记录抓取（S7）：启用且未提供本地副本时才走 FTP
|
||
selftest_local = args.selftest_log
|
||
if args.require_selftest_pass and not selftest_local:
|
||
selftest_local = _fetch_selftest_log(args.device_ip, out_dir, stamp)
|
||
|
||
# 6. 标记校验（含可选的自检记录联合判定）
|
||
from verify_log import verify
|
||
rc = verify(str(boot_txt), args.required, args.forbidden, args.show_tail_lines,
|
||
suite_summary=args.require_suite_summary,
|
||
selftest_pass=args.require_selftest_pass,
|
||
selftest_log=selftest_local)
|
||
print("RUN_TEST_RC={}".format(rc))
|
||
return rc
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|