1913 lines
69 KiB
Python
1913 lines
69 KiB
Python
#!/usr/bin/env python3
|
|
#
|
|
# The MIT License (MIT)
|
|
#
|
|
# Copyright (c) 2023 HiFiPhile
|
|
#
|
|
# Permission is hereby granted, free of charge, to any person obtaining a copy
|
|
# of this software and associated documentation files (the "Software"), to deal
|
|
# in the Software without restriction, including without limitation the rights
|
|
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
|
# copies of the Software, and to permit persons to whom the Software is
|
|
# furnished to do so, subject to the following conditions:
|
|
#
|
|
# The above copyright notice and this permission notice shall be included in
|
|
# all copies or substantial portions of the Software.
|
|
#
|
|
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
|
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
|
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
|
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
|
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
|
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
|
|
# THE SOFTWARE.
|
|
|
|
# Host setup (required: a missing tool fails its test rather than skipping it):
|
|
# - System packages: sudo apt install mtools libmtp9 alsa-utils iperf
|
|
# mtools - read_disk_file (device/cdc_msc, device/msc_dual_lun)
|
|
# libmtp9 - pymtp ctypes load (device/mtp); Debian 13 uses libmtp9t64
|
|
# alsa-utils - arecord (device/audio_test_freertos)
|
|
# iperf - throughput tests (device/net_lwip_*)
|
|
# - Python packages: pip install -r requirements.txt
|
|
#
|
|
# udev rules :
|
|
# ACTION=="add", SUBSYSTEM=="tty", SUBSYSTEMS=="usb", MODE="0666", PROGRAM="/bin/sh -c 'echo $$ID_SERIAL_SHORT | rev | cut -c -8 | rev'", SYMLINK+="ttyUSB_%c.%s{bInterfaceNumber}"
|
|
# ACTION=="add", SUBSYSTEM=="block", SUBSYSTEMS=="usb", ENV{ID_FS_USAGE}=="filesystem", MODE="0666", PROGRAM="/bin/sh -c 'echo $$ID_SERIAL_SHORT | rev | cut -c -8 | rev'", RUN{program}+="/usr/bin/systemd-mount --no-block --automount=yes --collect $devnode /media/blkUSB_%c.%s{bInterfaceNumber}"
|
|
|
|
import argparse
|
|
import io
|
|
import os
|
|
import random
|
|
import re
|
|
import select
|
|
import sys
|
|
import time
|
|
import signal
|
|
from contextlib import redirect_stdout
|
|
from pathlib import Path
|
|
from typing import Any, TypedDict, NotRequired, cast
|
|
|
|
import serial
|
|
import subprocess
|
|
import json
|
|
import glob
|
|
from multiprocessing import Pool, Lock
|
|
from multiprocessing import TimeoutError as MpTimeoutError
|
|
import hashlib
|
|
import ctypes
|
|
from pymtp import MTP
|
|
import string
|
|
|
|
ENUM_TIMEOUT = 15
|
|
|
|
STATUS_OK = "\033[32mOK\033[0m"
|
|
STATUS_FAILED = "\033[31mFailed\033[0m"
|
|
STATUS_SKIPPED = "\033[33mSkipped\033[0m"
|
|
|
|
# Plain (non-ANSI) cell symbols for the markdown matrix report (hil_report.md).
|
|
# A missing binary is reported as skipped too.
|
|
REPORT_CELL = {'pass': '✅', 'fail': '❌', 'skip': '⚪'}
|
|
|
|
verbose = False
|
|
test_only = []
|
|
board_test = {}
|
|
build_dir = 'cmake-build'
|
|
skip_flash = False
|
|
print_lock = None
|
|
|
|
|
|
def init_worker(lock):
|
|
global print_lock
|
|
print_lock = lock
|
|
|
|
|
|
def log_line(msg: str) -> None:
|
|
out = sys.__stdout__ if sys.__stdout__ is not None else sys.stdout
|
|
if print_lock is not None:
|
|
with print_lock:
|
|
print(msg, file=out, flush=True)
|
|
else:
|
|
print(msg, file=out, flush=True)
|
|
|
|
|
|
def compact_output(raw: str) -> str:
|
|
if not raw:
|
|
return ''
|
|
lines = [ln.strip() for ln in raw.replace('\r', '\n').split('\n') if ln.strip()]
|
|
return ' | '.join(lines)
|
|
|
|
class FlasherCfg(TypedDict):
|
|
name: str
|
|
uid: str
|
|
args: str
|
|
|
|
|
|
class AttachedDevCfg(TypedDict, total=False):
|
|
vid_pid: str
|
|
serial: str
|
|
is_cdc: bool
|
|
is_msc: bool
|
|
block_count: int
|
|
block_size: int
|
|
|
|
|
|
class TestsCfg(TypedDict, total=False):
|
|
device: bool
|
|
dual: bool
|
|
host: bool
|
|
only: list[str]
|
|
skip: list[str]
|
|
dev_attached: list[AttachedDevCfg]
|
|
|
|
|
|
class BuildCfg(TypedDict, total=False):
|
|
args: list[str]
|
|
|
|
|
|
class VariantCfg(TypedDict, total=False):
|
|
name: str # build dir (cmake-build-<name>) and HIL report row
|
|
flags: str # raw CFLAGS, e.g. "-DCFG_TUD_DWC2_DMA_ENABLE=1"
|
|
defines: list[str] # cmake -D defines, e.g. ["RHPORT_DEVICE=1"] (vs flags which are compiler-only)
|
|
|
|
|
|
class Board(TypedDict):
|
|
name: str
|
|
uid: str
|
|
tests: TestsCfg
|
|
flasher: FlasherCfg
|
|
build: NotRequired[BuildCfg]
|
|
variant: NotRequired[list[VariantCfg]]
|
|
toolchain: NotRequired[str] # CI build bucket override, e.g. "riscv-gcc" (consumed by hil_ci_set_matrix.py)
|
|
|
|
|
|
class HilConfig(TypedDict):
|
|
boards: list[Board]
|
|
|
|
CMD_TIMEOUT = int(os.getenv('HIL_CMD_TIMEOUT', '180'))
|
|
POOL_TIMEOUT = int(os.getenv('HIL_POOL_TIMEOUT', '3000'))
|
|
SERIAL_READ_TIMEOUT = float(os.getenv('HIL_SERIAL_READ_TIMEOUT', '5'))
|
|
SERIAL_WRITE_TIMEOUT = float(os.getenv('HIL_SERIAL_WRITE_TIMEOUT', '10'))
|
|
|
|
|
|
def cmd_stdout_text(out: Any) -> str:
|
|
if out is None:
|
|
return ''
|
|
if isinstance(out, bytes):
|
|
return out.decode('utf-8', errors='ignore')
|
|
return str(out)
|
|
|
|
WCH_RISCV_CONTENT = """
|
|
adapter driver wlinke
|
|
adapter speed 6000
|
|
transport select sdi
|
|
|
|
wlink_set_address 0x00000000
|
|
set _CHIPNAME wch_riscv
|
|
sdi newtap $_CHIPNAME cpu -irlen 5 -expected-id 0x00001
|
|
|
|
set _TARGETNAME $_CHIPNAME.cpu
|
|
|
|
target create $_TARGETNAME.0 wch_riscv -chain-position $_TARGETNAME
|
|
$_TARGETNAME.0 configure -work-area-phys 0x20000000 -work-area-size 10000 -work-area-backup 1
|
|
set _FLASHNAME $_CHIPNAME.flash
|
|
|
|
flash bank $_FLASHNAME wch_riscv 0x00000000 0 0 0 $_TARGETNAME.0
|
|
|
|
echo "Ready for Remote Connections"
|
|
"""
|
|
|
|
MSC_README_TXT = \
|
|
b"This is tinyusb's MassStorage Class demo.\r\n\r\n\
|
|
If you find any bugs or get any questions, feel free to file an\r\n\
|
|
issue at github.com/hathach/tinyusb"
|
|
|
|
# -------------------------------------------------------------
|
|
# Path
|
|
# -------------------------------------------------------------
|
|
OPENCOD_ADI_PATH = Path.home() / 'app' / 'openocd_adi'
|
|
TINYUSB_ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
# get usb serial by id
|
|
def get_serial_dev(id, vendor_str, product_str, ifnum):
|
|
if vendor_str and product_str:
|
|
# known vendor and product
|
|
vendor_str = vendor_str.replace(' ', '_')
|
|
product_str = product_str.replace(' ', '_')
|
|
return f'/dev/serial/by-id/usb-{vendor_str}_{product_str}_{id}-if{ifnum:02d}'
|
|
else:
|
|
# just use id: mostly for cp210x/ftdi flasher
|
|
pattern = f'/dev/serial/by-id/usb-*_{id}-if*'
|
|
port_list = glob.glob(pattern)
|
|
if len(port_list) == 0:
|
|
raise RuntimeError(f'No serial device found for {pattern}')
|
|
return port_list[0]
|
|
|
|
|
|
# get usb disk by id
|
|
def get_disk_dev(id, vendor_str, lun):
|
|
return f'/dev/disk/by-id/usb-{vendor_str}_Mass_Storage_{id}-0:{lun}'
|
|
|
|
|
|
def get_hid_dev(id, vendor_str, product_str, event):
|
|
return f'/dev/input/by-id/usb-{vendor_str}_{product_str}_{id}-{event}'
|
|
|
|
|
|
def get_alsa_capture_dev(id):
|
|
pattern = f'/dev/snd/by-id/usb-*_{id}-*'
|
|
for dev in glob.glob(pattern):
|
|
try:
|
|
link = os.path.basename(os.path.realpath(dev))
|
|
except OSError:
|
|
continue
|
|
m = re.match(r'controlC(\d+)', link)
|
|
if m:
|
|
return f'hw:{m.group(1)},0'
|
|
return None
|
|
|
|
|
|
def open_serial_dev(port: str):
|
|
timeout = ENUM_TIMEOUT
|
|
ser = None
|
|
while timeout > 0:
|
|
if os.path.exists(port):
|
|
try:
|
|
# write_timeout: a wedged device otherwise blocks ser.write() forever,
|
|
# hanging the worker until the pool/job timeout kills the whole run
|
|
ser = serial.Serial(port, baudrate=115200, timeout=SERIAL_READ_TIMEOUT,
|
|
write_timeout=SERIAL_WRITE_TIMEOUT)
|
|
break
|
|
except serial.SerialException:
|
|
print(f'serial {port} not reaady {timeout} sec')
|
|
pass
|
|
time.sleep(0.1)
|
|
timeout -= 0.1
|
|
|
|
assert timeout > 0, f'Cannot open port f{port}' if os.path.exists(port) else f'Port {port} not existed'
|
|
assert ser is not None
|
|
return ser
|
|
|
|
|
|
def serial_write_all(ser: serial.Serial, data: bytes):
|
|
# write_timeout is a total deadline for the whole call (pyserial keeps partial progress
|
|
# internally). A timeout means the device stopped draining — treat it as fatal: pyserial
|
|
# loses the partial-write count on raise, so retrying would duplicate bytes on the wire.
|
|
try:
|
|
ser.write(data)
|
|
except serial.SerialTimeoutException:
|
|
raise AssertionError(f'Serial write timeout after {SERIAL_WRITE_TIMEOUT:.1f}s')
|
|
|
|
|
|
def read_disk_file(uid: str, lun: int, fname: str) -> bytes:
|
|
# Reads a file from a FAT volume on a block device without mounting it.
|
|
# Requires mtools: `apt install mtools` (no pip dependency).
|
|
dev = get_disk_dev(uid, 'TinyUSB', lun)
|
|
timeout = ENUM_TIMEOUT
|
|
last_err = None
|
|
while timeout > 0:
|
|
if os.path.exists(dev):
|
|
try:
|
|
data = subprocess.check_output(
|
|
['mtype', '-i', dev, f'::/{fname}'], stderr=subprocess.PIPE)
|
|
assert data, f'Cannot read file {fname} from {dev}'
|
|
return data
|
|
except subprocess.CalledProcessError as e:
|
|
last_err = e.stderr.decode(errors='replace').strip()
|
|
time.sleep(1)
|
|
timeout -= 1
|
|
|
|
raise AssertionError(f'mtype failed on {dev}: {last_err}' if last_err else f'Storage {dev} not existed')
|
|
|
|
|
|
def open_mtp_dev(uid):
|
|
mtp = MTP()
|
|
timeout = ENUM_TIMEOUT
|
|
while timeout > 0:
|
|
# unmount gio/gvfs MTP mount which blocks libmtp from accessing the device
|
|
subprocess.run(f"gio mount -u mtp://TinyUsb_TinyUsb_Device_{uid}/",
|
|
shell=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
for raw in mtp.detect_devices():
|
|
mtp.device = mtp.mtp.LIBMTP_Open_Raw_Device(ctypes.byref(raw))
|
|
if mtp.device:
|
|
sn = mtp.get_serialnumber().decode('utf-8')
|
|
if sn == uid:
|
|
return mtp
|
|
mtp.disconnect()
|
|
time.sleep(1)
|
|
timeout -= 1
|
|
return None
|
|
|
|
|
|
def get_printer_dev(id: str, vendor_str, product_str, ifnum: int):
|
|
"""Find /dev/usb/lpX by matching USB serial, vendor, product, and interface number via sysfs"""
|
|
vendor_str = vendor_str.replace(' ', '_') if vendor_str else ''
|
|
product_str = product_str.replace(' ', '_') if product_str else ''
|
|
for lp in glob.glob('/sys/class/usbmisc/lp*'):
|
|
try:
|
|
sn = open(f'{lp}/device/../serial').read().strip()
|
|
if sn == id:
|
|
return f'/dev/usb/{os.path.basename(lp)}'
|
|
except (FileNotFoundError, PermissionError, ValueError):
|
|
pass
|
|
return None
|
|
|
|
|
|
def open_printer_dev(id: str, vendor_str, product_str, ifnum: int) -> str:
|
|
"""Wait for printer device to enumerate and return its path"""
|
|
timeout = ENUM_TIMEOUT
|
|
while timeout > 0:
|
|
lp_dev = get_printer_dev(id, vendor_str, product_str, ifnum)
|
|
if lp_dev and os.path.exists(lp_dev):
|
|
return lp_dev
|
|
time.sleep(1)
|
|
timeout -= 1
|
|
assert False, f'Printer device not found for {id} if{ifnum:02d}'
|
|
|
|
|
|
# -------------------------------------------------------------
|
|
# Flashing firmware
|
|
# -------------------------------------------------------------
|
|
def run_cmd(cmd: str, cwd: str | None = None, timeout: int = CMD_TIMEOUT) -> subprocess.CompletedProcess:
|
|
popen_kwargs = {
|
|
'cwd': cwd,
|
|
'shell': True,
|
|
'stdout': subprocess.PIPE,
|
|
'stderr': subprocess.STDOUT,
|
|
'text': True,
|
|
'encoding': 'utf-8',
|
|
'errors': 'replace',
|
|
}
|
|
if os.name != 'nt':
|
|
popen_kwargs['preexec_fn'] = os.setsid
|
|
|
|
p = subprocess.Popen(cmd, **popen_kwargs)
|
|
try:
|
|
out, _ = p.communicate(timeout=timeout)
|
|
r = subprocess.CompletedProcess(args=cmd, returncode=p.returncode, stdout=out)
|
|
except subprocess.TimeoutExpired as ex:
|
|
if os.name != 'nt':
|
|
try:
|
|
os.killpg(p.pid, signal.SIGKILL)
|
|
except ProcessLookupError:
|
|
pass
|
|
else:
|
|
p.kill()
|
|
out, _ = p.communicate()
|
|
timeout_out = ex.stdout or out or b''
|
|
title = f'COMMAND TIMEOUT ({timeout}s): {cmd}'
|
|
print()
|
|
if os.getenv('CI'):
|
|
print(f"::group::{title}")
|
|
print(cmd_stdout_text(timeout_out))
|
|
print(f"::endgroup::")
|
|
else:
|
|
print(title)
|
|
print(cmd_stdout_text(timeout_out))
|
|
return subprocess.CompletedProcess(args=cmd, returncode=124, stdout=timeout_out)
|
|
|
|
if r.returncode != 0:
|
|
title = f'COMMAND FAILED: {cmd}'
|
|
print()
|
|
if os.getenv('CI'):
|
|
print(f"::group::{title}")
|
|
print(cmd_stdout_text(r.stdout))
|
|
print(f"::endgroup::")
|
|
else:
|
|
print(title)
|
|
print(cmd_stdout_text(r.stdout))
|
|
elif verbose:
|
|
print(cmd)
|
|
print(cmd_stdout_text(r.stdout))
|
|
return r
|
|
|
|
|
|
def flash_jlink(board: Board, firmware: str) -> subprocess.CompletedProcess:
|
|
flasher = board['flasher']
|
|
script = ['halt', 'r', f'loadfile {firmware}.elf', 'r', 'go', 'exit']
|
|
f_jlink = Path(f'{board["name"]}_{Path(firmware).name}.jlink')
|
|
with f_jlink.open('w') as f:
|
|
f.writelines(f'{s}\n' for s in script)
|
|
ret = run_cmd(f'JLinkExe -USB {flasher["uid"]} {flasher["args"]} -if swd -JTAGConf -1,-1 -speed auto -NoGui 1 -ExitOnError 1 -CommandFile {f_jlink}')
|
|
f_jlink.unlink(missing_ok=True)
|
|
return ret
|
|
|
|
|
|
def reset_jlink(board: Board) -> subprocess.CompletedProcess:
|
|
flasher = board['flasher']
|
|
script = ['halt', 'r', 'go', 'exit']
|
|
f_jlink = Path(f'{board["name"]}_reset.jlink')
|
|
if not f_jlink.exists():
|
|
with f_jlink.open('w') as f:
|
|
f.writelines(f'{s}\n' for s in script)
|
|
ret = run_cmd(f'JLinkExe -USB {flasher["uid"]} {flasher["args"]} -if swd -JTAGConf -1,-1 -speed auto -NoGui 1 -ExitOnError 1 -CommandFile {f_jlink}')
|
|
return ret
|
|
|
|
|
|
def flash_stlink(board, firmware):
|
|
flasher = board['flasher']
|
|
return run_cmd(f'STM32_Programmer_CLI --connect port=swd sn={flasher["uid"]} --write {firmware}.elf --go')
|
|
|
|
|
|
def reset_stlink(board):
|
|
flasher = board['flasher']
|
|
return run_cmd(f'STM32_Programmer_CLI --connect port=swd sn={flasher["uid"]} --rst --go')
|
|
|
|
def flash_stflash(board, firmware):
|
|
flasher = board['flasher']
|
|
ret = run_cmd(f'st-flash --serial {flasher["uid"]} write {firmware}.bin 0x8000000')
|
|
return ret
|
|
|
|
|
|
def reset_stflash(board):
|
|
flasher = board['flasher']
|
|
return subprocess.CompletedProcess(args=['dummy'], returncode=0)
|
|
|
|
|
|
def flash_openocd(board, firmware):
|
|
flasher = board['flasher']
|
|
ret = run_cmd(f'openocd -c "tcl_port disabled" -c "gdb_port disabled" -c "adapter serial {flasher["uid"]}" '
|
|
f'{flasher["args"]} -c "init; halt; program {firmware}.elf verify; reset; exit"')
|
|
return ret
|
|
|
|
|
|
def reset_openocd(board):
|
|
flasher = board['flasher']
|
|
ret = run_cmd(f'openocd -c "tcl_port disabled" -c "gdb_port disabled" -c "adapter serial {flasher["uid"]}" '
|
|
f'{flasher["args"]} -c "init; reset run; exit"')
|
|
return ret
|
|
|
|
|
|
def flash_openocd_wch(board, firmware):
|
|
flasher = board['flasher']
|
|
f_wch = f"wch-riscv_{board['uid']}.cfg"
|
|
if not os.path.exists(f_wch):
|
|
with open(f_wch, 'w') as file:
|
|
file.write(WCH_RISCV_CONTENT)
|
|
|
|
ret = run_cmd(f'openocd_wch -c "adapter serial {flasher["uid"]}" -f {f_wch} '
|
|
f'-c "program {firmware}.elf reset exit"')
|
|
return ret
|
|
|
|
|
|
def reset_openocd_wch(board):
|
|
flasher = board['flasher']
|
|
f_wch = f"wch-riscv_{board['uid']}.cfg"
|
|
if not os.path.exists(f_wch):
|
|
with open(f_wch, 'w') as file:
|
|
file.write(WCH_RISCV_CONTENT)
|
|
|
|
ret = run_cmd(f'openocd_wch -c "adapter serial {flasher["uid"]}" -f {f_wch} -c "program reset exit"')
|
|
return ret
|
|
|
|
|
|
def flash_openocd_adi(board: Board, firmware: str) -> subprocess.CompletedProcess:
|
|
flasher = board['flasher']
|
|
openocd = OPENCOD_ADI_PATH / 'src' / 'openocd'
|
|
tcl_dir = OPENCOD_ADI_PATH / 'tcl'
|
|
ret = run_cmd(f'{openocd} -c "adapter serial {flasher["uid"]}" -s {tcl_dir} '
|
|
f'{flasher["args"]} -c "program {firmware}.elf reset exit"')
|
|
return ret
|
|
|
|
|
|
def reset_openocd_adi(board: Board) -> subprocess.CompletedProcess:
|
|
flasher = board['flasher']
|
|
openocd = OPENCOD_ADI_PATH / 'src' / 'openocd'
|
|
tcl_dir = OPENCOD_ADI_PATH / 'tcl'
|
|
ret = run_cmd(f'{openocd} -c "adapter serial {flasher["uid"]}" -s {tcl_dir} '
|
|
f'{flasher["args"]} -c "program reset exit"')
|
|
return ret
|
|
|
|
|
|
def flash_wlink_rs(board, firmware):
|
|
flasher = board['flasher']
|
|
# wlink use index for probe selection and lacking usb serial support
|
|
ret = run_cmd(f'wlink flash {firmware}.elf')
|
|
return ret
|
|
|
|
|
|
def reset_wlink_rs(board):
|
|
flasher = board['flasher']
|
|
# wlink use index for probe selection and lacking usb serial support
|
|
ret = run_cmd(f'wlink reset')
|
|
return ret
|
|
|
|
|
|
def flash_esptool(board: Board, firmware: str) -> subprocess.CompletedProcess:
|
|
flasher = board['flasher']
|
|
port = get_serial_dev(flasher["uid"], None, None, 0)
|
|
fw_dir = Path(f'{firmware}.bin').parent
|
|
with (fw_dir / 'config.env').open() as f:
|
|
idf_target = json.load(f)['IDF_TARGET']
|
|
with (fw_dir / 'flash_args').open() as f:
|
|
flash_args = f.read().strip().replace('\n', ' ')
|
|
command = (f'esptool --chip {idf_target} -p {port} {flasher["args"]} '
|
|
f'--before=default_reset --after=hard_reset write_flash {flash_args}')
|
|
ret = run_cmd(command, cwd=str(fw_dir))
|
|
return ret
|
|
|
|
|
|
def reset_esptool(board):
|
|
flasher = board['flasher']
|
|
return subprocess.CompletedProcess(args=['dummy'], returncode=0)
|
|
|
|
|
|
def flash_uniflash(board, firmware):
|
|
flasher = board['flasher']
|
|
ret = run_cmd(f'dslite.sh {flasher["args"]} -f {firmware}.hex')
|
|
return ret
|
|
|
|
|
|
def reset_uniflash(board):
|
|
flasher = board['flasher']
|
|
return subprocess.CompletedProcess(args=['dummy'], returncode=0)
|
|
|
|
|
|
def flash_lm4flash(board, firmware):
|
|
# TI Tiva-C / Stellaris ICDI: lightweight lm4flash, resets and runs after write
|
|
flasher = board['flasher']
|
|
ret = run_cmd(f'lm4flash -s {flasher["uid"]} {flasher["args"]} {firmware}.bin')
|
|
return ret
|
|
|
|
|
|
def reset_lm4flash(board):
|
|
# lm4flash has no reset-only mode; it resets+runs on flash, so reset is a no-op
|
|
flasher = board['flasher']
|
|
return subprocess.CompletedProcess(args=['dummy'], returncode=0)
|
|
|
|
|
|
# -------------------------------------------------------------
|
|
# Tests: dual
|
|
# -------------------------------------------------------------
|
|
def test_dual_host_info_to_device_cdc(board):
|
|
uid = board['uid']
|
|
declared_devs = [f'{d["vid_pid"]}_{d["serial"]}' for d in board['tests']['dev_attached']]
|
|
port = get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 0)
|
|
ser = open_serial_dev(port)
|
|
ser.timeout = 0.1
|
|
|
|
# read until all expected devices are enumerated
|
|
data = b''
|
|
timeout = ENUM_TIMEOUT
|
|
while timeout > 0:
|
|
new_data = ser.read(ser.in_waiting or 1)
|
|
if new_data:
|
|
data += new_data
|
|
# check if all devices found
|
|
enum_dev_sn = []
|
|
for l in data.decode('utf-8', errors='ignore').splitlines():
|
|
vid_pid_sn = re.search(r'ID ([0-9a-fA-F]+):([0-9a-fA-F]+) SN (\w+)', l)
|
|
if vid_pid_sn:
|
|
enum_dev_sn.append(f'{vid_pid_sn.group(1)}_{vid_pid_sn.group(2)}_{vid_pid_sn.group(3)}')
|
|
if set(declared_devs).issubset(set(enum_dev_sn)):
|
|
break
|
|
time.sleep(0.1)
|
|
timeout -= 0.1
|
|
ser.close()
|
|
|
|
if len(data) == 0:
|
|
assert False, 'No data from device'
|
|
lines = data.decode('utf-8', errors='ignore').splitlines()
|
|
|
|
enum_dev_sn = []
|
|
for l in lines:
|
|
vid_pid_sn = re.search(r'ID ([0-9a-fA-F]+):([0-9a-fA-F]+) SN (\w+)', l)
|
|
if vid_pid_sn:
|
|
print(f'\r\n {l} ', end='')
|
|
enum_dev_sn.append(f'{vid_pid_sn.group(1)}_{vid_pid_sn.group(2)}_{vid_pid_sn.group(3)}')
|
|
|
|
if set(declared_devs) != set(enum_dev_sn):
|
|
failed_msg = f'Expected {declared_devs}, Enumerated {enum_dev_sn}'
|
|
print('\n'.join(lines))
|
|
assert False, failed_msg
|
|
return 0
|
|
|
|
|
|
# -------------------------------------------------------------
|
|
# Tests: host
|
|
# -------------------------------------------------------------
|
|
def test_host_device_info(board):
|
|
flasher = board['flasher']
|
|
declared_devs = [f'{d["vid_pid"]}_{d["serial"]}' for d in board['tests']['dev_attached']]
|
|
|
|
port = get_serial_dev(flasher["uid"], None, None, 0)
|
|
ser = open_serial_dev(port)
|
|
ser.timeout = 0.1
|
|
|
|
# reset device since we can miss the first line
|
|
ret = globals()[f'reset_{flasher["name"].lower()}'](board)
|
|
assert ret.returncode == 0, 'Failed to reset device'
|
|
|
|
# read until all expected devices are enumerated
|
|
data = b''
|
|
timeout = ENUM_TIMEOUT
|
|
while timeout > 0:
|
|
new_data = ser.read(ser.in_waiting or 1)
|
|
if new_data:
|
|
data += new_data
|
|
# check if all devices found
|
|
enum_dev_sn = []
|
|
for l in data.decode('utf-8', errors='ignore').splitlines():
|
|
vid_pid_sn = re.search(r'ID ([0-9a-fA-F]+):([0-9a-fA-F]+) SN (\w+)', l)
|
|
if vid_pid_sn:
|
|
enum_dev_sn.append(f'{vid_pid_sn.group(1)}_{vid_pid_sn.group(2)}_{vid_pid_sn.group(3)}')
|
|
if set(declared_devs).issubset(set(enum_dev_sn)):
|
|
break
|
|
time.sleep(0.1)
|
|
timeout -= 0.1
|
|
ser.close()
|
|
|
|
if len(data) == 0:
|
|
assert False, 'No data from device'
|
|
lines = data.decode('utf-8', errors='ignore').splitlines()
|
|
|
|
enum_dev_sn = []
|
|
for l in lines:
|
|
vid_pid_sn = re.search(r'ID ([0-9a-fA-F]+):([0-9a-fA-F]+) SN (\w+)', l)
|
|
if vid_pid_sn:
|
|
print(f'\r\n {l} ', end='')
|
|
enum_dev_sn.append(f'{vid_pid_sn.group(1)}_{vid_pid_sn.group(2)}_{vid_pid_sn.group(3)}')
|
|
|
|
if set(declared_devs) != set(enum_dev_sn):
|
|
failed_msg = f'Expected {declared_devs}, Enumerated {enum_dev_sn}'
|
|
print('\n'.join(lines))
|
|
assert False, failed_msg
|
|
return 0
|
|
|
|
|
|
def check_msc_info(lines, msc_devs):
|
|
"""Print MSC info and verify block_count/block_size against config"""
|
|
inquiry = ''
|
|
disk_size = ''
|
|
for l in lines:
|
|
if re.match(r'^[A-Za-z].*\s+(rev\s+|[0-9])', l) and 'Disk Size' not in l:
|
|
inquiry = l.strip()
|
|
if 'Disk Size' in l:
|
|
disk_size = l.strip()
|
|
if inquiry or disk_size:
|
|
print(f'\r\n {inquiry} {disk_size} ', end='')
|
|
# Verify block_count and block_size from "Disk Size: COUNT SIZE-byte blocks: N MB"
|
|
if disk_size and msc_devs:
|
|
m = re.match(r'Disk Size:\s+(\d+)\s+(\d+)-byte blocks', disk_size)
|
|
if m:
|
|
actual_count = int(m.group(1))
|
|
actual_size = int(m.group(2))
|
|
for dev in msc_devs:
|
|
exp_count = dev.get('block_count')
|
|
exp_size = dev.get('block_size')
|
|
if exp_count and actual_count == exp_count:
|
|
assert actual_size == exp_size, (
|
|
f'MSC block_size mismatch: expected {exp_size}, got {actual_size}')
|
|
break
|
|
|
|
|
|
def test_host_cdc_msc_hid(board):
|
|
flasher = board['flasher']
|
|
dev_attached = board['tests'].get('dev_attached', [])
|
|
cdc_devs = [d for d in dev_attached if d.get('is_cdc')]
|
|
msc_devs = [d for d in dev_attached if d.get('is_msc')]
|
|
if not cdc_devs and not msc_devs:
|
|
return 'skipped'
|
|
|
|
port = get_serial_dev(flasher["uid"], None, None, 0)
|
|
ser = open_serial_dev(port)
|
|
ser.timeout = 0.1
|
|
|
|
# reset device to catch mount messages
|
|
ret = globals()[f'reset_{flasher["name"].lower()}'](board)
|
|
assert ret.returncode == 0, 'Failed to reset device'
|
|
|
|
# Wait for all expected mount messages
|
|
data = b''
|
|
timeout = ENUM_TIMEOUT
|
|
wait_cdc = len(cdc_devs) > 0
|
|
wait_msc = len(msc_devs) > 0
|
|
while timeout > 0:
|
|
new_data = ser.read(ser.in_waiting or 1)
|
|
if new_data:
|
|
data += new_data
|
|
cdc_ok = (not wait_cdc) or (b'CDC Interface is mounted' in data)
|
|
msc_ok = (not wait_msc) or (b'Disk Size' in data)
|
|
if cdc_ok and msc_ok:
|
|
break
|
|
time.sleep(0.1)
|
|
timeout -= 0.1
|
|
|
|
# Lookup serial chip name from vid_pid
|
|
vid_pid_name = {
|
|
'0403_6001': 'FTDI', '0403_6010': 'FTDI', '0403_6011': 'FTDI', '0403_6014': 'FTDI',
|
|
'10c4_ea60': 'CP210x', '10c4_ea70': 'CP210x',
|
|
'067b_2303': 'PL2303', '067b_23a3': 'PL2303',
|
|
'1a86_7523': 'CH340', '1a86_7522': 'CH340',
|
|
'1a86_55d3': 'CH9102', '1a86_55d4': 'CH9102',
|
|
}
|
|
|
|
lines = data.decode('utf-8', errors='ignore').splitlines()
|
|
|
|
# Verify and print CDC mount
|
|
if cdc_devs:
|
|
assert b'CDC Interface is mounted' in data, 'CDC device not mounted on host'
|
|
dev = cdc_devs[0]
|
|
chip_name = vid_pid_name.get(dev['vid_pid'], dev['vid_pid'])
|
|
for l in lines:
|
|
if 'CDC Interface is mounted' in l:
|
|
print(f'\r\n {chip_name}: {l} ', end='')
|
|
|
|
# Verify and print MSC mount (inquiry + disk size)
|
|
if msc_devs:
|
|
assert b'MassStorage device is mounted' in data, 'MSC device not mounted on host'
|
|
assert b'Disk Size' in data, 'MSC Disk Size not reported'
|
|
check_msc_info(lines, msc_devs)
|
|
|
|
# CDC echo test via flasher serial
|
|
if not cdc_devs:
|
|
ser.close()
|
|
return
|
|
|
|
time.sleep(2)
|
|
ser.read(ser.in_waiting)
|
|
ser.reset_input_buffer()
|
|
|
|
def rand_ascii(length):
|
|
return "".join(random.choices(string.ascii_letters + string.digits, k=length)).encode("ascii")
|
|
|
|
packet_size = 64
|
|
|
|
# Echo test: write random 1-packet_size chunks, wait for echo before sending next
|
|
echo_len = 1024
|
|
echo_data = rand_ascii(echo_len)
|
|
ser.reset_input_buffer()
|
|
offset = 0
|
|
while offset < echo_len:
|
|
chunk_size = min(random.randint(1, packet_size), echo_len - offset)
|
|
serial_write_all(ser, echo_data[offset:offset + chunk_size])
|
|
# wait until this chunk is echoed back
|
|
echo = b''
|
|
t_end = time.monotonic() + 1.0
|
|
while time.monotonic() < t_end and len(echo) < chunk_size:
|
|
rd = ser.read(chunk_size - len(echo))
|
|
if rd:
|
|
echo += rd
|
|
expected = echo_data[offset:offset + chunk_size]
|
|
assert echo == expected, (f'CDC echo mismatch at offset {offset} ({chunk_size} bytes):\n'
|
|
f' expected: {expected}\n received: {echo}')
|
|
offset += chunk_size
|
|
|
|
ser.close()
|
|
|
|
|
|
def test_host_msc_file_explorer(board):
|
|
flasher = board['flasher']
|
|
msc_devs = [d for d in board['tests'].get('dev_attached', []) if d.get('is_msc')]
|
|
if not msc_devs:
|
|
return 'skipped'
|
|
|
|
port = get_serial_dev(flasher["uid"], None, None, 0)
|
|
ser = open_serial_dev(port)
|
|
ser.timeout = 0.1
|
|
|
|
# reset device to catch mount messages
|
|
ret = globals()[f'reset_{flasher["name"].lower()}'](board)
|
|
assert ret.returncode == 0, 'Failed to reset device'
|
|
|
|
# Wait for MSC mount (Disk Size message)
|
|
data = b''
|
|
timeout = ENUM_TIMEOUT
|
|
while timeout > 0:
|
|
new_data = ser.read(ser.in_waiting or 1)
|
|
if new_data:
|
|
data += new_data
|
|
if b'Disk Size' in data:
|
|
break
|
|
time.sleep(0.1)
|
|
timeout -= 0.1
|
|
assert b'Disk Size' in data, 'MSC device not mounted'
|
|
lines = data.decode('utf-8', errors='ignore').splitlines()
|
|
check_msc_info(lines, msc_devs)
|
|
|
|
# Send "cat README.TXT" and check response (optional — file may not exist on all drives)
|
|
time.sleep(1)
|
|
ser.reset_input_buffer()
|
|
for ch in 'cat README.TXT\r':
|
|
serial_write_all(ser, ch.encode())
|
|
time.sleep(0.002)
|
|
|
|
resp = b''
|
|
t = 10.0
|
|
while t > 0:
|
|
rd = ser.read(max(1, ser.in_waiting))
|
|
if rd:
|
|
resp += rd
|
|
if b'>' in resp and resp.rstrip().endswith(b'>'):
|
|
break
|
|
time.sleep(0.05)
|
|
t -= 0.05
|
|
|
|
resp_text = resp.decode('utf-8', errors='ignore')
|
|
if MSC_README_TXT.decode() in resp_text:
|
|
print('README.TXT matched ', end='')
|
|
|
|
# MSC throughput test: send dd command to read sectors
|
|
time.sleep(0.5)
|
|
ser.reset_input_buffer()
|
|
for ch in 'dd 1024\r':
|
|
serial_write_all(ser, ch.encode())
|
|
time.sleep(0.002)
|
|
|
|
# Read dd output until prompt
|
|
resp = b''
|
|
t = 30.0
|
|
while t > 0:
|
|
rd = ser.read(max(1, ser.in_waiting))
|
|
if rd:
|
|
resp += rd
|
|
if b'KB/s' in resp and b'>' in resp:
|
|
break
|
|
time.sleep(0.05)
|
|
t -= 0.05
|
|
|
|
resp_text = resp.decode('utf-8', errors='ignore')
|
|
speed = None
|
|
for line in resp_text.splitlines():
|
|
if 'KB/s' in line:
|
|
print(f'{line.strip()} ', end='')
|
|
m = re.search(r'([\d.]+\s*[KMG]B/s)', line) # MSC read speed for the report cell
|
|
if m:
|
|
speed = 'rd ' + m.group(1).replace(' ', '')
|
|
break
|
|
|
|
ser.close()
|
|
return speed
|
|
|
|
|
|
def test_host_msc_file_explorer_freertos(board):
|
|
return test_host_msc_file_explorer(board)
|
|
|
|
|
|
# -------------------------------------------------------------
|
|
# Tests: device
|
|
# -------------------------------------------------------------
|
|
def test_device_board_test(board):
|
|
# Dummy test
|
|
pass
|
|
|
|
|
|
def test_device_cdc_dual_ports(board):
|
|
uid = board['uid']
|
|
port = [
|
|
get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 0),
|
|
get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 2)
|
|
]
|
|
ser = [open_serial_dev(p) for p in port]
|
|
|
|
def rand_ascii(length):
|
|
return "".join(random.choices(string.ascii_letters + string.digits, k=length)).encode("ascii")
|
|
|
|
sizes = [32, 64, 128, 256, 512, random.randint(2000, 5000)]
|
|
|
|
def write_and_check(writer, payload : bytes):
|
|
payload_len = len(payload)
|
|
for s in ser:
|
|
s.reset_input_buffer()
|
|
rd0 = b''
|
|
rd1 = b''
|
|
offset = 0
|
|
# Write in chunks of random 1-64 bytes (device has 64-byte buffer)
|
|
while offset < payload_len:
|
|
chunk_size = min(random.randint(1, 64), payload_len - offset)
|
|
serial_write_all(ser[writer], payload[offset:offset + chunk_size])
|
|
rd0 += ser[0].read(chunk_size)
|
|
rd1 += ser[1].read(chunk_size)
|
|
offset += chunk_size
|
|
assert rd0 == payload.lower(), f'Port0 wrong data ({payload_len}): expected {payload.lower()}... was {rd0}'
|
|
assert rd1 == payload.upper(), f'Port1 wrong data ({payload_len}): expected {payload.upper()}... was {rd1}'
|
|
|
|
for size in sizes:
|
|
payload0 = rand_ascii(size)
|
|
write_and_check(0, payload0)
|
|
|
|
payload1 = rand_ascii(size)
|
|
write_and_check(1, payload1)
|
|
ser[0].close()
|
|
ser[1].close()
|
|
|
|
|
|
def test_device_cdc_msc(board):
|
|
uid = board['uid']
|
|
# CDC Echo test
|
|
port = get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 0)
|
|
ser = open_serial_dev(port)
|
|
|
|
def rand_ascii(length):
|
|
return "".join(random.choices(string.ascii_letters + string.digits, k=length)).encode("ascii")
|
|
|
|
sizes = [32, 64, 128, 256, 512, random.randint(2000, 5000)]
|
|
for size in sizes:
|
|
test_str = rand_ascii(size)
|
|
rd_str = b''
|
|
offset = 0
|
|
# Write in chunks of random 1-64 bytes (device has 64-byte buffer)
|
|
while offset < size:
|
|
chunk_size = min(random.randint(1, 64), size - offset)
|
|
serial_write_all(ser, test_str[offset:offset + chunk_size])
|
|
rd_str += ser.read(chunk_size)
|
|
offset += chunk_size
|
|
assert rd_str == test_str, f'CDC wrong data ({size} bytes):\n expected: {test_str}\n received: {rd_str}'
|
|
ser.close()
|
|
|
|
# MSC Block test
|
|
data = read_disk_file(uid, 0, 'README.TXT')
|
|
assert data == MSC_README_TXT, f'MSC wrong data in README.TXT\n expected: {MSC_README_TXT.decode()}\n received: {data.decode()}'
|
|
|
|
|
|
def test_device_cdc_msc_freertos(board):
|
|
test_device_cdc_msc(board)
|
|
|
|
|
|
def test_device_cdc_msc_throughput(board):
|
|
uid = board['uid']
|
|
|
|
def parse_speed(dd_output):
|
|
for line in dd_output.splitlines():
|
|
m = re.search(r'([\d.]+)\s+([kMG]?B)/s', line)
|
|
if m:
|
|
return f'{float(m.group(1)):.1f} {m.group(2)}ps'
|
|
return '?'
|
|
|
|
# Wait for MSC disk enumeration
|
|
dev = get_disk_dev(uid, 'TinyUSB', 0)
|
|
timeout = ENUM_TIMEOUT
|
|
while timeout > 0:
|
|
if os.path.exists(dev):
|
|
break
|
|
time.sleep(0.1); timeout -= 0.1
|
|
assert timeout > 0, f'Disk {dev} not found'
|
|
|
|
# Wait for CDC tty enumeration
|
|
tty = get_serial_dev(uid, 'TinyUSB', 'Throughput', 0)
|
|
timeout = ENUM_TIMEOUT
|
|
while timeout > 0:
|
|
if os.path.exists(tty):
|
|
break
|
|
time.sleep(0.1); timeout -= 0.1
|
|
assert timeout > 0, f'CDC tty {tty} not found'
|
|
|
|
# Detect speed (12 Mbps FS / 480 Mbps HS) for payload scaling
|
|
is_fs = False
|
|
for f in glob.glob('/sys/bus/usb/devices/*/serial'):
|
|
try:
|
|
if open(f).read().strip() == uid:
|
|
is_fs = (open(os.path.join(os.path.dirname(f), 'speed')).read().strip() == '12')
|
|
break
|
|
except (OSError, ValueError):
|
|
pass
|
|
|
|
# Put tty in raw mode so dd sees pure binary throughput.
|
|
rs = run_cmd(f'timeout 30 stty -F {tty} raw -echo')
|
|
assert rs.returncode == 0, f'stty failed: {cmd_stdout_text(rs.stdout)}'
|
|
|
|
# Payload aim: ~5 s per direction at FS (~830 kB/s), much less at HS.
|
|
msc_count = 2 if is_fs else 16 # bs=1M
|
|
cdc_count = 16 if is_fs else 128 # bs=64K
|
|
|
|
tmp_file = f'/tmp/cdc_msc_tp_{uid}.bin'
|
|
|
|
rw = run_cmd(f'timeout 30 dd if=/dev/zero of={tty} bs=64K count={cdc_count} 2>&1')
|
|
assert rw.returncode == 0, f'CDC dd write failed: {cmd_stdout_text(rw.stdout)}'
|
|
cdc_w = parse_speed(cmd_stdout_text(rw.stdout))
|
|
|
|
rr = run_cmd(f'timeout 30 dd if={tty} of=/dev/null bs=64K count={cdc_count} iflag=fullblock 2>&1')
|
|
assert rr.returncode == 0, f'CDC dd read failed: {cmd_stdout_text(rr.stdout)}'
|
|
cdc_r = parse_speed(cmd_stdout_text(rr.stdout))
|
|
|
|
rmr = run_cmd(f'dd if={dev} of={tmp_file} bs=1M count={msc_count} iflag=direct 2>&1')
|
|
assert rmr.returncode == 0, f'MSC dd read failed: {cmd_stdout_text(rmr.stdout)}'
|
|
msc_r = parse_speed(cmd_stdout_text(rmr.stdout))
|
|
|
|
rmw = run_cmd(f'dd if={tmp_file} of={dev} bs=1M count={msc_count} oflag=direct 2>&1')
|
|
assert rmw.returncode == 0, f'MSC dd write failed: {cmd_stdout_text(rmw.stdout)}'
|
|
msc_w = parse_speed(cmd_stdout_text(rmw.stdout))
|
|
|
|
try:
|
|
os.remove(tmp_file)
|
|
except OSError:
|
|
pass
|
|
|
|
print(f' CDC read {cdc_r} write {cdc_w}, MSC read {msc_r} write {msc_w} ', end='')
|
|
# compact read/write speed for the report cell, e.g. "✅ CDC 652k/422k MSC 1.1M/783k"
|
|
short = lambda s: (s.split()[0].rstrip('0').rstrip('.') + s.split()[-1][0]) if ' ' in s else s
|
|
return f'{REPORT_CELL["pass"]} CDC {short(cdc_r)}/{short(cdc_w)} MSC {short(msc_r)}/{short(msc_w)}'
|
|
|
|
|
|
def test_device_dfu(board):
|
|
uid = board['uid']
|
|
|
|
# Wait device enum
|
|
timeout = ENUM_TIMEOUT
|
|
while timeout > 0:
|
|
ret = run_cmd(f'dfu-util -l')
|
|
stdout = cmd_stdout_text(ret.stdout)
|
|
if f'serial="{uid}"' in stdout and 'Found DFU: [cafe:4000]' in stdout:
|
|
break
|
|
time.sleep(1)
|
|
timeout = timeout - 1
|
|
|
|
assert timeout > 0, 'Device not available'
|
|
|
|
f_dfu0 = f'dfu0_{uid}'
|
|
f_dfu1 = f'dfu1_{uid}'
|
|
|
|
# Test upload
|
|
try:
|
|
os.remove(f_dfu0)
|
|
os.remove(f_dfu1)
|
|
except OSError:
|
|
pass
|
|
|
|
ret = run_cmd(f'dfu-util -S {uid} -a 0 -U {f_dfu0}')
|
|
assert ret.returncode == 0, 'Upload failed'
|
|
|
|
ret = run_cmd(f'dfu-util -S {uid} -a 1 -U {f_dfu1}')
|
|
assert ret.returncode == 0, 'Upload failed'
|
|
|
|
with open(f_dfu0) as f:
|
|
assert 'Hello world from TinyUSB DFU! - Partition 0' in f.read(), 'Wrong uploaded data'
|
|
|
|
with open(f_dfu1) as f:
|
|
assert 'Hello world from TinyUSB DFU! - Partition 1' in f.read(), 'Wrong uploaded data'
|
|
|
|
os.remove(f_dfu0)
|
|
os.remove(f_dfu1)
|
|
|
|
|
|
def test_device_dfu_runtime(board):
|
|
uid = board['uid']
|
|
# Wait device enum
|
|
timeout = ENUM_TIMEOUT
|
|
while timeout > 0:
|
|
ret = run_cmd(f'dfu-util -l')
|
|
stdout = cmd_stdout_text(ret.stdout)
|
|
if f'serial="{uid}"' in stdout and 'Found Runtime: [cafe:4000]' in stdout:
|
|
break
|
|
time.sleep(1)
|
|
timeout = timeout - 1
|
|
|
|
assert timeout > 0, 'Device not available'
|
|
|
|
|
|
def test_device_hid_boot_interface(board):
|
|
uid = board['uid']
|
|
kbd = get_hid_dev(uid, 'TinyUSB', 'TinyUSB_Device', 'event-kbd')
|
|
mouse1 = get_hid_dev(uid, 'TinyUSB', 'TinyUSB_Device', 'if01-event-mouse')
|
|
mouse2 = get_hid_dev(uid, 'TinyUSB', 'TinyUSB_Device', 'if01-mouse')
|
|
# Wait device enum
|
|
timeout = ENUM_TIMEOUT
|
|
while timeout > 0:
|
|
if os.path.exists(kbd) and os.path.exists(mouse1) and os.path.exists(mouse2):
|
|
break
|
|
time.sleep(1)
|
|
timeout = timeout - 1
|
|
|
|
assert timeout > 0, 'HID device not available'
|
|
|
|
|
|
def test_device_hid_composite_freertos(id):
|
|
# TODO implement later
|
|
pass
|
|
|
|
|
|
def test_device_printer_to_cdc(board):
|
|
import threading
|
|
|
|
uid = board['uid']
|
|
|
|
# Wait for CDC port and printer device
|
|
cdc_port = get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 0)
|
|
ser = open_serial_dev(cdc_port)
|
|
lp_dev = open_printer_dev(uid, 'TinyUSB', 'TinyUSB_Device', 2)
|
|
|
|
# Test 0: Verify IEEE 1284 Device ID from sysfs
|
|
expected_id = 'MFG:TinyUSB;MDL:Printer to CDC;CMD:PS;CLS:PRINTER;'
|
|
lp_name = os.path.basename(lp_dev)
|
|
sysfs_id_path = f'/sys/class/usbmisc/{lp_name}/device/ieee1284_id'
|
|
if os.path.exists(sysfs_id_path):
|
|
with open(sysfs_id_path) as f:
|
|
ieee1284_id = f.read().strip()
|
|
if ieee1284_id:
|
|
assert ieee1284_id == expected_id, (f'IEEE 1284 ID mismatch:\n'
|
|
f' expected: {expected_id}\n got: {ieee1284_id}')
|
|
|
|
def rand_ascii(length):
|
|
return "".join(random.choices(string.ascii_letters + string.digits, k=length)).encode("ascii")
|
|
|
|
sizes = [32, 64, 128, 256, 512, random.randint(2000, 5000)]
|
|
|
|
# flush any stale data
|
|
ser.reset_input_buffer()
|
|
|
|
# Test 1: Printer -> CDC with multiple sizes, write in random 1-64 byte chunks
|
|
LP_WRITE_TIMEOUT = 5.0 # seconds; firmware may stall draining the printer OUT endpoint
|
|
for size in sizes:
|
|
test_data = rand_ascii(size)
|
|
ser.reset_input_buffer()
|
|
rd = b''
|
|
offset = 0
|
|
lp_fd = os.open(lp_dev, os.O_WRONLY | os.O_NONBLOCK)
|
|
try:
|
|
while offset < size:
|
|
chunk_size = min(random.randint(1, 64), size - offset)
|
|
buf = test_data[offset:offset + chunk_size]
|
|
written = 0
|
|
while written < len(buf):
|
|
_, wr, _ = select.select([], [lp_fd], [], LP_WRITE_TIMEOUT)
|
|
assert wr, f'Printer write timeout after {LP_WRITE_TIMEOUT}s (firmware not draining OUT endpoint)'
|
|
n = os.write(lp_fd, buf[written:])
|
|
written += n
|
|
rd += ser.read(chunk_size)
|
|
offset += chunk_size
|
|
finally:
|
|
os.close(lp_fd)
|
|
# read any remaining bytes (fullspeed devices may need extra time)
|
|
while len(rd) < size:
|
|
remaining = ser.read(size - len(rd))
|
|
if not remaining:
|
|
break
|
|
rd += remaining
|
|
assert rd == test_data, (f'Printer->CDC wrong data ({size} bytes):\n'
|
|
f' expected: {test_data[:64]}\n received: {rd[:64]}')
|
|
|
|
# Test 2: CDC -> Printer with multiple sizes, write in random 1-64 byte chunks
|
|
# Use a thread to read from printer since /dev/usb/lp read blocks
|
|
ser.reset_input_buffer()
|
|
time.sleep(0.5)
|
|
for size in sizes:
|
|
test_data = rand_ascii(size)
|
|
rd_result = [b'', None] # [data, error]
|
|
reader_ready = threading.Event()
|
|
|
|
def lp_reader():
|
|
try:
|
|
rd = b''
|
|
fd = os.open(lp_dev, os.O_RDONLY)
|
|
reader_ready.set()
|
|
try:
|
|
while len(rd) < size:
|
|
chunk = os.read(fd, min(64, size - len(rd)))
|
|
if not chunk:
|
|
break
|
|
rd += chunk
|
|
finally:
|
|
os.close(fd)
|
|
rd_result[0] = rd
|
|
except Exception as e:
|
|
rd_result[1] = e
|
|
reader_ready.set()
|
|
|
|
reader = threading.Thread(target=lp_reader, daemon=True)
|
|
reader.start()
|
|
# wait for reader to open lp device before writing
|
|
reader_ready.wait(timeout=5)
|
|
time.sleep(0.1)
|
|
|
|
# Write to CDC in small chunks with flush to avoid overflowing device FIFO
|
|
offset = 0
|
|
while offset < size:
|
|
chunk_size = min(random.randint(1, 64), size - offset)
|
|
serial_write_all(ser, test_data[offset:offset + chunk_size])
|
|
time.sleep(0.01)
|
|
offset += chunk_size
|
|
|
|
reader.join(timeout=10)
|
|
assert not reader.is_alive(), f'CDC->Printer timeout ({size} bytes)'
|
|
assert rd_result[1] is None, f'CDC->Printer read error: {rd_result[1]}'
|
|
assert rd_result[0] == test_data, (f'CDC->Printer wrong data ({size} bytes):\n'
|
|
f' expected: {test_data[:64]}\n received: {rd_result[0][:64]}')
|
|
time.sleep(0.2)
|
|
|
|
ser.close()
|
|
|
|
|
|
def test_device_mtp(board):
|
|
uid = board['uid']
|
|
|
|
# --- BEFORE: mute C-level stderr for libmtp vid/pid warnings ---
|
|
fd = sys.stderr.fileno()
|
|
_saved = os.dup(fd)
|
|
_null = os.open(os.devnull, os.O_WRONLY)
|
|
os.dup2(_null, fd)
|
|
|
|
mtp = open_mtp_dev(uid)
|
|
|
|
# --- AFTER: restore stderr ---
|
|
os.dup2(_saved, fd)
|
|
os.close(_null)
|
|
os.close(_saved)
|
|
|
|
if mtp is None or mtp.device is None:
|
|
assert False, 'MTP device not found'
|
|
|
|
try:
|
|
assert b"TinyUSB" == mtp.get_manufacturer(), 'MTP wrong manufacturer'
|
|
assert b"MTP Example" == mtp.get_modelname(), 'MTP wrong model'
|
|
assert b'1.0' == mtp.get_deviceversion(), 'MTP wrong version'
|
|
assert b'TinyUSB MTP' == mtp.get_devicename(), 'MTP wrong device name'
|
|
|
|
# read and compare readme.txt and logo.png
|
|
f1_expect = b'TinyUSB MTP Filesystem example'
|
|
f2_md5_expect = '40ef23fc2891018d41a05d4a0d5f822f' # md5sum of logo.png
|
|
f1 = uid.encode("utf-8") + b'_file1'
|
|
f2 = uid.encode("utf-8") + b'_file2'
|
|
f3 = uid.encode("utf-8") + b'_file3'
|
|
mtp.get_file_to_file(1, f1)
|
|
with open(f1, 'rb') as file:
|
|
f1_data = file.read()
|
|
os.remove(f1)
|
|
assert f1_data == f1_expect, 'MTP file1 wrong data'
|
|
mtp.get_file_to_file(2, f2)
|
|
with open(f2, 'rb') as file:
|
|
f2_data = file.read()
|
|
os.remove(f2)
|
|
assert f2_md5_expect == hashlib.md5(f2_data).hexdigest(), 'MTP file2 wrong data'
|
|
# test send file
|
|
with open(f3, "wb") as file:
|
|
f3_data = os.urandom(random.randint(1024, 3*1024))
|
|
file.write(f3_data)
|
|
file.close()
|
|
fid = mtp.send_file_from_file(f3, b'file3')
|
|
f3_readback = f3 + b'_readback'
|
|
mtp.get_file_to_file(fid, f3_readback)
|
|
with open(f3_readback, 'rb') as f:
|
|
f3_rb_data = f.read()
|
|
os.remove(f3_readback)
|
|
assert f3_rb_data == f3_data, 'MTP file3 wrong data'
|
|
os.remove(f3)
|
|
mtp.delete_object(fid)
|
|
finally:
|
|
mtp.disconnect()
|
|
|
|
|
|
def test_device_net_lwip_webserver(board):
|
|
# MAC hard-coded in examples/device/net_lwip_webserver/src/main.c; Linux names the
|
|
# USB network interface enx<MAC_lowercase_no_colons>. Device IP is 192.168.7.1 and
|
|
# the example runs an iperf2 TCP server on port 5001 (INCLUDE_IPERF).
|
|
import socket
|
|
mac_no_colons = '0202846a9600'
|
|
iface = 'enx' + mac_no_colons
|
|
device_ip = '192.168.7.1'
|
|
iperf_port = 5001
|
|
|
|
# Wait for the host to get an IPv4 address in the device's subnet (DHCP served by the device).
|
|
# USB enum + DHCP serve can take longer on the CI HIL hardware than on local — give it 30s.
|
|
iface_timeout = 30
|
|
deadline = time.time() + iface_timeout
|
|
host_ip = None
|
|
while time.time() < deadline:
|
|
ret = subprocess.run(['ip', '-o', '-4', 'addr', 'show', iface],
|
|
capture_output=True, text=True, timeout=2)
|
|
m = re.search(r'inet (192\.168\.7\.\d+)/', ret.stdout) if ret.returncode == 0 else None
|
|
if m:
|
|
host_ip = m.group(1)
|
|
break
|
|
time.sleep(0.5)
|
|
assert host_ip, f'USB net iface {iface} did not come up with 192.168.7.x within {iface_timeout}s'
|
|
|
|
# Poll the iperf TCP port until the device is accepting. The net stack comes up a bit
|
|
# after DHCP completes; iperf server binding isn't instantaneous after reflash.
|
|
deadline = time.time() + ENUM_TIMEOUT
|
|
last_err = None
|
|
while time.time() < deadline:
|
|
try:
|
|
with socket.create_connection((device_ip, iperf_port), timeout=1):
|
|
last_err = None
|
|
break
|
|
except OSError as e:
|
|
last_err = e
|
|
time.sleep(0.3)
|
|
assert last_err is None, f'iperf TCP {device_ip}:{iperf_port} not accepting within {ENUM_TIMEOUT}s: {last_err}'
|
|
|
|
# Throughput: 5-second iperf2 TCP test, CSV output for stable parsing.
|
|
# iperf2 CSV final summary line: timestamp,src_ip,src_port,dst_ip,dst_port,id,interval,bytes,bps
|
|
ret = subprocess.run(['iperf', '-c', device_ip, '-t', '5', '-y', 'C'],
|
|
capture_output=True, text=True, timeout=30)
|
|
stderr = ret.stderr.strip()
|
|
stdout = ret.stdout.strip()
|
|
assert ret.returncode == 0, f'iperf rc={ret.returncode}: stderr={stderr!r} stdout={stdout!r}'
|
|
lines = [l for l in stdout.splitlines() if l]
|
|
assert lines, f'iperf produced no output (rc={ret.returncode}, stderr={stderr!r})'
|
|
try:
|
|
bps = int(lines[-1].split(',')[-1])
|
|
except (ValueError, IndexError) as e:
|
|
raise AssertionError(f'could not parse iperf output: {lines[-1]!r} ({e})')
|
|
mbps = bps / 1e6
|
|
print(f' iperf {mbps:5.1f} Mbps', end='')
|
|
|
|
# Reject implausibly low throughput - a working USB-net link should clear this easily.
|
|
assert mbps >= 1.0, f'iperf throughput too low: {mbps:.2f} Mbps'
|
|
|
|
|
|
def test_device_msc_dual_lun(board):
|
|
uid = board['uid']
|
|
|
|
# Read README from LUN 0
|
|
data0 = read_disk_file(uid, 0, 'README0.TXT')
|
|
readme0 = b"LUN0: " + MSC_README_TXT
|
|
assert data0 == readme0, f'MSC LUN0 wrong data in README0.TXT\n expected: {readme0}\n received: {data0}'
|
|
|
|
# Read README from LUN 1
|
|
data1 = read_disk_file(uid, 1, 'README1.TXT')
|
|
readme1 = b"LUN1: " + MSC_README_TXT
|
|
assert data1 == readme1, f'MSC LUN1 wrong data in README1.TXT\n expected: {readme1}\n received: {data1}'
|
|
|
|
|
|
def test_device_midi_test(board):
|
|
uid = board['uid']
|
|
|
|
# Find MIDI device via /dev/snd/by-id using board UID
|
|
timeout = ENUM_TIMEOUT
|
|
midi_port = None
|
|
while timeout > 0:
|
|
pattern = f'/dev/snd/by-id/usb-*_{uid}-*'
|
|
devs = glob.glob(pattern)
|
|
if devs:
|
|
# by-id entry points to controlCX, derive card number for midiCXD0
|
|
link = os.path.basename(os.readlink(devs[0])) # e.g. "controlC2"
|
|
card_num = link.replace('controlC', '')
|
|
midi_path = f'/dev/snd/midiC{card_num}D0'
|
|
if os.path.exists(midi_path):
|
|
midi_port = midi_path
|
|
break
|
|
time.sleep(1)
|
|
timeout -= 1
|
|
assert midi_port is not None, f'MIDI device not found for {uid}'
|
|
|
|
# Read MIDI messages and verify note on/off
|
|
import select
|
|
with open(midi_port, 'rb') as f:
|
|
notes = []
|
|
# Read for up to 3 seconds to capture a few notes (286ms interval)
|
|
end_time = time.time() + 3
|
|
while time.time() < end_time:
|
|
ready, _, _ = select.select([f], [], [], 0.5)
|
|
if ready:
|
|
data = f.read(64)
|
|
if data:
|
|
# Parse MIDI bytes: note_on = 0x90, note_off = 0x80
|
|
i = 0
|
|
while i + 2 < len(data):
|
|
status = data[i]
|
|
if (status & 0xF0) == 0x90: # Note On
|
|
notes.append(data[i + 1])
|
|
i += 3
|
|
elif (status & 0xF0) == 0x80: # Note Off
|
|
i += 3
|
|
else:
|
|
i += 1
|
|
|
|
assert len(notes) >= 2, f'Expected at least 2 MIDI notes, got {len(notes)}'
|
|
# Verify notes are from the expected sequence
|
|
note_sequence = [
|
|
74, 78, 81, 86, 90, 93, 98, 102, 57, 61, 66, 69, 73, 78, 81, 85,
|
|
88, 92, 97, 100, 97, 92, 88, 85, 81, 78, 74, 69, 66, 62, 57, 62,
|
|
66, 69, 74, 78, 81, 86, 90, 93, 97, 102, 97, 93, 90, 85, 81, 78,
|
|
73, 68, 64, 61, 56, 61, 64, 68, 74, 78, 81, 86, 90, 93, 98, 102
|
|
]
|
|
for n in notes:
|
|
assert n in note_sequence, f'Unexpected MIDI note {n}'
|
|
|
|
|
|
def test_device_audio_test_freertos(board):
|
|
uid = board['uid']
|
|
|
|
if os.name == 'nt':
|
|
return 'skipped'
|
|
|
|
pcm = None
|
|
timeout = ENUM_TIMEOUT
|
|
while timeout > 0:
|
|
pcm = get_alsa_capture_dev(uid)
|
|
if pcm:
|
|
break
|
|
time.sleep(1)
|
|
timeout -= 1
|
|
|
|
assert pcm is not None, f'ALSA capture device not found for {uid}'
|
|
|
|
raw_path = f'/tmp/tinyusb_audio_{uid}.raw'
|
|
cmd = [
|
|
'arecord',
|
|
'-D', pcm,
|
|
'-q',
|
|
'-f', 'S16_LE',
|
|
'-c', '1',
|
|
'-r', '48000',
|
|
'-d', '2',
|
|
'-t', 'raw',
|
|
raw_path,
|
|
]
|
|
|
|
ret = subprocess.run(cmd, capture_output=True, text=True, timeout=20)
|
|
assert ret.returncode == 0, f'arecord failed: {ret.stderr.strip() or ret.stdout.strip()}'
|
|
|
|
try:
|
|
with open(raw_path, 'rb') as f:
|
|
raw = f.read()
|
|
finally:
|
|
try:
|
|
os.remove(raw_path)
|
|
except OSError:
|
|
pass
|
|
|
|
assert len(raw) >= 48000, f'Captured too little audio: {len(raw)} bytes'
|
|
assert (len(raw) % 2) == 0, f'Invalid 16-bit audio length: {len(raw)}'
|
|
|
|
sample_count = len(raw) // 2
|
|
samples = [int.from_bytes(raw[i:i + 2], 'little', signed=False) for i in range(0, len(raw), 2)]
|
|
assert sample_count > 1024, f'Not enough samples captured: {sample_count}'
|
|
|
|
# The firmware sends a continuous uint16 ramp. Using ALSA hw: capture bypasses
|
|
# PulseAudio processing, so most adjacent samples should differ by exactly 1.
|
|
total_diffs = sample_count - 1
|
|
one_step = 0
|
|
near_step = 0
|
|
for i in range(total_diffs):
|
|
d = (samples[i + 1] - samples[i]) & 0xFFFF
|
|
if d == 1:
|
|
one_step += 1
|
|
if d in (0, 1, 2, 47, 48, 49):
|
|
near_step += 1
|
|
|
|
one_ratio = one_step / total_diffs
|
|
near_ratio = near_step / total_diffs
|
|
assert one_ratio >= 0.85, f'Unexpected audio pattern (strict ratio={one_ratio:.3f})'
|
|
assert near_ratio >= 0.98, f'Unexpected audio pattern (relaxed ratio={near_ratio:.3f})'
|
|
|
|
print(f' ALSA {pcm} strict={one_ratio:.3f} relaxed={near_ratio:.3f}', end='')
|
|
|
|
|
|
def test_device_hid_generic_inout(board):
|
|
uid = board['uid']
|
|
import hid # cython-hidapi (pip: hidapi, apt: python3-hid)
|
|
|
|
# Find HID device by UID (VID=0xCafe)
|
|
timeout = ENUM_TIMEOUT
|
|
dev = None
|
|
while timeout > 0:
|
|
for d in hid.enumerate(0xCafe):
|
|
if d['serial_number'] == uid:
|
|
dev = d
|
|
break
|
|
if dev:
|
|
break
|
|
time.sleep(1)
|
|
timeout -= 1
|
|
assert dev is not None, f'HID device not found for {uid}'
|
|
|
|
h = hid.device()
|
|
h.open(dev['vendor_id'], dev['product_id'], uid)
|
|
try:
|
|
# Echo test: send random data and verify echo
|
|
for size in [8, 32, 63]:
|
|
# Report ID (0) + payload, padded to 64 bytes
|
|
payload = bytes([random.randint(1, 255) for _ in range(size)])
|
|
report = bytes([0]) + payload + bytes(64 - size)
|
|
h.write(report)
|
|
echo = h.read(64, 2000)
|
|
assert echo and len(echo) >= size, (
|
|
f'HID echo timeout or short read ({size} bytes)')
|
|
assert bytes(echo[:size]) == payload, (
|
|
f'HID echo wrong data ({size} bytes):\n'
|
|
f' expected: {payload.hex()}\n received: {bytes(echo[:size]).hex()}')
|
|
finally:
|
|
h.close()
|
|
|
|
|
|
# -------------------------------------------------------------
|
|
# Main
|
|
# -------------------------------------------------------------
|
|
# device tests
|
|
# note don't test 2 examples with cdc or 2 msc next to each other
|
|
device_tests = [
|
|
# Order matters: cdc_msc and cdc_msc_throughput share the same VID:PID (cafe:4003), so keep a
|
|
# differently-PID'd example (dfu, cafe:4000) between them. Boards whose CPU-reset does not drop
|
|
# D+ (e.g. WCH CH58x via openocd) only re-enumerate when the PID changes; back-to-back same-PID
|
|
# firmware would otherwise leave the host on the previous example's cached descriptors.
|
|
'device/cdc_dual_ports',
|
|
'device/cdc_msc',
|
|
'device/dfu',
|
|
'device/cdc_msc_throughput',
|
|
'device/audio_test_freertos',
|
|
'device/dfu_runtime',
|
|
'device/cdc_msc_freertos',
|
|
'device/hid_boot_interface',
|
|
'device/msc_dual_lun',
|
|
'device/hid_generic_inout',
|
|
'device/printer_to_cdc',
|
|
'device/midi_test',
|
|
'device/mtp',
|
|
# 'device/net_lwip_webserver', # disabled for PR #3605: USB net iface enum is flaky on the CI HIL host
|
|
]
|
|
|
|
dual_tests = [
|
|
'dual/host_info_to_device_cdc',
|
|
]
|
|
|
|
host_test = [
|
|
'host/cdc_msc_hid',
|
|
'host/msc_file_explorer',
|
|
'host/msc_file_explorer_freertos',
|
|
'host/device_info',
|
|
]
|
|
|
|
|
|
def find_firmware(variant: str, example: str):
|
|
"""Locate a built example's firmware base path (no extension) under
|
|
cmake-build-<variant>/<example>/. Accepts the single-config layout (firmware
|
|
directly in the example dir) or Ninja Multi-Config (a per-config subdir like
|
|
RelWithDebInfo/). Returns the base Path, or None if not built."""
|
|
fw_dir = TINYUSB_ROOT / build_dir / f'cmake-build-{variant}' / example
|
|
base = Path(example).name
|
|
if fw_dir.is_dir():
|
|
for cand in [fw_dir / base, fw_dir / 'RelWithDebInfo' / base,
|
|
*(p.with_suffix('') for p in sorted(fw_dir.glob(f'*/{base}.elf')))]:
|
|
if cand.with_suffix('.elf').exists() or cand.with_suffix('.bin').exists():
|
|
return cand
|
|
return None
|
|
|
|
|
|
def test_example(board: Board, variant: str, example: str) -> tuple[int, str]:
|
|
"""
|
|
Test example firmware
|
|
:param board: board dict
|
|
:param variant: build variant name = build dir (cmake-build-<variant>) and report row
|
|
:param example: example name
|
|
:return: (err_count, status, metric) where err_count is 0 on success/skip or
|
|
1 on failure, status is one of 'pass'/'fail'/'skip' (a missing binary
|
|
counts as 'skip'), and metric is an optional string a test returns to
|
|
show in its report cell instead of the pass symbol (e.g. speed)
|
|
"""
|
|
err_count = 0
|
|
result_status = 'fail'
|
|
metric = None
|
|
|
|
test_name = f'{variant:40} {example:30} ...'
|
|
|
|
fw_name = find_firmware(variant, example)
|
|
if fw_name is None:
|
|
log_line(f'{test_name} Skip (no binary)')
|
|
return 0, 'skip', None
|
|
|
|
if verbose:
|
|
log_line(f'Flashing {fw_name}.elf')
|
|
|
|
# flash firmware (unless --skip-flash), then run the test. Both may fail randomly,
|
|
# retry a few times.
|
|
start_s = time.time()
|
|
flash_ok = True
|
|
last_err = ''
|
|
last_detail = ''
|
|
for i in range(max_retry):
|
|
attempt_out = io.StringIO()
|
|
with redirect_stdout(attempt_out):
|
|
if not skip_flash:
|
|
ret = globals()[f'flash_{board["flasher"]["name"].lower()}'](board, str(fw_name))
|
|
flash_ok = (ret.returncode == 0)
|
|
if flash_ok:
|
|
try:
|
|
tret = globals()[f'test_{example.replace("/", "_")}'](board)
|
|
last_detail = compact_output(attempt_out.getvalue())
|
|
if tret == 'skipped':
|
|
status = STATUS_SKIPPED
|
|
result_status = 'skip'
|
|
else:
|
|
status = STATUS_OK
|
|
result_status = 'pass'
|
|
# a test may return a string to show in its report cell (e.g. speed)
|
|
metric = tret if isinstance(tret, str) else None
|
|
msg = f'{test_name} {status}'
|
|
if last_detail:
|
|
msg += f' {last_detail}'
|
|
msg += f' in {time.time() - start_s:.1f}s'
|
|
log_line(msg)
|
|
break
|
|
except Exception as e:
|
|
last_err = str(e)
|
|
last_detail = compact_output(attempt_out.getvalue())
|
|
if i == max_retry - 1:
|
|
err_count += 1
|
|
msg = f'{test_name} {STATUS_FAILED}: {e}'
|
|
if last_detail:
|
|
msg += f' {last_detail}'
|
|
msg += f' in {time.time() - start_s:.1f}s'
|
|
log_line(msg)
|
|
else:
|
|
msg = f'{test_name} retry {i+2}/{max_retry}: test failed: {e}'
|
|
if last_detail:
|
|
msg += f' {last_detail}'
|
|
log_line(msg)
|
|
time.sleep(0.5)
|
|
else:
|
|
last_err = 'Flash failed'
|
|
last_detail = compact_output(attempt_out.getvalue())
|
|
if i < max_retry - 1:
|
|
msg = f'{test_name} retry {i+2}/{max_retry}: flash failed'
|
|
if last_detail:
|
|
msg += f' {last_detail}'
|
|
log_line(msg)
|
|
time.sleep(0.5)
|
|
|
|
if not flash_ok:
|
|
err_count += 1
|
|
msg = f'{test_name} Flash {STATUS_FAILED}'
|
|
if last_err:
|
|
msg += f': {last_err}'
|
|
if last_detail:
|
|
msg += f' {last_detail}'
|
|
msg += f' in {time.time() - start_s:.1f}s'
|
|
log_line(msg)
|
|
|
|
return err_count, result_status, metric
|
|
|
|
|
|
def build_board(board: Board) -> tuple[str, int]:
|
|
"""Build firmware for this board via tools/build.py.
|
|
Honors board config's variant list and build.args defines.
|
|
Output goes to cmake-build/cmake-build-<variant>/ (tools/build.py layout)."""
|
|
name = board['name']
|
|
bcfg = cast(BuildCfg, board.get('build', {}))
|
|
extra_defs = bcfg.get('args', [])
|
|
variants = board.get('variant') or [{'name': name, 'flags': ''}]
|
|
|
|
failed = 0
|
|
for v in variants:
|
|
cmd = [sys.executable, str(TINYUSB_ROOT / 'tools' / 'build.py'), '-b', name]
|
|
for d in extra_defs:
|
|
cmd += ['-D', d]
|
|
if v['name'] != name:
|
|
cmd += ['--build-name', v['name']]
|
|
for d in v.get('defines', []):
|
|
cmd += ['-D', d]
|
|
for tok in v.get('flags', '').split():
|
|
cmd += [f'--cflag={tok}']
|
|
if verbose:
|
|
cmd.append('-v')
|
|
print(f' + {" ".join(cmd)}')
|
|
r = subprocess.run(cmd, cwd=TINYUSB_ROOT)
|
|
if r.returncode != 0:
|
|
failed += 1
|
|
return name, failed
|
|
|
|
|
|
def test_board(board: Board) -> tuple[str, int, list[str], list]:
|
|
name = board['name']
|
|
flasher = board['flasher']
|
|
|
|
# default to all tests
|
|
test_list = []
|
|
|
|
if name in board_test:
|
|
test_list = board_test[name]
|
|
elif len(test_only) > 0:
|
|
# Explicit -t: filter against the board's capabilities so a device-only
|
|
# board doesn't try to run host/dual tests (the test functions need a
|
|
# `dev_attached` entry in the board config that won't exist).
|
|
board_tests = board.get('tests', {})
|
|
if 'only' in board_tests:
|
|
allowed = set(board_tests['only'])
|
|
test_list = [t for t in test_only if t in allowed]
|
|
else:
|
|
for t in test_only:
|
|
category = t.split('/', 1)[0]
|
|
if board_tests.get(category) is True:
|
|
test_list.append(t)
|
|
else:
|
|
if 'tests' in board:
|
|
board_tests = board['tests']
|
|
if board_tests.get('device') is True:
|
|
test_list += list(device_tests)
|
|
if board_tests.get('dual') is True:
|
|
test_list += dual_tests
|
|
if board_tests.get('host') is True:
|
|
test_list += host_test
|
|
if 'only' in board_tests:
|
|
test_list = board_tests['only']
|
|
if 'skip' in board_tests:
|
|
for skip in board_tests['skip']:
|
|
if skip in test_list:
|
|
test_list.remove(skip)
|
|
log_line(f'{name:25} {skip:30} ... Skip')
|
|
|
|
err_count = 0
|
|
failed_tests = []
|
|
rows = [] # list of (row_label, {example: status}) — one row per build variant
|
|
variants = board.get('variant') or [{'name': name, 'flags': ''}]
|
|
|
|
for v in variants:
|
|
vname = v['name']
|
|
cells = {}
|
|
for test in test_list:
|
|
ec, status, metric = test_example(board, vname, test)
|
|
err_count += ec
|
|
cells[test] = metric if metric else status
|
|
if ec > 0:
|
|
failed_tests.append(test)
|
|
rows.append((vname, cells))
|
|
|
|
# flash board_test last to disable board's usb (skipped when --skip-flash is set);
|
|
# this is teardown/park, not a test — not recorded in the report
|
|
if not skip_flash:
|
|
test_example(board, variants[0]['name'], 'device/board_test')
|
|
|
|
return name, err_count, sorted(set(failed_tests)), rows
|
|
|
|
|
|
REPORT_MD = 'hil_report.md'
|
|
REPORT_JSON = 'hil_report.json'
|
|
|
|
|
|
def render_matrix(rows_all: list) -> str:
|
|
"""Render rows (list of (row_label, {example: status})) as an aligned markdown
|
|
matrix: columns = tests (bare names) centered, boards left-aligned."""
|
|
canonical = device_tests + dual_tests + host_test
|
|
seen = set()
|
|
for _, cells in rows_all:
|
|
seen.update(cells)
|
|
if not seen:
|
|
return 'No tests were run.'
|
|
|
|
# columns: canonical order first, then any extras (e.g. from -t) alphabetically
|
|
columns = [t for t in canonical if t in seen]
|
|
columns += [t for t in sorted(seen) if t not in canonical]
|
|
headers = [c.rsplit('/', 1)[-1] for c in columns] # bare example name
|
|
|
|
def cell(cells, col):
|
|
v = cells.get(col)
|
|
if v is None:
|
|
return ''
|
|
return REPORT_CELL.get(v, v) # status symbol, or a metric string (e.g. speed) verbatim
|
|
|
|
board_hdr = 'Board'
|
|
board_w = max([len(board_hdr)] + [len(lbl) for lbl, _ in rows_all])
|
|
col_w = [max([len(h)] + [len(cell(cells, c)) for _, cells in rows_all])
|
|
for h, c in zip(headers, columns)]
|
|
|
|
def line(label, values):
|
|
padded = [label.ljust(board_w)] + [v.center(w) for v, w in zip(values, col_w)]
|
|
return '| ' + ' | '.join(padded) + ' |'
|
|
|
|
header = line(board_hdr, headers)
|
|
sep = '| ' + '-' * board_w + ' | ' + ' | '.join(':' + '-' * (w - 2) + ':' for w in col_w) + ' |'
|
|
body = [line(lbl, [cell(cells, c) for c in columns]) for lbl, cells in rows_all]
|
|
|
|
# tally run cells (blank/not-run cells are absent from the dicts); a metric string counts as pass
|
|
failed = sum(v == 'fail' for _, cells in rows_all for v in cells.values())
|
|
skipped = sum(v == 'skip' for _, cells in rows_all for v in cells.values())
|
|
passed = sum(v not in ('fail', 'skip') for _, cells in rows_all for v in cells.values())
|
|
summary = (f'**{REPORT_CELL["pass"]} {passed} passed · {REPORT_CELL["fail"]} {failed} failed · '
|
|
f'{REPORT_CELL["skip"]} {skipped} skipped · blank not run**')
|
|
|
|
return summary + '\n\n' + '\n'.join([header, sep] + body)
|
|
|
|
|
|
def accumulate_report(mret: list, report_dir: Path, fresh: bool) -> str:
|
|
"""Merge this run's results into hil_report.json in report_dir, then (re)write
|
|
the markdown matrix to hil_report.md. `fresh` (a full run, no --skip-board/-bt)
|
|
starts a new report; otherwise a re-run accumulates so boards/tests that
|
|
already passed are preserved while re-run cells are updated. Returns the md."""
|
|
acc = {} # ordered {row_label: {example: status}}
|
|
jpath = report_dir / REPORT_JSON
|
|
if not fresh and jpath.is_file():
|
|
try:
|
|
for entry in json.loads(jpath.read_text()).get('rows', []):
|
|
acc[entry['board']] = dict(entry['cells'])
|
|
except (ValueError, KeyError, TypeError):
|
|
pass # corrupt/old sidecar: start fresh
|
|
|
|
# merge this run: current cells override prior for boards/tests that ran
|
|
for _, _, _, rows in mret:
|
|
for row_label, cells in rows:
|
|
acc.setdefault(row_label, {}).update(cells)
|
|
|
|
report_dir.mkdir(parents=True, exist_ok=True)
|
|
jpath.write_text(json.dumps({'rows': [{'board': k, 'cells': v} for k, v in acc.items()]},
|
|
indent=2) + '\n')
|
|
|
|
md = render_matrix(list(acc.items()))
|
|
(report_dir / REPORT_MD).write_text(md + '\n', encoding='utf-8')
|
|
return md
|
|
|
|
|
|
def main() -> None:
|
|
"""
|
|
Hardware test on specified boards
|
|
"""
|
|
global verbose
|
|
global test_only
|
|
global board_test
|
|
global build_dir
|
|
global max_retry
|
|
global skip_flash
|
|
|
|
duration = time.time()
|
|
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument('config_file', help='Configuration JSON file')
|
|
parser.add_argument('-b', '--board', action='append', default=[], help='Boards to test, all if not specified')
|
|
parser.add_argument('-s', '--skip-board', action='append', default=[], help='Skip boards from test')
|
|
parser.add_argument('-sf', '--skip-flash', action='store_true', help='Run tests without flashing firmware (use whatever is already on the board)')
|
|
parser.add_argument('-t', '--test-only', action='append', default=[], help='Tests to run, all if not specified')
|
|
parser.add_argument('-bt', '--board-test', action='append', default=[],
|
|
help='Per-board test list as BOARD:test1,test2 (overrides -t for that board); repeat for multiple boards')
|
|
parser.add_argument('-B', '--build-dir', default='cmake-build', help='Build folder name (default: cmake-build)')
|
|
parser.add_argument('--build', action='store_true', help='Build firmware for selected boards with cmake before running tests')
|
|
parser.add_argument('-r', '--retry', type=int, default=3, help='Retry count for failed tests (default: 3)')
|
|
parser.add_argument('-v', '--verbose', action='store_true', help='Verbose output')
|
|
args = parser.parse_args()
|
|
|
|
config_file = Path(args.config_file)
|
|
boards = args.board
|
|
skip_boards = args.skip_board
|
|
verbose = args.verbose
|
|
test_only = args.test_only
|
|
for entry in args.board_test:
|
|
bname, _, tnames = entry.partition(':')
|
|
if not bname or not tnames:
|
|
parser.error(f'invalid --board-test value: {entry!r} (expected BOARD:test1,test2)')
|
|
board_test[bname] = [t for t in tnames.split(',') if t]
|
|
build_dir = args.build_dir
|
|
max_retry = args.retry
|
|
skip_flash = args.skip_flash
|
|
|
|
# if config file is not found, try to find it in the same directory as this script
|
|
if not config_file.exists():
|
|
config_file = Path(__file__).resolve().parent / config_file
|
|
with config_file.open() as f:
|
|
config = cast(HilConfig, json.load(f))
|
|
|
|
if len(boards) == 0:
|
|
config_boards = [e for e in config['boards'] if e['name'] not in skip_boards]
|
|
else:
|
|
config_boards = [e for e in config['boards'] if e['name'] in boards]
|
|
|
|
build_err = 0
|
|
if args.build:
|
|
if build_dir != 'cmake-build':
|
|
print(f'warning: --build writes into cmake-build/, but -B is {build_dir!r}; '
|
|
f'tests will not find the freshly built firmware')
|
|
print('-' * 30)
|
|
print(f'Build phase: {len(config_boards)} board(s)')
|
|
print('-' * 30)
|
|
for board in config_boards:
|
|
_, nfail = build_board(board)
|
|
build_err += nfail
|
|
print('-' * 30)
|
|
print(f'Build phase done: {build_err} failed')
|
|
print('-' * 30)
|
|
|
|
# HIL report sidecar (hil_report.json/.md). A full run starts fresh; a re-run
|
|
# (--skip-board / -bt, i.e. the .skip file) accumulates so already-passed
|
|
# boards/tests are preserved. Clear any prior report up front on a fresh run so
|
|
# a crash mid-run can't leave stale results to be merged by a retry or posted.
|
|
report_dir = Path(os.environ.get('HIL_REPORT_DIR', '.'))
|
|
fresh = not (args.skip_board or args.board_test)
|
|
if fresh:
|
|
report_dir.mkdir(parents=True, exist_ok=True)
|
|
for f in (REPORT_JSON, REPORT_MD):
|
|
(report_dir / f).unlink(missing_ok=True)
|
|
|
|
with Pool(processes=os.cpu_count() or 1, initializer=init_worker, initargs=(Lock(),)) as pool:
|
|
async_ret = pool.map_async(test_board, config_boards)
|
|
try:
|
|
mret = async_ret.get(timeout=POOL_TIMEOUT)
|
|
except MpTimeoutError:
|
|
pool.terminate()
|
|
pool.join()
|
|
raise RuntimeError(f'HIL worker pool timed out after {POOL_TIMEOUT}s')
|
|
|
|
err_count = build_err + sum(e[1] for e in mret)
|
|
# generate skip list for next re-run if failed: skip boards that fully passed,
|
|
# and emit -bt BOARD:t1,t2 so each failed board only re-runs its own failed tests.
|
|
skip_fname = config_file.with_suffix(config_file.suffix + '.skip')
|
|
if err_count > 0:
|
|
skip_boards += [name for name, err, _, _ in mret if err == 0]
|
|
parts = [f'--skip-board {i}' for i in skip_boards]
|
|
parts += [f'-bt {name}:{",".join(fts)}' for name, err, fts, _ in mret if err > 0 and fts]
|
|
with skip_fname.open('w') as f:
|
|
f.write(' '.join(parts))
|
|
elif skip_fname.exists():
|
|
skip_fname.unlink()
|
|
|
|
# board x test result matrix -> hil_report.md (accumulates across re-runs) + stdout
|
|
report = accumulate_report(mret, report_dir, fresh)
|
|
print()
|
|
print(report)
|
|
print(f'\nReport written to {(report_dir / REPORT_MD).resolve()}')
|
|
|
|
duration = time.time() - duration
|
|
print()
|
|
print("-" * 30)
|
|
print(f'Total failed: {err_count} in {duration:.1f}s')
|
|
print("-" * 30)
|
|
sys.exit(err_count)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|