K1Guitar/tools/k1_harness/suites/app_protocol.py

# -*- coding: utf-8 -*-
"""app 套件：App SysEx 全协议（组 01~06）。

- transport=com: 进程内复用 test_protocol_app_sim.run_tests（UART4 桥）
- transport=ble: 子进程跑 test_protocol_ble.py（ bleak + BLE-MIDI framing），
  解析其 [PASS]/[FAIL]/[SKIP]/[SENT] 输出行进统一报告
"""
from __future__ import annotations

import os
import re
import subprocess
import sys

from ..paths import TOOLS_DIR

ROW_RE = re.compile(r"^\[(PASS|FAIL|SKIP|SENT)\s*\]\s*(.+?)(?:\s*\|\s*(.*))?$")


def _absorb_output(report, suite: str, text: str, stderr: str = "") -> None:
    n = 0
    for line in text.splitlines():
        m = ROW_RE.match(line.strip())
        if m:
            report.add(suite, m.group(2), m.group(1), m.group(3) or "")
            n += 1
    if n == 0:
        blob = (text or "") + "\n" + (stderr or "")
        if "未找到设备" in blob or "扫描 BLE" in blob:
            report.add(
                suite,
                "BLE 扫描连接",
                "FAIL",
                "未发现 Smart Guitar MIDI（确认已开机且蓝牙开；勿在软关机态跑 BLE）",
            )
            return
        tail = "\\n".join((text or "").splitlines()[-5:])
        report.add(suite, "子进程输出解析", "FAIL", f"未找到结果行；tail: {tail}")


def run(report, ctx) -> None:
    suite = "app"
    if ctx.transport == "com":
        if not ctx.port:
            report.add(suite, "COM 连接", "FAIL", "transport=com 需要 --port COMx")
            return
        import test_protocol_app_sim as appsim

        if appsim.serial is None:
            report.add(suite, "pyserial", "FAIL", "pip install pyserial")
            return
        sim = appsim.AppSim(ctx.port, ctx.baud)
        try:
            res = appsim.run_tests(sim, allow_poweroff=ctx.allow_poweroff)
        finally:
            sim.close()
        report.extend_rows(suite, res.rows)
        return

    # BLE：子进程调 test_protocol_ble.py（其内部已处理 bleak 异步/解配对/报告）
    cmd = [
        sys.executable, os.path.join(TOOLS_DIR, "test_protocol_ble.py"),
        "--transport", "midi", "--no-unpair",
    ]
    if ctx.allow_poweroff:
        cmd.append("--allow-poweroff")
    proc = subprocess.run(cmd, capture_output=True, text=True, timeout=600)
    _absorb_output(report, suite, proc.stdout, proc.stderr or "")
    if proc.returncode != 0 and not any(c.suite == suite and c.status == "FAIL" for c in report.cases):
        report.add(suite, "BLE 协议子进程", "FAIL", f"exit={proc.returncode} {proc.stderr[-300:]}")