71 lines
2.5 KiB
Python
71 lines
2.5 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""app 套件：App SysEx 全协议（组 01~07）。
|
||
|
||
- 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:]}")
|