diff options
Diffstat (limited to 'test/hil/hil_test.py')
| -rwxr-xr-x | test/hil/hil_test.py | 2168 |
1 files changed, 1393 insertions, 775 deletions
diff --git a/test/hil/hil_test.py b/test/hil/hil_test.py index 4b87154b6..174251343 100755 --- a/test/hil/hil_test.py +++ b/test/hil/hil_test.py @@ -23,11 +23,15 @@ # 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_*) +# - System packages: sudo apt install mtools libmtp9 libmtp-runtime alsa-utils iperf +# mtools read_disk_file (device/cdc_msc, device/msc_dual_lun) +# libmtp9 pymtp ctypes load (device/mtp); Debian 13 uses libmtp9t64 +# libmtp-runtime mtp-probe and the completed-device /dev/libmtp-* marker +# alsa-utils arecord (device/audio_test_freertos) +# iperf throughput tests (device/net_lwip_*) +# openocd unified openocd from https://github.com/hathach/openocd (branch tinyusb) for wch, rp2040/rp2350, analog max32 +# - device/usbtest: usbtest kernel module + testusb binary (kernel tools/usb/testusb.c) on PATH, +# plus sudo for modprobe / sysfs writes # - Python packages: pip install -r requirements.txt # # udev rules : @@ -36,97 +40,114 @@ import argparse import io +import itertools import os import random import re import select +import signal +import shlex import sys +import tempfile import time -import signal from contextlib import redirect_stdout from pathlib import Path -from typing import Any, TypedDict, NotRequired, cast +from typing import TypedDict, NotRequired, cast import serial import subprocess +import traceback import json import glob -from multiprocessing import Pool, Lock +import multiprocessing from multiprocessing import TimeoutError as MpTimeoutError -import hashlib -import ctypes -from pymtp import MTP + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) # PYTHONSAFEPATH drops it +import hil_flash +from helper import hil_health, hil_lock, hil_util +from helper.hil_util import device_tests, dual_tests, host_test + +# Raw Lock/Semaphore objects in Pool initargs are inheritable only under fork +# (spawn/forkserver pickle them and fail at Pool creation), so pin it against an +# interpreter default change. Windows has no fork: fall back so it still IMPORTS there. + +_mp = multiprocessing.get_context('fork') if os.name != 'nt' else multiprocessing.get_context() +Pool, Lock, Semaphore, Manager = _mp.Pool, _mp.Lock, _mp.Semaphore, _mp.Manager import string -# --- per-board dev-session locks (see test/hil/board_lock.py) ------------ -BOARD_LOCK_DIR = '/tmp/tinyusb-hil-locks' +# Enumeration wait budget: first attempt ENUM_TIMEOUT, retries the shorter +# ENUM_TIMEOUT_RETRY -- a device that will enumerate shows up within seconds, so a failing +# test costs ~3-5x a passing one instead of 10-30x. Set per attempt by test_example(); a +# module global is safe because each pool worker is its own process. +ENUM_TIMEOUT = 8 +ENUM_TIMEOUT_RETRY = 4 +_enum_timeout = ENUM_TIMEOUT -def acquire_board_lock(board_name): - """Take this board's flock for the duration of its flash+test. - Returns an open file handle (keep it referenced; closing releases it), - or None when HIL_NO_BOARD_LOCK=1 or the lock dir is unusable (fail-open: - locking must never break a test run by itself). - Raises RuntimeError only when another session holds the board.""" - import fcntl - if os.environ.get('HIL_NO_BOARD_LOCK') == '1': - return None # user-authorized bypass — see board_lock.py / hil skill - try: - os.makedirs(BOARD_LOCK_DIR, exist_ok=True) - fd = os.open(os.path.join(BOARD_LOCK_DIR, f'{board_name}.lock'), - os.O_RDWR | os.O_CREAT, 0o666) - fh = os.fdopen(fd, 'r+') - except OSError as e: - # odd lock dir (perms, path collision): proceed unlocked, but say so — - # a silent fail-open is indistinguishable from the intentional bypass - print(f'warning: board lock unavailable for {board_name} ({e}); proceeding unlocked', - flush=True) - return None - try: - fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB) - except OSError: - try: - info = fh.read(500).strip() - except (OSError, UnicodeDecodeError): - info = '' - fh.close() - raise RuntimeError(f'board locked: {info or "unknown holder"}') - # announce ourselves so the other side's conflict message is truthful; - # best-effort — the flock itself is already held - try: - fh.truncate(0) - fh.seek(0) - json.dump({'pid': os.getpid(), 'reason': 'hil_test.py', - 'since': time.strftime('%Y-%m-%dT%H:%M:%S%z')}, fh) - fh.flush() - except OSError: - pass - return fh + +def enum_timeout() -> int: + """Enumeration wait budget for the current test attempt.""" + return _enum_timeout -ENUM_TIMEOUT = 15 +def wait_until(predicate, step: float = 1.0, timeout: float | None = None): + """Poll predicate under the per-attempt enum budget. Deadline-based so a slow predicate + body (subprocess, libmtp scan) counts against the budget. An explicit timeout overrides + that budget. Returns the first truthy predicate value, or None on timeout.""" + deadline = time.monotonic() + (enum_timeout() if timeout is None else timeout) + while True: + r = predicate() + if r: + return r + if time.monotonic() >= deadline: + return None + time.sleep(step) 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. +# Plain (non-ANSI) cell symbols for hil_report.md; a missing binary counts as skipped. REPORT_CELL = {'pass': '✅', 'fail': '❌', 'skip': '⚪'} + +class TestFail(AssertionError): + """Fail a test but still surface a metric string in its report cell (e.g. usbtest's '❌ 29/30' + instead of a bare ❌). The cell metric is icon-prefixed so render/tally treat it as a failure.""" + def __init__(self, msg: str, metric: str | None = None, parsed: bool = False): + super().__init__(msg) + self.metric = metric + # parsed=True: a real per-case verdict, so a retry would only re-observe it + # (test_example skips the rest). A failure to RUN the tool stays retryable. + self.parsed = parsed + + verbose = False +# Set when a HUNG usbtest case could not be recovered: the DUT's usbfs node still has a +# D-state holder, so every later flash on that board enumerates into it, blocks, survives +# SIGKILL and becomes another stray. maxtasksperchild=1 gives each board its own worker, +# so this global is board-scoped; test_board resets it anyway. +board_wedged = '' +max_retry = 1 # mirrors argparse's -r default (see main); defined HERE too so + # test_example is callable (and testable) without going through main() +PROFILE = os.environ.get('HIL_PROFILE') == '1' # timestamped logs + permit/flash timing + ctrl-map dump test_only = [] board_test = {} -build_dir = 'cmake-build' skip_flash = False print_lock = None +shuffle_seed = None # per-run seed for the per-board test-order shuffle (HIL_SHUFFLE_SEED to replay) +_current_fw = None # firmware test_example resolved for the RUNNING test (set before each test fn) -def init_worker(lock): - global print_lock +def init_worker(lock, seed, b_mutexes, f_sems, cmap, cmeta, hints_by_uid): + global print_lock, shuffle_seed print_lock = lock + shuffle_seed = seed + hil_lock.init_scheduling(b_mutexes, f_sems, cmap, cmeta, hints_by_uid, log_fn=log_line) def log_line(msg: str) -> None: + if PROFILE: + msg = f'{time.time():.3f} {msg}' out = sys.__stdout__ if sys.__stdout__ is not None else sys.stdout if print_lock is not None: with print_lock: @@ -138,13 +159,21 @@ def log_line(msg: str) -> None: 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()] + # Defense in depth (the emitter already suppresses them, see _ci_log_groups): markers + # piped into this capture land mid-row, where GitHub renders them literally. + lines = [] + for ln in raw.replace('\r', '\n').split('\n'): + ln = hil_util.strip_workflow_markers(ln.strip()).strip() + if ln: + lines.append(ln) return ' | '.join(lines) class FlasherCfg(TypedDict): name: str uid: str - args: str + args: NotRequired[str] # stlink entries carry no args + vid_pid: NotRequired[str] # openocd probe pin, verbatim (e.g. "0x2e8a 0x000c") + verify: NotRequired[bool] # openocd read-back verify opt-out (WCH) class AttachedDevCfg(TypedDict, total=False): @@ -188,67 +217,41 @@ class Board(TypedDict): 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')) +# Below the CI job ceilings so THIS guard fires first and still writes a report, well +# above a healthy fleet run (~14 min measured), and deliberately generous: firing early +# abandons boards that were still in flight (30 min fired on 5 of the last 8 HIL jobs), +# while firing late costs minutes on an already-wedged run. The drain keeps whatever had +# already finished either way. +POOL_TIMEOUT = hil_util.pos_int_env('HIL_POOL_TIMEOUT', 3600) -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) +# Headroom on top of a battery's own budget so ONE HUNG recovery (case timeout, SIGKILL +# wait, bounded reflash, settle) can finish. Only spent when cases actually time out. +USBTEST_RECOVERY_BUDGET = hil_util.pos_int_env('HIL_USBTEST_RECOVERY_BUDGET', 250) +# How long usbtest.py may keep starting new cases (--budget). The outer run_cmd timeout is +# always this PLUS the recovery headroom, never a separate literal, or lowering one eats +# the reserve the recovery needs. 0 is refused (usbtest.py reads it as "no limit"); the +# margin over a healthy battery (~200s) keeps contention from becoming BUDGET entries. +USBTEST_BATTERY_BUDGET = hil_util.pos_int_env('HIL_USBTEST_BATTERY_BUDGET', 260) -WCH_RISCV_CONTENT = """ -adapter driver wlinke -adapter speed 6000 -transport select sdi +# The battery checks its budget BEFORE dispatching a case, so it can overshoot by one +# already-started case. Our outer kill must sit ABOVE that or we SIGKILL the battery just +# as it goes to print its JSON, turning ~29 real per-case verdicts into "usbtest did not +# run" and re-paying the whole battery on retry. +# Worst case, from usbtest.py: --timeout 60 (the case) + 5s post-SIGKILL reap + +# dmesg_tail(), which is bounded by HELPER_TIMEOUT=30 and runs on BOTH the FAIL and HUNG +# timeout paths = 95s. 120 leaves a margin; 75 (my first estimate, taken before checking +# dmesg_tail) was 20s SHORT and would have killed the battery mid-print. +USBTEST_OVERSHOOT = 120 +SERIAL_READ_TIMEOUT = hil_util.pos_float_env('HIL_SERIAL_READ_TIMEOUT', 5) +SERIAL_WRITE_TIMEOUT = hil_util.pos_float_env('HIL_SERIAL_WRITE_TIMEOUT', 10) -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}' @@ -271,13 +274,12 @@ def get_alsa_capture_dev(id): def open_serial_dev(port: str): - timeout = ENUM_TIMEOUT + 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 + # write_timeout: see serial_write_all ser = serial.Serial(port, baudrate=115200, timeout=SERIAL_READ_TIMEOUT, write_timeout=SERIAL_WRITE_TIMEOUT) break @@ -293,53 +295,71 @@ def open_serial_dev(port: str): 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. + # write_timeout is a deadline for the whole call. A timeout means the device stopped + # draining, and it is 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') +LP_OPEN_TIMEOUT = 5 # bound on opening the printer lp node; see test_device_printer_to_cdc +# Runs under hil_util.run_alongside as `python3 -c`. Inline rather than a file so hil_ci.sh's +# staging list does not need another entry to keep the rig working. +LP_READER = ( + 'import os, sys\n' + 'fd = os.open(sys.argv[1], os.O_RDONLY)\n' + # readiness marker: the parent must not send a byte before the node is open, or the + # bytes are lost. A blind sleep raced CPython start-up on a loaded rig. + 'open(sys.argv[3], "w").close()\n' + 'want = int(sys.argv[2])\n' + 'buf = b""\n' + 'while len(buf) < want:\n' + ' chunk = os.read(fd, min(64, want - len(buf)))\n' + ' if not chunk:\n' + ' break\n' + ' buf += chunk\n' + 'sys.stdout.buffer.write(buf)\n' +) +MTYPE_TIMEOUT = 30 # a README-sized read is <1 s; bounds a D-state hang on a wedged device + + 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). + # Reads a file from an unmounted FAT volume; needs mtools. run_cmd everywhere in this + # file rather than subprocess.run/check_output: its post-timeout reap is an unbounded + # communicate() with no killpg (CPython 3.13.5 subprocess.py:558-565 -- kill(), then + # communicate() with NO timeout), which never returns on a device wedged in D state, + # where the kill is queued and never delivered. binary + # keeps the bytes exact, split_stderr keeps mtype warnings out of them. 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 try_read(): + nonlocal last_err + if not os.path.exists(dev): + return None + r = hil_util.run_cmd(f"mtype -i {shlex.quote(dev)} ::/{shlex.quote(fname)}", + timeout=MTYPE_TIMEOUT, binary=True, split_stderr=True, quiet=True) + if r.returncode == 0: + if r.stdout: + return r.stdout + # rc 0 with no data is an answer (empty file, zeroed sectors), not "not + # ready" — fail now instead of spinning the budget + raise AssertionError(f'Cannot read file {fname} from {dev}: mtype returned no data') + last_err = (r.stderr or b'').decode(errors='replace').strip() or f'mtype rc {r.returncode}' + return None + data = wait_until(try_read) + if data is None: + raise AssertionError(f'Cannot read file {fname} from {dev}: {last_err}' if last_err + else f'Storage {dev} not existed') + return data -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 + +# ~5 KB of transfers plus libmtp setup takes seconds, not minutes; a larger value makes a +# wedged MTP board cost that much on every retry, all charged to the pool guard. +MTP_SESSION_MARGIN = 30 # transfer budget after enumeration; past it the session is killed def get_printer_dev(id: str, vendor_str, product_str, ifnum: int): @@ -348,235 +368,30 @@ def get_printer_dev(id: str, vendor_str, product_str, ifnum: int): 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() + # bounded: same device_lock() exposure as the sibling reads (see read_sysfs) + sn = hil_util.read_sysfs(f'{lp}/device/../serial') + # UNKNOWN is not None: the sentinel has no __eq__, so an unanswered read + # would fall through both tests and read as 'not this board' -- the exact + # absence/unknown conflation read_sysfs exists to prevent. + if sn is None or sn is hil_util.SYSFS_UNKNOWN: + continue if sn == id: return f'/dev/usb/{os.path.basename(lp)}' - except (FileNotFoundError, PermissionError, ValueError): + except OSError: # read_sysfs swallows its own OSError/ValueError; glob can race 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: + def try_find(): 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') - + return lp_dev if lp_dev and os.path.exists(lp_dev) else None -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) + lp_dev = wait_until(try_find) + assert lp_dev, (f'Printer device not found for {id} if{ifnum:02d}' + + hil_util.sysfs_blind_note()) + return lp_dev # ------------------------------------------------------------- @@ -585,18 +400,16 @@ def reset_lm4flash(board): 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) + port = hil_util.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 + 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) @@ -633,22 +446,20 @@ 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) + port = hil_util.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) + ret = getattr(hil_flash, 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 + 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) @@ -712,17 +523,16 @@ def test_host_cdc_msc_hid(board): if not cdc_devs and not msc_devs: return 'skipped' - port = get_serial_dev(flasher["uid"], None, None, 0) + port = hil_util.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) + ret = getattr(hil_flash, 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 + timeout = enum_timeout() wait_cdc = len(cdc_devs) > 0 wait_msc = len(msc_devs) > 0 while timeout > 0: @@ -736,7 +546,6 @@ def test_host_cdc_msc_hid(board): 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', @@ -747,7 +556,6 @@ def test_host_cdc_msc_hid(board): 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] @@ -756,7 +564,6 @@ def test_host_cdc_msc_hid(board): 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' @@ -776,7 +583,6 @@ def test_host_cdc_msc_hid(board): 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() @@ -784,7 +590,6 @@ def test_host_cdc_msc_hid(board): 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: @@ -805,17 +610,16 @@ def test_host_msc_file_explorer(board): if not msc_devs: return 'skipped' - port = get_serial_dev(flasher["uid"], None, None, 0) + port = hil_util.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) + ret = getattr(hil_flash, 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 + timeout = enum_timeout() while timeout > 0: new_data = ser.read(ser.in_waiting or 1) if new_data: @@ -850,14 +654,12 @@ def test_host_msc_file_explorer(board): 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: @@ -874,12 +676,13 @@ def test_host_msc_file_explorer(board): 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 + 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(' ', '') + speed = f'{m.group(1)} {m.group(2)}' break ser.close() + assert speed is not None, 'MSC read produced no speed report (dd stalled or failed)' return speed @@ -891,15 +694,14 @@ def test_host_msc_file_explorer_freertos(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) + hil_util.get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 0), + hil_util.get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 2) ] ser = [open_serial_dev(p) for p in port] @@ -938,7 +740,7 @@ def test_device_cdc_dual_ports(board): def test_device_cdc_msc(board): uid = board['uid'] # CDC Echo test - port = get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 0) + port = hil_util.get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 0) ser = open_serial_dev(port) def rand_ascii(length): @@ -967,6 +769,20 @@ def test_device_cdc_msc_freertos(board): test_device_cdc_msc(board) +def link_is_fs(speed) -> bool: + """Payload scaling from a `speed` attribute. Anything not positively read as high speed + counts as FS -- including None and SYSFS_UNKNOWN: the FS payload merely tests an HS + board less, while the HS payload hard-fails a healthy FS board.""" + return speed not in ('480', '5000', '10000') + + +def dd_timeout(mib: float) -> int: + """Bound one dd by what was ASKED for: 2.5 s/MiB is the slowest rate this test has + measured (FS CDC, ~420 kB/s), over a 30 s floor. A flat bound fails a healthy board as + soon as the payload grows or the leaf-hub uplink is shared.""" + return int(30 + 2.5 * mib) + + def test_device_cdc_msc_throughput(board): uid = board['uid'] @@ -977,59 +793,64 @@ def test_device_cdc_msc_throughput(board): 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 + 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 + tty = hil_util.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 + # Detect speed (12 Mbps FS / 480 Mbps HS) for payload scaling; a device we never find + # keeps the FS payload (see link_is_fs) + # usb_scan, not a private glob: it skips root hubs and remembers paths that already + # stranded, so one wedged peer cannot spend this worker's blindness budget four reads + # at a time. + is_fs = True + speed_known = False + devs, _ = hil_util.usb_scan(vid='cafe', serial=uid) + if devs: + speed = hil_util.read_sysfs(os.path.join(devs[0]['dir'], 'speed')) + is_fs = link_is_fs(speed) + speed_known = speed not in (None, hil_util.SYSFS_UNKNOWN) # 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)}' + rs = hil_util.run_cmd(f'timeout 30 stty -F {tty} raw -echo') + assert rs.returncode == 0, f'stty failed: {hil_util.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' + t_cdc, t_msc = dd_timeout(cdc_count / 16), dd_timeout(msc_count) - 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)) + rw = hil_util.run_cmd(f'timeout {t_cdc} dd if=/dev/zero of={tty} bs=64K count={cdc_count} 2>&1') + assert rw.returncode == 0, f'CDC dd write failed: {hil_util.cmd_stdout_text(rw.stdout)}' + cdc_w = parse_speed(hil_util.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)) + rr = hil_util.run_cmd(f'timeout {t_cdc} dd if={tty} of=/dev/null bs=64K count={cdc_count} iflag=fullblock 2>&1') + assert rr.returncode == 0, f'CDC dd read failed: {hil_util.cmd_stdout_text(rr.stdout)}' + cdc_r = parse_speed(hil_util.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)) + # inner bound, like the CDC pair above: run_cmd's SIGKILL is merely QUEUED against a + # dd blocked in the block layer on a half-dead device, so without one the call rides + # CMD_TIMEOUT and is abandoned holding the disk and usbfs nodes. + rmr = hil_util.run_cmd(f'timeout {t_msc} dd if={dev} of={tmp_file} bs=1M count={msc_count} iflag=direct 2>&1') + assert rmr.returncode == 0, f'MSC dd read failed: {hil_util.cmd_stdout_text(rmr.stdout)}' + msc_r = parse_speed(hil_util.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)) + rmw = hil_util.run_cmd(f'timeout {t_msc} dd if={tmp_file} of={dev} bs=1M count={msc_count} oflag=direct 2>&1') + assert rmw.returncode == 0, f'MSC dd write failed: {hil_util.cmd_stdout_text(rmw.stdout)}' + msc_w = parse_speed(hil_util.cmd_stdout_text(rmw.stdout)) try: os.remove(tmp_file) @@ -1037,40 +858,66 @@ def test_device_cdc_msc_throughput(board): 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)}' + + # report cell, e.g. "✅ C 652/422k M 1.1M/783k" (C=CDC, M=MSC; shared unit shown once) + def short(s): + return (s.split()[0].rstrip('0').rstrip('.') + s.split()[-1][0]) if ' ' in s else s + + def pair(r, w): + r, w = short(r), short(w) + if r[-1:] == w[-1:] and r[-1:].isalpha(): + r = r[:-1] + return f'{r}/{w}' + + # 'FS?' when the speed could not be read: the numbers below were produced against the FS + # payload, so an HS board reads as suspiciously slow. Say so rather than publish a green + # cell whose scale is a guess. + scale = '' if speed_known else ' FS?' + return f'{REPORT_CELL["pass"]} C {pair(cdc_r, cdc_w)} M {pair(msc_r, msc_w)}{scale}' def test_device_dfu(board): uid = board['uid'] + vid_pid = 'cafe:400b' - # 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: + # Deadline-based: dfu-util takes ~1 s per call, which a countdown would not charge + # against the budget. -d pins enumeration to THIS example's ids: a bare `-l` opens every + # DFU-capable node, and one wedged node blocks that open in D state. The pair is doubled + # because dfu-util matches run-time and DFU-mode devices against SEPARATE id pairs + # (parse_vendprod: an omitted DFU-mode pair matches ANY DFU-mode device). The deadline + # is only tested BETWEEN calls, so the per-call bound is what caps a blocked open. + deadline = time.monotonic() + enum_timeout() + found = False + while time.monotonic() < deadline: + ret = hil_util.run_cmd(f'dfu-util -d {vid_pid},{vid_pid} -l', timeout=15) + stdout = hil_util.cmd_stdout_text(ret.stdout) + if f'serial="{uid}"' in stdout and f'Found DFU: [{vid_pid}]' in stdout: + found = True break time.sleep(1) - timeout = timeout - 1 - assert timeout > 0, 'Device not available' + assert found, '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}') + # -d as well as -S: dfu-util matches the SERIAL only after libusb_open() (dfu_util.c + # probes the descriptor for iSerialNumber), so -S alone still opens every DFU-capable + # node. The id filter runs BEFORE the open; -S then picks our board (see the poll). + # Each partition is one short string, so a healthy upload is ~1 s; the bound is there + # for a node that stops answering mid-transfer. + ret = hil_util.run_cmd(f'dfu-util -d {vid_pid},{vid_pid} -S {uid} -a 0 -U {f_dfu0}', + timeout=30) assert ret.returncode == 0, 'Upload failed' - ret = run_cmd(f'dfu-util -S {uid} -a 1 -U {f_dfu1}') + ret = hil_util.run_cmd(f'dfu-util -d {vid_pid},{vid_pid} -S {uid} -a 1 -U {f_dfu1}', + timeout=30) assert ret.returncode == 0, 'Upload failed' with open(f_dfu0) as f: @@ -1085,17 +932,19 @@ def test_device_dfu(board): 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: + vid_pid = 'cafe:400c' + # enumeration pinned to this example's ids, same per-call bound (see test_device_dfu) + deadline = time.monotonic() + enum_timeout() + found = False + while time.monotonic() < deadline: + ret = hil_util.run_cmd(f'dfu-util -d {vid_pid},{vid_pid} -l', timeout=15) + stdout = hil_util.cmd_stdout_text(ret.stdout) + if f'serial="{uid}"' in stdout and f'Found Runtime: [{vid_pid}]' in stdout: + found = True break time.sleep(1) - timeout = timeout - 1 - assert timeout > 0, 'Device not available' + assert found, 'Device not available' def test_device_hid_boot_interface(board): @@ -1103,8 +952,7 @@ def test_device_hid_boot_interface(board): 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 + timeout = enum_timeout() while timeout > 0: if os.path.exists(kbd) and os.path.exists(mouse1) and os.path.exists(mouse2): break @@ -1120,12 +968,9 @@ def test_device_hid_composite_freertos(id): 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) + cdc_port = hil_util.get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 0) ser = open_serial_dev(cdc_port) lp_dev = open_printer_dev(uid, 'TinyUSB', 'TinyUSB_Device', 2) @@ -1145,7 +990,6 @@ def test_device_printer_to_cdc(board): 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 @@ -1155,7 +999,17 @@ def test_device_printer_to_cdc(board): ser.reset_input_buffer() rd = b'' offset = 0 - lp_fd = os.open(lp_dev, os.O_WRONLY | os.O_NONBLOCK) + # bounded: O_NONBLOCK does NOT save us -- usblp_open() takes the device mutex + # first -- and this open runs on the worker itself, with no thread to abandon + lp_fd = hil_util.bounded_open(lp_dev, os.O_WRONLY | os.O_NONBLOCK, 5) + # Three-valued on purpose: an OSError here is a FACT about the node (EBUSY from + # usblp's single-opener rule, ENOENT from a re-enumeration race, EACCES from a + # udev gap) and must not be reported as a wedge -- that sends the operator to + # usb-kernel-recover for hardware that is fine. + assert lp_fd is not hil_util.SYSFS_UNKNOWN, ( + f'printer: opening {lp_dev} for write blocked (device wedged)' + f'{hil_util.sysfs_blind_note()}') + assert lp_fd is not None, f'printer: {lp_dev} could not be opened for write' try: while offset < size: chunk_size = min(random.randint(1, 64), size - offset) @@ -1179,132 +1033,92 @@ def test_device_printer_to_cdc(board): 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 + # Test 2: CDC -> Printer with multiple sizes, write in random 1-64 byte chunks. + # The lp read runs in a PROCESS, not a thread: /dev/usb/lp* blocks on read, usblp + # allows a SINGLE opener, and a blocked thread cannot be abandoned without keeping + # that fd -- which poisoned the node for every later test this worker ran. A killed + # process takes its fd with it. 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) + ready = Path(tempfile.gettempdir()) / f'hil-lp-ready-{os.getpid()}-{size}' + ready.unlink(missing_ok=True) - # 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 + def write_cdc(): + # WAIT for the reader to have the node open. The child has to fork, exec and + # boot a CPython interpreter; on a loaded rig that routinely exceeds the 0.3s + # this used to sleep, and every byte sent early is lost -- surfacing as a + # spurious data mismatch rather than a timeout. + deadline = time.monotonic() + LP_OPEN_TIMEOUT + 5 + while not ready.exists(): + if time.monotonic() > deadline: + return # reader never opened; the rc/compare below reports it + time.sleep(0.02) + 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]}') + try: + r = hil_util.run_alongside( + [sys.executable, '-c', LP_READER, lp_dev, str(size), str(ready)], + write_cdc, LP_OPEN_TIMEOUT + 12) + finally: + ready.unlink(missing_ok=True) + # stderr, not stdout: run_alongside keeps the payload stream clean, so a traceback + # from the reader now arrives on its own pipe + assert r.returncode == 0, (f'CDC->Printer reader failed ({size} bytes, rc ' + f'{r.returncode}): {hil_util.cmd_stdout_text(r.stderr)[:200]}') + assert r.stdout == test_data, (f'CDC->Printer wrong data ({size} bytes):\n' + f' expected: {test_data[:64]}\n received: {r.stdout[:64]}') time.sleep(0.2) ser.close() def test_device_mtp(board): + # The whole session lives in mtp_test.py under run_cmd: libmtp calls are synchronous + # ctypes that block unkillably (D state) on a wedged device, so a disposable process is + # the only thing the harness can walk away from. 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() + script = Path(__file__).resolve().parent / 'mtp_test.py' + # 2x, as master's in-process open_mtp_dev used: libmtp-runtime publishes + # /dev/libmtp-* only after its SYNCHRONOUS mtp-probe finishes, seconds on a freshly + # flashed FS board, and the gio unmount eats part of what is left before the first + # probe. Extracting the session into a subprocess halved this by accident (8s/4s), + # which fails healthy hardware on the retry. + t = 2 * enum_timeout() + r = hil_util.run_cmd( + f'{shlex.quote(sys.executable)} {shlex.quote(str(script))} --uid {shlex.quote(uid)} --timeout {t}', + timeout=t + MTP_SESSION_MARGIN) + if r.returncode == 124: + # "abandoned", not "killed": a session blocked in a usbfs ioctl (D state) never + # receives the SIGKILL -- it lingers until its device path clears, by design + raise AssertionError(f'MTP session wedged (abandoned after {t + MTP_SESSION_MARGIN}s; ' + f'the session process may linger unkillable in D state)') + assert r.returncode == 0, f'MTP session failed (rc {r.returncode}):\n{r.stdout}' 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). + # iface enx<MAC_lowercase_no_colons>. Device IP 192.168.7.1, iperf2 TCP server on 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. + # Wait for an IPv4 address in the device's subnet (it serves DHCP); 30s because USB + # enum + DHCP serve is slower on the CI HIL hardware than locally. iface_timeout = 30 - deadline = time.time() + iface_timeout + deadline = time.monotonic() + iface_timeout host_ip = None - while time.time() < deadline: + while time.monotonic() < 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 @@ -1314,11 +1128,10 @@ def test_device_net_lwip_webserver(board): 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 + # Poll until the device accepts: the net stack and the iperf bind come up after DHCP. + deadline = time.monotonic() + enum_timeout() last_err = None - while time.time() < deadline: + while time.monotonic() < deadline: try: with socket.create_connection((device_ip, iperf_port), timeout=1): last_err = None @@ -1326,14 +1139,14 @@ def test_device_net_lwip_webserver(board): 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}' + 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() + # 5-second iperf2 TCP test; -y C for stable parsing (final summary line is + # timestamp,src_ip,src_port,dst_ip,dst_port,id,interval,bytes,bps). + ret = hil_util.run_cmd(f'iperf -c {device_ip} -t 5 -y C', + timeout=30, split_stderr=True, quiet=True) + stderr = (ret.stderr or '').strip() + stdout = (ret.stdout or '').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})' @@ -1344,19 +1157,16 @@ def test_device_net_lwip_webserver(board): 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}' @@ -1365,8 +1175,7 @@ def test_device_msc_dual_lun(board): def test_device_midi_test(board): uid = board['uid'] - # Find MIDI device via /dev/snd/by-id using board UID - timeout = ENUM_TIMEOUT + timeout = enum_timeout() midi_port = None while timeout > 0: pattern = f'/dev/snd/by-id/usb-*_{uid}-*' @@ -1383,31 +1192,40 @@ def test_device_midi_test(board): 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 = [] + midi_fd = os.open(midi_port, os.O_RDONLY | os.O_NONBLOCK) + try: + data = bytearray() # 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 + end_time = time.monotonic() + 3 + while (remaining := end_time - time.monotonic()) > 0: + ready, _, _ = select.select([midi_fd], [], [], min(0.5, remaining)) + if not ready: + continue + try: + chunk = os.read(midi_fd, 64) + except BlockingIOError: + continue + if not chunk: + break + data.extend(chunk) + finally: + os.close(midi_fd) + + notes = [] + # 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, @@ -1425,7 +1243,7 @@ def test_device_audio_test_freertos(board): return 'skipped' pcm = None - timeout = ENUM_TIMEOUT + timeout = enum_timeout() while timeout > 0: pcm = get_alsa_capture_dev(uid) if pcm: @@ -1448,8 +1266,11 @@ def test_device_audio_test_freertos(board): 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()}' + # run_cmd: ALSA capture from a wedged device blocks in D state (see read_disk_file) + ret = hil_util.run_cmd(' '.join(shlex.quote(c) for c in cmd), + timeout=20, split_stderr=True, quiet=True) + assert ret.returncode == 0, \ + f'arecord failed: {(ret.stderr or "").strip() or (ret.stdout or "").strip()}' try: with open(raw_path, 'rb') as f: @@ -1467,32 +1288,23 @@ def test_device_audio_test_freertos(board): 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 + # The producer is already running while ALSA activates streaming, so the + # initial overwritable software FIFO (at most 224 samples) can transition + # between ramp generations. After that startup window, require an exact ramp. + startup_samples = 256 + for i in range(startup_samples, sample_count - 1): + expected = (samples[i] + 1) & 0xFFFF + assert samples[i + 1] == expected, ( + f'Audio mismatch at sample {i + 1}: expected {expected}, got {samples[i + 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='') + print(f' ALSA {pcm}', 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 + timeout = enum_timeout() dev = None while timeout > 0: for d in hid.enumerate(0xCafe): @@ -1508,7 +1320,6 @@ def test_device_hid_generic_inout(board): 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)]) @@ -1524,60 +1335,186 @@ def test_device_hid_generic_inout(board): 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 -] +def test_device_usbtest(board): + global board_wedged + # Runs test/hil/usbtest.py against the cafe:4010 device; the pass count goes in the + # report cell ("✅ 30/30", or "❌ 29/30" on a partial). + uid = board['uid'] + + def usbtest_enumerated(): + """True, False, or None when a bounded read did not answer -- absence unproven.""" + # vid_pid FIRST: right after flashing, the previous example's enumeration (same + # serial, different PID) can linger and would fail usbtest.py's lookup -- and + # filtering on the two lock-free descriptor fields rules out every other device + # on the bus before the one read that can block. usb_scan memoises paths that + # already stranded, so one wedged peer cannot spend the blindness budget here. + devs, unknown = hil_util.usb_scan(vid_pid=('cafe', '4010'), serial=uid) + if devs: + return True + return None if unknown else False -dual_tests = [ - 'dual/host_info_to_device_cdc', -] + end = time.monotonic() + enum_timeout() + seen = usbtest_enumerated() + while time.monotonic() < end and seen is not True: + time.sleep(0.2) + seen = usbtest_enumerated() + # fail before usbtest_permit: an absent device would otherwise queue on the battery + # mutex for minutes behind real batteries just to have usbtest.py report "no device" + if seen is not True: + # 0/30 rather than a bare cell: the battery never ran (30 = standard case count) + raise TestFail( + f'no cafe:4010 device with serial {uid}' if seen is False else + f'cannot tell whether cafe:4010 {uid} is present: the bounded sysfs reads did ' + f'not answer{hil_util.sysfs_blind_note()}', + metric=f'{REPORT_CELL["fail"]} 0/30') + # settle: right after flashing the enumeration can bounce once (and on dual-port parts + # the other port's stale node — same serial and PID — lingers), and testusb run into + # that gap sees the device drop mid-case + time.sleep(3) -host_test = [ - 'host/cdc_msc_hid', - 'host/msc_file_explorer', - 'host/msc_file_explorer_freertos', - 'host/device_info', -] + # --keep-binding is required for concurrent batteries: usbtest.py's cleanup unbinds + # EVERY usbtest-bound interface, killing a peer battery under USBTEST_PARALLEL > 1, and + # that unbind path has also wedged a host xHCI (usb_hcd_alloc_bandwidth) here. Harmless + # to leave: the next example enumerates under a different PID. + script = Path(__file__).resolve().parent / 'usbtest.py' + # --budget makes the battery a real bound: repeated case timeouts (a FAIL, not a HUNG, + # so the battery keeps going) can otherwise spend the whole outer timeout inside the + # case loop, leaving the recovery below nothing. + cmd = (f'{shlex.quote(sys.executable)} {shlex.quote(str(script))} ' + f'--serial {shlex.quote(uid)} --json --keep-binding ' + f'--timeout 60 --budget {USBTEST_BATTERY_BUDGET}') + # Post-hang recovery reflashes the DUT through its own probe, NEVER a root-port cycle + # (one board reached instead of every fixture under the port; see usb-kernel-recover). + # _current_fw is the artifact test_example flashed for THIS test: re-deriving it from + # board['name'] reflashes the wrong build on variant-only boards. --outer-timeout lets + # usbtest skip a reflash it cannot finish before our run_cmd kill, which would orphan + # the flasher (own session) on the probe. Never under --skip-flash -- and say so: a + # HUNG case then holds the DUT's usbfs lock for the rest of the run, and a probe reset + # is no substitute (the DWC2 pullup survives a core halt). + # ...and only when this flasher can DELIVER that reflash past a poisoned node + # (hil_flash.convoy_safe). Otherwise the flags cost twice: the delivery adds a SECOND + # stray, and the board reserves recovery budget for a path that cannot fire. + # The RECOVERY flasher, which may be the roster's optional `flasher_recover` rather + # than the primary -- a jlink/stlink board can name an openocd entry that reaches the + # same probe convoy-safely without changing how the board is normally flashed. + _rec_flasher = hil_flash.recover_flasher(board) + recovery = bool(_current_fw and not skip_flash and hil_flash.convoy_safe(_rec_flasher)) + # ONE bound, computed here and used for BOTH the child's --outer-timeout and our own + # run_cmd kill below. Three separate expressions disagreed: --skip-flash appended no + # --outer-timeout at all (usbtest reads 0 as "no limit"), and the no-recovery branch + # narrowed only the CHILD's view while run_cmd still waited the full reserve -- so a + # board that cannot recover held a pool worker AND its battery permit idle for + # USBTEST_RECOVERY_BUDGET it had no way to spend, under a usbtest width of 2. + outer = USBTEST_BATTERY_BUDGET + (USBTEST_RECOVERY_BUDGET if recovery + else USBTEST_OVERSHOOT) + if _current_fw and skip_flash: + print('note: --skip-flash disables usbtest hang recovery; a HUNG case will leave ' + 'the device wedged until it is reflashed', flush=True) + elif _current_fw and not recovery: + print(f'note: {_rec_flasher["name"]} cannot deliver a reflash past a poisoned ' + f'usbfs node, so usbtest hang recovery is disabled for {board["name"]}; a ' + f'HUNG case will leave it wedged for the rest of the run', flush=True) + if recovery: + # ship the RECOVERY flasher as `flasher`: usbtest.py, recovery_steps and + # convoy_safe all read board['flasher'], so substituting here keeps the entire + # child side unaware that a second roster entry exists + rb = json.dumps({'name': board['name'], 'flasher': _rec_flasher}) + cmd += f' --recover-board {shlex.quote(rb)} --recover-fw {shlex.quote(_current_fw)}' + cmd += f' --outer-timeout {outer}' + # The reserve above USBTEST_BATTERY_BUDGET exists because the battery can overrun by + # one already-started case, and a hang there needs room for the recovery (whose reflash + # is bounded by usbtest.RECOVER_FLASH_TIMEOUT, not HIL_CMD_TIMEOUT). Without it run_cmd + # SIGKILLs usbtest.py mid-recovery, losing the JSON and the diagnosis. + with hil_lock.usbtest_permit(uid): + # split_stderr: the battery's final JSON is parsed from stdout, and stderr is the + # only detail left when the outer timeout kills the battery before it prints + r = hil_util.run_cmd(cmd, timeout=outer, split_stderr=True) + out = hil_util.cmd_stdout_text(r.stdout) + brace = out.find('{') + try: + # brace < 0 would slice from the END ('...rc 0' -> '0' -> int 0, whose subscript + # raises TypeError outside the tuple below and loses the diagnosis) + if brace < 0: + raise ValueError('no JSON object on stdout') + data = json.loads(out[brace:]) + passed, failed = int(data['passed']), int(data['failed']) + except (ValueError, KeyError, TypeError, json.JSONDecodeError): + # compact BOTH, never `or`: a battery SIGKILLed mid-print leaves a truthy JSON + # fragment on stdout, so an `or` drops the stderr that explains the failure + parts = [compact_output(hil_util.cmd_stdout_text(r.stderr)), compact_output(out)] + detail = ' | '.join(p for p in parts if p) + # Retryable even on rc 124 (run_cmd's outer kill), though the retry re-pays the + # whole budget: 124 only says the timer expired, which a healthy battery can hit + # under load, and test_example REFLASHES before each attempt. Where usbtest's + # in-band recovery is off (--skip-flash, a flasher failing convoy_safe, a terminal + # wedge) that reflash is the only thing left to unpoison the DUT for the boards + # that share its controller. + # No JSON to read the verdict from, so fall back to the text: a battery SIGKILLed + # mid-hang still says HUNG on stdout, and this raise happens BEFORE the latch below + # -- which is why the outer-timeout case, the likeliest real wedge, never latched. + if 'HUNG' in out: + board_wedged = (f'{board["name"]}: usbtest reported a hang and was killed ' + f'before it could report a verdict') + raise TestFail(f'usbtest did not run: {detail}', + metric=f'{REPORT_CELL["fail"]} 0/30') + # A HUNG case that recovery could not clear leaves a D-state holder on this board's + # usbfs node. Latch it: the remaining examples would each flash THROUGH that node, + # block, survive SIGKILL and add another stray -- turning one wedge into one stray per + # remaining example, which is the convoy this branch exists to contain. + # The battery's OWN verdict first: `recovery` only says the flags were passed, not that + # the reflash worked, so a convoy-safe board whose recovery failed used to come back + # unlatched and flash every remaining example through the poisoned node. + if data.get('wedged') or (not recovery and 'HUNG' in out): + # _rec_flasher, NOT board['flasher']: recovery was decided against recover_flasher() + # at the top of this function, and the two diverge as soon as a roster carries the + # optional `flasher_recover` key -- naming the wrong one sends the operator to the + # wrong probe. The wording stays on what usbtest actually reported ("still wedged"), + # because unrecovered_hang is also set by the ambiguous/inconclusive aborts, where + # nothing hung and the old text was false on both clauses. + board_wedged = (f'{board["name"]}: usbtest reports the device still wedged ' + + (f'after a recovery reflash via {_rec_flasher["name"]}' if recovery + else f'and {_rec_flasher["name"]} cannot deliver a recovery reflash')) -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 + # notrun counts toward the denominator but is NOT a failure: listing cases that never + # ran as failures sends a maintainer bisecting one of them. + notrun = int(data.get('notrun', 0)) + total = passed + failed + notrun + if board_wedged and failed == 0 and notrun == 0: + # Every case passed and the device STILL wedged -- usbtest's inconclusive/ambiguous + # abort fires after the last case, so nothing back-fills a BUDGET entry. Reporting + # the pass would exit 0 with a D-state holder on the rig and the board absent from + # the re-run spec. parsed=True: a retry re-pays the whole battery to re-observe a + # wedge, and flashes through the poisoned node to do it. + raise TestFail(f'usbtest {passed}/{total} but the device wedged ({board_wedged})', + metric=f'{REPORT_CELL["fail"]} {passed}/{total}', parsed=True) + if failed == 0 and notrun == 0 and total > 0: + return f'{REPORT_CELL["pass"]} {passed}/{total}' + bad = [c.get('num') for c in data.get('cases', []) + if c.get('status') not in ('PASS', 'BUDGET')] + why = f'usbtest {passed}/{total}' + if bad: + why += f' (cases failed: {bad})' + if notrun: + # the reason is per BUDGET entry: a hang or a device drop also aborts the battery, + # and blaming the budget points the maintainer at the wrong thing + reasons = {c.get('detail', '') for c in data.get('cases', []) + if c.get('status') == 'BUDGET'} + reason = (reasons.pop().replace('not run: ', '') if len(reasons) == 1 + else 'the battery stopped early') + why += f'; {notrun} case(s) never ran ({reason}), so this says nothing about them' + # parsed ONLY when every case ran: an aborted battery (budget expiry, kernel hang, bus + # drop) leaves BUDGET entries, and those are exactly what a reflash retry can fix. + raise TestFail(why, metric=f'{REPORT_CELL["fail"]} {passed}/{total}', + parsed=(notrun == 0)) -def test_example(board: Board, variant: str, example: str) -> tuple[int, str]: +# ------------------------------------------------------------- +# Main +# ------------------------------------------------------------- + + +def test_example(board: Board, variant: str, example: str) -> tuple[int, str, str | None]: """ Test example firmware :param board: board dict @@ -1594,26 +1531,69 @@ def test_example(board: Board, variant: str, example: str) -> tuple[int, str]: test_name = f'{variant:40} {example:30} ...' - fw_name = find_firmware(variant, example) + # --skip-flash runs whatever is already on the board, so any build counts as present; + # filtering by flasher there would skip the test over an extension it never uses. + fw_name = hil_flash.find_firmware(variant, example, + flasher=None if skip_flash else board['flasher']['name']) if fw_name is None: log_line(f'{test_name} Skip (no binary)') return 0, 'skip', None + # usbtest's hang recovery reflashes the exact artifact under test; re-deriving it from + # board['name'] breaks on variant-only boards + global _current_fw + _current_fw = str(fw_name) if verbose: - log_line(f'Flashing {fw_name}.elf') + log_line(f'Firmware {fw_name}') - # flash firmware (unless --skip-flash), then run the test. Both may fail randomly, - # retry a few times. + global _enum_timeout start_s = time.time() flash_ok = True last_err = '' last_detail = '' + wedge_break = False for i in range(max_retry): + if board_wedged and i: + # The latch is set MID-attempt (a HUNG usbtest whose flasher cannot recover), + # so test_board's check between tests is too late for THIS test's own retries: + # every further attempt re-flashes into the D-state-held node, blocks, survives + # SIGKILL and leaves another stray. The wedge is not something a retry can fix. + log_line(f'{test_name} not retrying: {board_wedged}') + # COUNT it. Breaking out here skips the i == max_retry - 1 branch that would + # have incremented err_count, so the board rendered a red cell, contributed 0 + # to the exit status and was omitted from the re-run spec -- a rig left with a + # D-state holder published under sys.exit(0). Latent at CI's --retry 1, live + # for every local run and for the workflows that pass no -r. + wedge_break = True + break + _enum_timeout = ENUM_TIMEOUT if i == 0 else ENUM_TIMEOUT_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) + with hil_lock.flash_permit(board['uid']): + t_flash = time.monotonic() + try: + ret = getattr(hil_flash, + f'flash_{board["flasher"]["name"].lower()}')(board, str(fw_name)) + except Exception as e: + # A flasher that RAISES (esptool's get_serial_dev when the adapter + # drops off the bus, a missing config.env, an unwritable CWD) would + # propagate out of the worker and abort the whole drain, costing + # every board still in flight. + print(f'flash raised: {type(e).__name__}: {e}', flush=True) + ret = subprocess.CompletedProcess(args='flash', returncode=1, + stdout=f'{type(e).__name__}: {e}') + if PROFILE: + log_line(f'[prof] {variant} {example} flash attempt {i + 1}: ' + f'{time.monotonic() - t_flash:.1f}s rc={ret.returncode}') + flash_ok = (ret.returncode == 0) + # A wedged RP2040/RP2350 DAP answers nothing and the probe has no + # reset line, so the retry fails identically; POR it via the Rescue DP + # first (no-op otherwise). NOT gated on a remaining attempt: CI HIL jobs + # run --retry 1, and this leaves the DAP POR'd for the jobs that follow. + if not flash_ok and \ + hil_flash.rescue_openocd(board, hil_util.cmd_stdout_text(ret.stdout)): + log_line(f'{variant} {example}: DAP wedged, rescued via Rescue DP') if flash_ok: try: tret = globals()[f'test_{example.replace("/", "_")}'](board) @@ -1624,7 +1604,6 @@ def test_example(board: Board, variant: str, example: str) -> tuple[int, str]: 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: @@ -1635,8 +1614,21 @@ def test_example(board: Board, variant: str, example: str) -> tuple[int, str]: except Exception as e: last_err = str(e) last_detail = compact_output(attempt_out.getvalue()) + if getattr(e, 'parsed', False): + # a PARSED per-case result (usbtest's "29/30"): retrying re-pays + # the whole battery, inside the fleet's usbtest permit, to + # re-observe a number the JSON already reported. Only that case. + err_count += 1 + metric = getattr(e, 'metric', None) + 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) + break if i == max_retry - 1: err_count += 1 + metric = getattr(e, 'metric', None) msg = f'{test_name} {STATUS_FAILED}: {e}' if last_detail: msg += f' {last_detail}' @@ -1668,13 +1660,21 @@ def test_example(board: Board, variant: str, example: str) -> tuple[int, str]: msg += f' in {time.time() - start_s:.1f}s' log_line(msg) + if wedge_break and not err_count: + # ONE error for the test, never two: a board that also failed to flash has already + # been counted just above. Without this the test returns 0 -- red cell, clean exit + # status, absent from the re-run spec. + err_count += 1 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).""" + Output goes to cmake-build/cmake-build-<variant>/ (tools/build.py layout). + + Unbounded on purpose: --build is a local convenience (no CI workflow passes it), so + the developer watching the build is the timeout.""" name = board['name'] bcfg = cast(BuildCfg, board.get('build', {})) extra_defs = bcfg.get('args', []) @@ -1682,7 +1682,7 @@ def build_board(board: Board) -> tuple[str, int]: failed = 0 for v in variants: - cmd = [sys.executable, str(TINYUSB_ROOT / 'tools' / 'build.py'), '-b', name] + cmd = [sys.executable, str(hil_util.TINYUSB_ROOT / 'tools' / 'build.py'), '-b', name] for d in extra_defs: cmd += ['-D', d] if v['name'] != name: @@ -1694,33 +1694,51 @@ def build_board(board: Board) -> tuple[str, int]: if verbose: cmd.append('-v') print(f' + {" ".join(cmd)}') - r = subprocess.run(cmd, cwd=TINYUSB_ROOT) - if r.returncode != 0: + # stdio is inherited so the build STREAMS: a silent buffer is + # indistinguishable from a stall. + proc = subprocess.Popen(cmd, cwd=hil_util.TINYUSB_ROOT, start_new_session=True) + try: + rc = proc.wait() + except KeyboardInterrupt: + # start_new_session means the build never saw the terminal's SIGINT + try: + os.killpg(proc.pid, signal.SIGKILL) + except OSError: + proc.kill() + raise + if rc != 0: failed += 1 return name, failed -def test_board(board: Board) -> tuple[str, int, list[str], list]: +# pseudo-test column for a variant boundary the park-flash could not clear (see below) +BOUNDARY_CELL = 'same-PID boundary' + + +def test_board(board: Board) -> tuple[str, int, list[str], list, float]: + swept = False name = board['name'] flasher = board['flasher'] + global board_wedged + board_wedged = '' try: - _lock_fh = acquire_board_lock(name) + _lock_fh = hil_lock.acquire_board_lock(name) except RuntimeError as e: log_line(f'{name:25} {STATUS_FAILED}: {e}') - # visible report row so the ❌ matches the exit code; failed-tests stays - # empty so a re-run repeats the whole board (no bogus -bt test filter) - return name, 1, [], [(name, {'board-locked': 'fail'})] + # visible report row so the ❌ matches the exit code; failed-tests stays empty so a + # re-run repeats the whole board (no bogus -bt filter) + return name, 1, [], [(name, {'board-locked': 'fail'}, None)], 0.0 + # after the lock: flock wait behind a concurrent run is not board cost + t_board = time.monotonic() try: - # 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). + # Explicit -t: filter against the board's capabilities, or a device-only board + # runs host/dual tests whose `dev_attached` config entry does not exist. board_tests = board.get('tests', {}) if 'only' in board_tests: allowed = set(board_tests['only']) @@ -1749,33 +1767,132 @@ def test_board(board: Board) -> tuple[str, int, list[str], list]: err_count = 0 failed_tests = [] - rows = [] # list of (row_label, {example: status}) — one row per build variant + board_wide_fail = False # re-run the whole board, not a subset of its tests + rows = [] # list of (row_label, {example: status}, duration) — one per build variant + # a -t/-bt filtered run times only a subset; report no duration so an accumulate + # re-run keeps the previous full-run value + partial = bool(test_only) or name in board_test variants = board.get('variant') or [{'name': name, 'flags': ''}] + prev_last = None # last test of the previous variant: the variant boundary is an adjacency too for v in variants: vname = v['name'] + # Shuffle each (board, variant)'s run order: spreads batteries and flash churn + # across the timeline instead of convoying, and surfaces order-dependent bugs. + # Seeded for replay (HIL_SHUFFLE_SEED). Unique per-example PIDs re-enumerate + # between examples; only the variant boundary can repeat one. + run_list = list(test_list) + if shuffle_seed is not None and len(run_list) > 1: + random.Random(f'{shuffle_seed}:{name}:{vname}').shuffle(run_list) + if run_list[0] == prev_last: + run_list[0], run_list[-1] = run_list[-1], run_list[0] cells = {} - for test in test_list: + if run_list and run_list[0] == prev_last and not skip_flash: + # Same example (same PID) still repeats across the boundary (a one-test + # -bt run has nothing to swap with). Park on board_test first: it disables + # the board's USB, so the next flash must re-enumerate to be seen. + t_park = time.monotonic() + # _should_park, same as the teardown park: this is attempt 0, so + # test_example's retry guard does not stop it flashing into a poisoned node + park_ec, park_status, _ = ( + test_example(board, vname, 'device/board_test') if _should_park(skip_flash) + else (0, 'skip', None)) + if park_ec or park_status == 'skip': + # Boundary not cleared: the previous variant may still be enumerated + # under the same PID, so this variant's tests could pass against ITS + # firmware. Skip them and record the boundary as the failure, so the + # report matches the exit code instead of rendering all-green. + # A 'skip' here has two very different causes: no board_test build, or + # _should_park refusing to flash a WEDGED board. Reporting the latter as + # a missing binary sends the operator hunting a build that exists. + wedge_skip = park_status == 'skip' and bool(board_wedged) + why = ('the board is wedged' if wedge_skip else + 'no board_test binary' if park_status == 'skip' else + 'park flash failed') + log_line(f'{vname:40} {"same-PID boundary":30} {STATUS_FAILED}: not cleared ({why}); ' + f'skipping {len(run_list)} test(s) on this variant') + # the wedge already charged its own error through test_device_usbtest; + # charging again would double-count one incident in the exit code + if not wedge_skip: + err_count += 1 + cells[BOUNDARY_CELL] = 'fail' + # blaming run_list[0] would re-run an innocent test that then passes, + # leaving the boundary unretested; re-run the whole board instead + board_wide_fail = True + # leave prev_last alone: the board still holds the previous variant's + # firmware, so the next variant must attempt the park again + run_list = [] + t_board += time.monotonic() - t_park # park is teardown, not board cost + if run_list: + prev_last = run_list[-1] + t_variant = time.monotonic() + for test in run_list: + if board_wedged: + # Do NOT flash through a poisoned node: each attempt enumerates into + # it, blocks uninterruptibly and leaves another stray behind. Report + # the skip so the cell is not mistaken for a pass. + cells[test] = f'{REPORT_CELL["skip"]} board wedged' + # ...and re-run the WHOLE board, like the boundary-failure path above: + # these tests never executed, so naming them individually in the .failed + # spec is not enough -- an --accumulate re-run that fixes only the wedged + # test would merge a green cell over it and leave these skips standing + # from the earlier attempt, forever, under a green job. + board_wide_fail = True + continue 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)) + if board_wedged: + log_line(f'{vname:40} SKIPPING the rest of this board: {board_wedged}; ' + f'flashing through the poisoned node would add a stray per test') + dur = f'{time.monotonic() - t_variant:.0f}s' if run_list and not partial else None + rows.append((vname, cells, dur)) - # 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: + # excludes the teardown park-flash below; a partial (filtered) run reports 0.0 so + # it never overwrites a cached full-run duration + t_total = 0.0 if partial else time.monotonic() - t_board + + # park: flash board_test last to disable the board's usb; teardown, not a test, + # so it is not recorded in the report. + # + # NOT on a wedged board: the latch has just skipped every remaining test precisely + # because flashing through a D-state-held node blocks, survives SIGKILL and leaves + # a stray -- and this park is a flash like any other. test_example's own guard does + # not stop it (that one only suppresses RETRIES, and this is attempt 0), so the + # containment path would add the very stray it exists to prevent. + if _should_park(skip_flash): test_example(board, variants[0]['name'], 'device/board_test') - return name, err_count, sorted(set(failed_tests)), rows + # Sweep HERE, not in main()'s finally: maxtasksperchild=1 retires this process as + # soon as it returns, reparenting anything it spawned to init and off the pool's + # ppid tree, so the main-side sweep walks fresh idle workers and finds nothing. + # Measured: 4 tasks, zero overlap, sweep 0, all 4 strays alive. + stray = hil_health.kill_own_children() + swept = True + + # LAST fields: whether this worker ran out of bounded-read budget, and what it could + # not kill. Only the worker can answer either -- the blindness latch is + # process-global and this is a separate process -- and the result tuple already + # crosses back, so no Manager round-trip. + return (name, err_count, [] if board_wide_fail else sorted(set(failed_tests)), + rows, t_total, hil_util.sysfs_blind(), stray) finally: + # A raise skips the sweep above, and maxtasksperchild=1 retires this process + # immediately afterwards -- reparenting its flasher to init and erasing the ppid + # link, so main's sweep cannot see it either. The count cannot reach the report on + # this path (there is no result tuple), but the KILL still frees the probe. + if not swept: + try: + hil_health.kill_own_children() + except Exception as se: # noqa: BLE001 - never mask the original failure + print(f'warning: stray sweep failed: {type(se).__name__}: {se}', flush=True) if _lock_fh: try: - # clear our pid record before dropping the flock: this worker - # process lives on (pool reuse), so a stale record would make - # board_lock.py's pid-liveness checks report a freed board as - # still locked for the rest of the run + # clear our pid record before dropping the flock: this worker process + # lives on (pool reuse), so a stale record would make hil_lock's + # pid-liveness checks report a freed board as locked for the rest of the run _lock_fh.truncate(0) except OSError: pass @@ -1784,22 +1901,43 @@ def test_board(board: Board) -> tuple[str, int, list[str], list]: REPORT_MD = 'hil_report.md' REPORT_JSON = 'hil_report.json' +# controller hints from previous runs: uid -> {'name', 'pci', 'duration'}. Only 'pci' is +# consumed (dispatch order and first-flash budgeting, never battery serialization). PCI +# addresses are boot-stable, so the cache survives reboots and goes stale on re-cabling. +CONTROLLER_CACHE = Path.home() / '.cache' / 'tinyusb-hil' / 'controller_cache.json' + + +def schedule_boards(boards: list, pci_of_uid: dict) -> list: + """Dispatch order: round-robin across host controllers so every controller's + serialized usbtest battery chain is fed from t=0 instead of one card's boards + convoying at the head of the queue. Boards without a controller hint form their + own bucket; config order is kept within a bucket.""" + buckets = {} + for b in boards: + buckets.setdefault(pci_of_uid.get(b['uid'], '?'), []).append(b) + return [b for grp in itertools.zip_longest(*buckets.values()) for b in grp if b is not None] 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 + """Render rows (list of (row_label, {example: status}, duration)) as an aligned + markdown matrix: columns = tests (bare names) centered, boards left-aligned, + per-row duration as the trailing column.""" seen = set() - for _, cells in rows_all: + 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 + # metric-bearing columns pinned first, the rest alphabetical: stable regardless of the + # shuffled execution order + pinned = ['usbtest', 'cdc_msc_throughput', 'msc_file_explorer', 'msc_file_explorer_freertos'] + + def col_key(t): + name = t.rsplit('/', 1)[-1] + return (pinned.index(name) if name in pinned else len(pinned), name, t) + + columns = sorted(seen, key=col_key) + headers = [c.rsplit('/', 1)[-1] for c in columns] + ['duration'] # bare example names def cell(cells, col): v = cells.get(col) @@ -1807,10 +1945,12 @@ def render_matrix(rows_all: list) -> str: return '' return REPORT_CELL.get(v, v) # status symbol, or a metric string (e.g. speed) verbatim + rows_vals = [(lbl, [cell(cells, c) for c in columns] + [dur or '']) + for lbl, cells, dur in rows_all] 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)] + board_w = max([len(board_hdr)] + [len(lbl) for lbl, _ in rows_vals]) + col_w = [max([len(h)] + [len(vals[i]) for _, vals in rows_vals]) + for i, h in enumerate(headers)] def line(label, values): padded = [label.ljust(board_w)] + [v.center(w) for v, w in zip(values, col_w)] @@ -1818,56 +1958,280 @@ def render_matrix(rows_all: list) -> str: 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] + body = [line(lbl, vals) for lbl, vals in rows_vals] - # 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()) + # tally run cells (not-run cells are absent from the dicts). A cell is a bare status or + # a metric string carrying its own icon ("❌ 29/30"), so classify by the leading icon. + def cell_kind(v): + if v == 'fail' or (isinstance(v, str) and v.startswith(REPORT_CELL['fail'])): + return 'fail' + if v == 'skip' or (isinstance(v, str) and v.startswith(REPORT_CELL['skip'])): + return 'skip' + return 'pass' + kinds = [cell_kind(v) for _, cells, _ in rows_all for v in cells.values()] + failed = kinds.count('fail') + skipped = kinds.count('skip') + passed = kinds.count('pass') 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: +def _write_failed_spec(failed_fname: Path, report_dir: Path, mret: list) -> None: + """Re-run spec: only the failed boards (-b), each restricted to its own failed tests + (-bt); a board with failures but no test list re-runs entirely. + + Shared with the pool-guard path, which feeds it the boards that never reported. That + path used to leave this unwritten -- and a fresh run has already unlinked it -- so + build.yml's "Get re-run spec" step found nothing and the GitHub re-run repeated the + whole fleet to find the one board that wedged.""" + parts = ['--accumulate'] + for name, err, fts, *_ in mret: + if err > 0: + parts.append(f'-b {name}') + if fts: + parts.append(f'-bt {name}:{",".join(fts)}') + if len(parts) > 1: # build-only failures have no boards to re-run + report_dir.mkdir(parents=True, exist_ok=True) + with failed_fname.open('w') as f: + f.write(' '.join(parts)) + else: + failed_fname.unlink(missing_ok=True) + + +class PoolDrainTimeout(MpTimeoutError): + """Guard expiry, carrying the rows that DID finish. + + They ride on the exception because the raise is the containment path: losing them here + is what map_async did, and what the drain exists to stop. + """ + + def __init__(self, finished: list): + super().__init__() + self.finished = finished + + +def drain_pool(it, boards: list, deadline: float, out: list | None = None) -> list: + """Collect imap_unordered results against ONE deadline. Returns the finished rows. + + Raises PoolDrainTimeout (carrying those same rows) when the deadline passes with boards + still in flight -- the caller keeps them, names only what is missing, and writes a + re-run spec covering just those. + + A function, not an inline loop, so the tests can call THIS instead of a copy of it: the + loop's previous test built its own ThreadPool and its own drain and asserted on those, + so deleting the real one outright kept the suite green. + """ + # `out` is the CALLER's list: a worker that raises something other than a timeout + # (get_serial_dev on a dropped adapter, a Manager EOFError) propagates bare, and a + # local accumulator would take every finished board with it -- the exact loss the + # drain replaced map_async to prevent. + mret: list = out if out is not None else [] + for _ in boards: + left = deadline - time.monotonic() + if left <= 0: + raise PoolDrainTimeout(mret) + try: + mret.append(it.next(timeout=left)) + except MpTimeoutError: + raise PoolDrainTimeout(mret) from None + return mret + + +def _should_park(skip_flash: bool) -> bool: + """Flash the teardown park (device/board_test, to switch the DUT's USB off)? + + Not on a wedged board. The latch has just skipped every remaining test precisely + because flashing through a D-state-held node blocks, survives SIGKILL and leaves a + stray -- and the park is a flash like any other. test_example's own guard does not stop + it either: that one only suppresses RETRIES, and the park is always attempt 0. So the + containment path would end by adding the very stray it exists to prevent. + """ + return not skip_flash and not board_wedged + + +def _stray_note(mret: list) -> str: + """Name the strays the workers could not kill, for the report banner. + + Summed from the result tuples rather than computed in main()'s finally: that finally + runs AFTER accumulate_report on both abort paths, so a banner appended there was + written to a variable nobody read again. + """ + dirty = [(r[0], r[6]) for r in mret if len(r) > 6 and r[6]] + if not dirty: + return '' + total = sum(n for _, n in dirty) + return (f'> **Rig dirty.** {total} process(es) survived SIGKILL and still hold a probe ' + f'or usbfs node into the next job: ' + f'{", ".join(f"{b} ({n})" for b, n in dirty)}.\n') + + +def _blind_note(mret: list) -> str: + """Name the boards whose worker went blind, for the report banner. + + A blind worker answers SYSFS_UNKNOWN for every attribute, so its "device not found" is + "could not tell". That already reaches the log and the per-cell failure text, but the + TABLE is what gets quoted -- and a red cell there is read as a broken board. Seen live + (run 31794359407): four workers blind, several cells red because of it, and a report + that said nothing. + + Per-board, not global: maxtasksperchild=1 gives every board a fresh worker, so a board + that ran on a healthy one is not smeared by a neighbour's wedge. Rows synthesised by + the timeout path are 5 fields wide and have nothing to report. + """ + blind = [r[0] for r in mret if len(r) > 5 and r[5]] + if not blind: + return '' + return (f'> **Not all verdicts are evidence.** {len(blind)} board(s) ran on a worker ' + f'that went blind on sysfs -- too many bounded reads stranded on a wedged ' + f'device -- so "not found" from them means "could not tell": ' + f'{", ".join(blind)}. See the usb-kernel-recover skill.\n') + + +def accumulate_report(mret: list, report_dir: Path, fresh: bool, scope: str = '', + banner: str = '') -> 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) + the markdown matrix to hil_report.md. `fresh` (a first run, no --accumulate) 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}} + already passed are preserved while re-run cells are updated. `scope` names the + board filter, if any, so a scoped table is not mistaken for a full one. + Returns the md.""" + acc = {} # ordered {row_label: [cells dict, duration str|None]} + prior_banner = '' 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']) + saved = json.loads(jpath.read_text()) + # CI keys the report dir by run id, so the sidecar is from an earlier attempt + for entry in saved.get('rows', []): + acc[entry['board']] = [dict(entry['cells']), entry.get('duration')] + # ... and so is the caveat those cells were collected under. A rerun on a rig + # that has since recovered contributes no banner, and the .failed spec reruns + # only FAILURES -- so the earlier attempt's passes are never re-earned and + # would be published as clean results of a rig that was not. + prior_banner = saved.get('banner', '') except (ValueError, KeyError, TypeError): pass # corrupt/old sidecar: start fresh - # merge this run: current cells override prior for boards/tests that ran - for name, _, _, rows in mret: - if rows and not any('board-locked' in cells for _, cells in rows): - # board ran for real this time: clear a stale lock-failure cell - # (its row is keyed by board name; test rows may be variant names) + # current cells override prior for boards/tests that ran; a filtered run reports + # duration None, keeping the previous full-run value + for name, _, _, rows, *_ in mret: + if rows and not any('board-locked' in cells for _, cells, _ in rows): + # board ran for real: clear a stale lock-failure cell (its row is keyed by + # board name; test rows may be variant names) stale = acc.get(name) if stale is not None: - stale.pop('board-locked', None) - if not stale: - # variant-keyed boards never repopulate the board-name row — - # drop it or it renders as a blank ghost row + stale[0].pop('board-locked', None) + if not stale[0]: + # variant-keyed boards never repopulate the board-name row, so drop it + # or it renders as a blank ghost row del acc[name] - for row_label, cells in rows: - acc.setdefault(row_label, {}).update(cells) + for row_label, cells, dur in rows: + row = acc.setdefault(row_label, [{}, None]) + # the boundary cell is only ever written on failure, so a re-run of this + # variant that cleared the boundary must drop the previous attempt's ❌ + if BOUNDARY_CELL not in cells: + row[0].pop(BOUNDARY_CELL, None) + row[0].update(cells) + if dur is not None: + row[1] = dur 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') + # by LINE, deduped: attempts repeat the same caveat far more often than they add a new + # one, and three copies of the D-state note reads as three incidents + seen, merged = set(), [] + for line in (prior_banner + banner).splitlines(): + if line.strip() and line not in seen: + seen.add(line) + merged.append(line) + banner = '\n'.join(merged) + '\n' if merged else '' + jpath.write_text(json.dumps({'rows': [{'board': k, 'cells': c, 'duration': d} + for k, (c, d) in acc.items()], + 'banner': banner}, indent=2) + '\n') - md = render_matrix(list(acc.items())) + md = render_matrix([(k, c, d) for k, (c, d) in acc.items()]) + if scope: + # a scoped run's small table is otherwise indistinguishable from a full one, and + # it replaces the previous full table in the sticky PR comment + md = f'_Scoped run: {scope}. Boards/tests not listed were not run._\n\n' + md + # LAST, so it is outermost: a rig-health caveat outranks the table AND the scope note, + # and the top of the report is where hil/SKILL.md tells the agent to look for it. + if banner: + md = banner + '\n' + md (report_dir / REPORT_MD).write_text(md + '\n', encoding='utf-8') return md +# containment paths print through hil_health._p: stdout may already be a dead pipe (a +# dropped ssh session), and a BrokenPipeError there would skip os._exit +_p = hil_health._p + + +def _abandon_exit(pool, mgr, abandoned: bool, err_count: int, + report: Path | None = None) -> None: + """Free the runner when the pool could not be shut down. Returns only if not abandoned. + + Must run even while an exception is propagating: multiprocessing's atexit handler + SIGTERMs its daemon workers (ignored in uninterruptible sleep) and then join()s them + with NO timeout, so an abandoned pool plus any raise between the pool's finally and + here hangs the interpreter until the job ceiling kills it. Reproduced: rc=124 at 25s + with SIGTERM-ignoring workers standing in for D state.""" + if not abandoned: + return + try: + if sys.exc_info()[0] is not None: + # os._exit below discards the traceback, and this is often the only place the + # real failure would ever be printed + traceback.print_exc() + except OSError: + pass + # Word this on evidence: shutdown_pool also returns False when terminate() RAISES, and + # a live worker after terminate() is what distinguishes a wedge from a harness bug. + # Count WORKERS only -- _pool_procs appends the Manager, our own healthy child, so + # including it made n >= 1 always and the harness-error branch unreachable. It is killed + # separately: os._exit skips its finalizer, and orphaned it holds the runner's stdout. + n = hil_health.kill_pool_children(pool) + hil_health.kill_pool_children(None, mgr) + if n: + _p(f'HIL worker pool would not terminate ({n} worker(s) still live, ' + f'uninterruptible); SIGKILLed them and abandoned the rest to free the ' + f'runner. Boards held by any leaked worker stay locked until the host is ' + f'power-cycled.', flush=True) + else: + _p('HIL worker pool shutdown failed but left no live worker behind, so this is ' + 'a harness error rather than a wedged rig -- see the Pool.terminate() ' + 'warning above. Exiting early anyway to free the runner; no board should ' + 'stay locked.', flush=True) + # A report already written by accumulate_report says nothing about the abandon, and a + # green table under a red job is how an agent ends up pasting it as this run's result. + # Prepend the caveat; best-effort, never at the cost of exiting. + if report is not None: + try: + if report.exists(): + # utf-8 explicitly (the cells are ✅/❌/⚪) and catch ValueError too: a torn + # report or a LANG=C locale raises UnicodeDecodeError -- NOT an OSError -- + # straight past os._exit, stranding the runner. + body = report.read_text(encoding='utf-8', errors='replace') + # Only when no banner is there yet, searched anywhere in the head rather + # than at char 0: write_timeout_report's banner must stay FIRST (its table + # is a PREVIOUS attempt's) and it puts the rig-health quote above itself. + if '**HIL run ab' not in body[:2000]: + report.write_text( + '**HIL run abandoned: the worker pool would not shut down.** The ' + 'table below was collected before the abandon; treat board ' + 'results as unverified.\n\n' + body, encoding='utf-8') + except (OSError, ValueError): + pass + try: + sys.stdout.flush() + except OSError: + pass + # Clamped: os._exit takes a status byte, so err_count == 256 would truncate to 0 and + # report a failing, abandoned run as green. + os._exit(min(err_count, 125) if err_count else 1) + + def main() -> None: """ Hardware test on specified boards @@ -1875,7 +2239,6 @@ def main() -> None: global verbose global test_only global board_test - global build_dir global max_retry global skip_flash @@ -1884,28 +2247,42 @@ def main() -> None: 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('--flasher', action='append', default=[], + help='Only boards using these flashers, e.g. esptool ' + '(for splitting one config across CI jobs)') + parser.add_argument('--exclude-flasher', action='append', default=[], + help='Exclude boards using these flashers') + parser.add_argument('-a', '--accumulate', action='store_true', + help='Merge results into the existing report instead of starting fresh ' + '(re-runs; the .failed file starts with this)') 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)') + # default 1, not 3: the pool guard is a FLAT 3600s that does not scale with max_retry, + # and one usbtest test at default 3 can burn 1530s of it (510s outer x3) for a single + # board. Every CI caller already pins --retry 1; the bare invocations in the hil skill + # and hil-validate.js run against the same one-slot rig and used to inherit 3. + parser.add_argument('-r', '--retry', type=int, default=1, help='Retry count for failed tests (default: 1)') parser.add_argument('-v', '--verbose', action='store_true', help='Verbose output') args = parser.parse_args() + if args.retry < 1: + # 0 would make every test loop body never run: all-red cells, exit 0 + parser.error('--retry must be >= 1') config_file = Path(args.config_file) boards = args.board - skip_boards = args.skip_board verbose = args.verbose + hil_util.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 + hil_flash.build_dir = args.build_dir max_retry = args.retry skip_flash = args.skip_flash @@ -1916,7 +2293,7 @@ def main() -> None: 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] + config_boards = list(config['boards']) else: unknown = [b for b in boards if b not in {e['name'] for e in config['boards']}] if unknown: @@ -1924,11 +2301,39 @@ def main() -> None: print(f'ERROR: board(s) not in {config_file.name}: {", ".join(unknown)}') sys.exit(1) config_boards = [e for e in config['boards'] if e['name'] in boards] + config_boards = [e for e in config_boards if e['flasher']['name'] not in args.exclude_flasher + and (not args.flasher or e['flasher']['name'] in args.flasher)] + if not config_boards: + # same reason the unknown -b board exits 1: 'No tests were run.' with rc 0 reads as + # a green HIL leg, so a roster edit emptying a leg's filter stops testing silently + msg = (f'No boards left after the flasher filter (--flasher ' + f'{args.flasher or "-"}, --exclude-flasher {args.exclude_flasher or "-"})') + print(msg, flush=True) + # loud AND leaving evidence: exiting with no report at all lets the PR comment + # keep the previous push's stale table under a red job + try: + rd = Path(os.environ.get('HIL_REPORT_DIR', '.')) + rd.mkdir(parents=True, exist_ok=True) + (rd / REPORT_MD).write_text(f'**HIL run selected no boards.** {msg}\n', + encoding='utf-8') + except OSError: + pass + sys.exit(1) + + + # Before the build: the probe needs nothing from it, and the annotation is more useful + # early than after a multi-board cmake build has been paid for. + # One line, not a probe: a D-state pid at start-up is a hint for whoever reads a red + # cell, never a reason to refuse the run. hil_pool_check does diagnosis. + note = hil_health.d_state_note() + if note: + log_line(f'rig note: {note}') + health_banner = f'> **Rig note.** {note}. Not a fault on its own -- a healthy testusb sits in D state for most of every case.\n' if note else '' 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}; ' + if hil_flash.build_dir != 'cmake-build': + print(f'warning: --build writes into cmake-build/, but -B is {hil_flash.build_dir!r}; ' f'tests will not find the freshly built firmware') print('-' * 30) print(f'Build phase: {len(config_boards)} board(s)') @@ -1940,51 +2345,264 @@ def main() -> None: 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. + # The report sidecar and the .failed re-run spec live in report_dir (CI keys it by run + # id: persistent across attempts, private to one run). A full run starts fresh; a re-run + # (--accumulate, which .failed always starts with) merges so already-passed boards + # survive. -bt alone is not a re-run marker. report_dir = Path(os.environ.get('HIL_REPORT_DIR', '.')) - fresh = not (args.skip_board or args.board_test) + failed_fname = report_dir / (config_file.name + '.failed') + fresh = not args.accumulate + # The unlink is DEFERRED to inside the pool try/except below: wiping here leaves + # Manager() and Pool() running with the old report gone and no report-writing path + # armed, so an EAGAIN/ENOMEM on fork gives CI an EMPTY report dir with no reason. + + seed = os.getenv('HIL_SHUFFLE_SEED') or str(int(time.time())) + log_line(f'test-order shuffle seed: {seed} (HIL_SHUFFLE_SEED={seed} to replay); ' + f'flash/usbtest parallel per controller: {hil_lock.FLASH_PARALLEL}/{hil_lock.USBTEST_PARALLEL}; ' + f'enum timeout first/retry: {ENUM_TIMEOUT}/{ENUM_TIMEOUT_RETRY}s; ' + # all three are env-tunable, so a run that dies on the guard is otherwise + # unattributable from the log alone + f'pool guard: {POOL_TIMEOUT}s') + + hints = {} + try: + with CONTROLLER_CACHE.open() as f: + loaded = json.load(f) + # tolerate a hand-edited/torn cache: keep only the expected uid -> dict shape + if isinstance(loaded, dict): + hints = {k: v for k, v in loaded.items() if isinstance(v, dict)} + except (OSError, ValueError): + pass + hints_by_uid = {uid: h['pci'] for uid, h in hints.items() if h.get('pci')} + config_boards = schedule_boards(config_boards, hints_by_uid) + log_line('dispatch order: ' + ', '.join(b['name'] for b in config_boards)) + + # Bound BEFORE the try so the finally can name them whatever failed: Pool() forks, and + # the EAGAIN/ENOMEM the wipe comment below worries about is most likely to come from + # that fork -- after a convoy, where every stranded read holds a thread and an fd. Left + # outside, an OSError there escaped with mgr LIVE and `pool` unbound, so no report was + # written and the interpreter unwound into multiprocessing's unbounded atexit join. + pool = mgr = cmap = None + # Defined before the pool so _abandon_exit always has a value: a raise before + # `err_count = build_err + ...` would turn the containment path into a NameError. + err_count = build_err + # Fail CLOSED: only a shutdown_pool() that actually returned True clears this, and the + # assignment sits at the END of the inner finally, so anything raising before it + # (kill_worker_children, a BrokenPipeError from its print) leaves _abandon_exit armed. + pool_abandoned = True + # BEFORE Manager()/Pool(), not inside the try: hil_ci.sh reuses a persistent REMOTE_DIR + # and scp's the report back unconditionally, so if a fork failure (OSError/EAGAIN right + # after a convoy -- the case this whole block guards) skipped the wipe, the finally's + # _abandon_exit would prepend "HIL run abandoned" to the PREVIOUS run's table and + # publish last night's board results as this run's. Nothing is live yet here, so an + # OSError from the wipe itself just exits with its traceback -- it cannot strand the + # interpreter in multiprocessing's unbounded atexit join, which is what deferring it + # was protecting against. if fresh: report_dir.mkdir(parents=True, exist_ok=True) for f in (REPORT_JSON, REPORT_MD): (report_dir / f).unlink(missing_ok=True) + failed_fname.unlink(missing_ok=True) + try: + mgr = Manager() + cmap = mgr.dict() + initargs = (Lock(), seed, + hil_lock.make_permit_sems(Semaphore, hil_lock.USBTEST_PARALLEL), + hil_lock.make_permit_sems(Semaphore, hil_lock.FLASH_PARALLEL), + cmap, Lock(), hints_by_uid) + # maxtasksperchild=1: the sysfs blindness latch is process-global and permanent + # (no decrement anywhere -- see hil_util.SYSFS_STUCK_MAX), so a worker that goes + # blind on ONE wedged board would report 0/30 and "probe missing" for the 2-3 + # healthy boards it picked up afterwards. A fresh worker per board confines the + # damage to the board that caused it; the extra fork is noise against a + # flash+test cycle. + pool = Pool(processes=os.cpu_count() or 1, initializer=init_worker, + initargs=initargs, maxtasksperchild=1) + # OUTER: encloses the pool block too, not just the reporting below. An exception + # escaping async_ret.get() (a worker exception, a Ctrl-C) runs the pool finally and + # then propagates straight out of main(); with _abandon_exit in a sibling try it + # was never reached. + try: + # imap_unordered, NOT map_async: map_async is all-or-nothing, so a guard expiry + # threw away every board that had already finished -- up to a worker-width of + # completed rig time -- and left the re-run spec unwritten, so CI re-tested all + # ~26 boards to find the one that wedged. Draining as results arrive keeps what + # finished and names only what was still in flight. + it = pool.imap_unordered(test_board, config_boards) + mret = [] + deadline = time.monotonic() + POOL_TIMEOUT + try: + mret = drain_pool(it, config_boards, deadline, out=mret) + except MpTimeoutError as te: + mret = te.finished + stuck = [b['name'] for b in config_boards + if b['name'] not in {r[0] for r in mret}] + # The re-run spec FIRST and before the raise: a fresh run already unlinked + # it, so leaving it unwritten is what made the GitHub re-run repeat the + # whole fleet. Only the boards that never reported go in it. + _write_failed_spec(failed_fname, report_dir, + [(n, 1, [], None, 0) for n in stuck] + + [r for r in mret if r[1] > 0]) + # Then the report, with the rows that DID finish, before anything that can + # block. Then RAISE into the ONE containment path: the inner finally runs + # the ordered sweep (kill_worker_children BEFORE terminate, or a reaped + # worker's flasher reparents out of reach), the outer one os._exit's. + banner = (f'**HIL run abandoned: worker pool timed out after ' + f'{POOL_TIMEOUT}s.** {len(mret)} board(s) below finished and ' + f'are this run\'s; {len(stuck)} never reported and are NOT in ' + f'the table: {", ".join(stuck)}. Re-run covers those.\n') + try: + accumulate_report(mret, report_dir, fresh, '', + health_banner + _blind_note(mret) + + _stray_note(mret) + banner) + except Exception as rerr: # noqa: BLE001 - the raise below must still happen + # FALL BACK, do not just warn: accumulate_report can raise on an + # unwritable/root-owned report dir or a torn JSON, and _abandon_exit + # only PREPENDS to a report that exists. Without this the artifact + # upload finds nothing (if-no-files-found: ignore) and the sticky PR + # comment keeps the previous push's green table under a red job. + print(f'warning: partial report failed: {type(rerr).__name__}: {rerr}; ' + f'falling back to the board list', flush=True) + try: + hil_health.write_timeout_report( + report_dir, [b for b in config_boards + if b['name'] in stuck], POOL_TIMEOUT, REPORT_MD, + prefix=health_banner) + except Exception as re2: # noqa: BLE001 + print(f'warning: fallback report failed too: ' + f'{type(re2).__name__}: {re2}', flush=True) + _p(f'HIL worker pool timed out after {POOL_TIMEOUT}s; sweeping and ' + f'shutting it down (abandoning it if a worker is unkillable)', + flush=True) + raise RuntimeError(f'HIL worker pool timed out after {POOL_TIMEOUT}s') + except Exception as e: + # A worker RAISED -- e.g. a flasher adapter dropping off the bus makes + # get_serial_dev raise in the worker's flash section, which no per-test + # handler guards. Same treatment as the timeout path: the drain means + # `mret` already holds every board that finished, so keep those rows and + # name only the ones still in flight. (Under map_async they were all lost, + # which is what the old banner here claimed.) + done = {r[0] for r in mret} + stuck = [b['name'] for b in config_boards if b['name'] not in done] + _write_failed_spec(failed_fname, report_dir, + [(n, 1, [], None, 0) for n in stuck] + + [r for r in mret if r[1] > 0]) + banner = (f'**HIL run aborted: a worker raised {type(e).__name__}: {e}.** ' + f'{len(mret)} board(s) below finished and are this run\'s; ' + f'{len(stuck)} did not report: {", ".join(stuck)}.\n') + try: + accumulate_report(mret, report_dir, fresh, '', + health_banner + _blind_note(mret) + + _stray_note(mret) + banner) + except Exception as re2: # noqa: BLE001 - the raise below must still happen + print(f'warning: partial report failed: {type(re2).__name__}: {re2}', + flush=True) + raise - with Pool(processes=os.cpu_count() or 1, initializer=init_worker, initargs=(Lock(),)) as pool: - async_ret = pool.map_async(test_board, config_boards) + err_count = build_err + sum(e[1] for e in mret) + _write_failed_spec(failed_fname, report_dir, mret) + finally: + # Not `with Pool(...)`: its __exit__ joins the workers unbounded, hanging on + # any worker in uninterruptible sleep. shutdown_pool bounds the same terminate() + # by a grace period, so the pool is NOT cleanly closed/joined when it returns + # False. Record the outcome but never exit here: the report below is the only + # record of a run that otherwise passed. + # + # Same ordering as the timeout path: what the workers spawned must be + # snapshotted and killed while its parent is alive, or terminate() reparents it + # out of reach. + # + # Both calls must stay guarded: a raise here skips accumulate_report(), so a run + # whose boards ALL passed publishes an empty report dir -- and both can raise + # for reasons unrelated to the results. pool_abandoned stays fail-CLOSED, so + # _abandon_exit still arms. + try: + # Still worth running for the TIMEOUT path, where the workers are + # genuinely stuck mid-task and their children are still reachable through + # the pool's ppid tree. On the normal path every worker has already swept + # its own (kill_own_children) and retired, so this finds nothing. + # + # No banner from here: this finally runs AFTER accumulate_report on both + # abort paths, so anything appended to health_banner now is written to a + # variable nobody reads again. The report gets its count from the result + # tuples instead, via _stray_note. + hil_health.kill_worker_children(pool, mgr) + except Exception as e: + print(f'warning: worker-child sweep failed: {type(e).__name__}: {e}', + flush=True) + try: + pool_abandoned = not hil_health.shutdown_pool(pool) + except Exception as e: + print(f'warning: pool shutdown failed: {type(e).__name__}: {e}', flush=True) + + # refresh controller hints: pci resolved this run, plus durations from full runs + # only (a filtered run would understate the board's real cost) 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') + if PROFILE: + # debug snapshot of the run's live uid->PCI / PCI->slot resolutions + report_dir.mkdir(parents=True, exist_ok=True) + with (report_dir / 'hil_profile_ctrl.json').open('w') as f: + json.dump(dict(cmap), f, indent=1, sort_keys=True) + uid_of = {b['name']: b['uid'] for b in config['boards']} + for name, _, _, _, dur, *_ in mret: + uid = uid_of.get(name) + if uid is None: + continue + h = dict(hints.get(uid) or {}) + h['name'] = name # informational: cache is keyed by uid + h['pci'] = cmap.get(f'uid:{uid}') or h.get('pci') + if dur > 0: # test_board reports 0.0 for filtered (partial) runs + h['duration'] = round(dur, 1) + hints[uid] = h + # merge-on-write: another HIL job (e.g. the esp split) may have finished since + # our startup read, so overlay only this run's boards and replace atomically + merged = {} + try: + with CONTROLLER_CACHE.open() as f: + cur = json.load(f) + if isinstance(cur, dict): + merged = {k: v for k, v in cur.items() if isinstance(v, dict)} + except (OSError, ValueError): + pass + merged.update({uid_of[n]: hints[uid_of[n]] for n, *_ in mret if n in uid_of}) + CONTROLLER_CACHE.parent.mkdir(parents=True, exist_ok=True) + tmp = CONTROLLER_CACHE.with_suffix('.json.tmp') + with tmp.open('w') as f: + json.dump(merged, f, indent=1, sort_keys=True) + tmp.replace(CONTROLLER_CACHE) + except Exception as e: + # Deliberately broad, and it must stay that way: this best-effort refresh makes + # Manager proxy RPCs that raise EOFError / BrokenPipeError / RemoteError when + # the Manager child has died, none of them OSErrors -- an OSError-only guard let + # those skip accumulate_report(). Nothing here is worth the report. + print(f'warning: cannot persist controller hints to {CONTROLLER_CACHE}: ' + f'{type(e).__name__}: {e}') - 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()}') + # board x test result matrix -> hil_report.md (accumulates across re-runs) + stdout. + # -b/-bt means a filtered run (PR selection or a re-run spec): say so, or the report + # looks exactly like a full run that happened to be small + scoped = sorted(set(args.board) | set(board_test)) + scope = f'{len(scoped)} board(s) — {", ".join(scoped)}' if scoped else '' + report = accumulate_report(mret, report_dir, fresh, scope, + health_banner + _blind_note(mret) + + _stray_note(mret)) + 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) + duration = time.time() - duration + print() + print("-" * 30) + print(f'Total failed: {err_count} in {duration:.1f}s') + print("-" * 30) + finally: + # In the finally, not after: any raise above (accumulate_report sits outside the + # OSError handler) would skip the abandon path and unwind into multiprocessing's + # unbounded atexit join, hanging the runner. + _abandon_exit(pool, mgr, pool_abandoned, err_count, report_dir / REPORT_MD) + # Same clamp: exit status is a byte either way, so 256 failures would report green. + sys.exit(min(err_count, 125)) if __name__ == '__main__': |