#!/usr/bin/env python3 """Run the Linux kernel usbtest/testusb battery against a TinyUSB usbtest device. Device firmware: examples/device/usbtest (VID:PID cafe:4010). The firmware advertises its capability tier in bcdDevice low byte; the battery is selected accordingly (see examples/device/usbtest/README.md). Requires: usbtest kernel module (CONFIG_USB_TEST), testusb binary (built from kernel tools/usb/testusb.c), sudo for driver binding + usbfs ioctls. testusb reporting quirks this script works around: - its exit code is always 0 when the device exists: results are parsed from stdout - a case gated off by the driver's capability profile (or an in-kernel parameter check) returns -EOPNOTSUPP, which testusb silently skips: a missing result line means NOT RUN, and is reported as a failure since every case in the selected battery is expected to run. Binding uses the 5-field new_id form referencing Gadget Zero (0525:a4a0) so the dynamic id inherits its capability profile (autoconf + ctrl_out + iso + intr). Never register a plain "vid pid" dynamic id with usbtest: the dynid then has driver_info == 0 and usbtest_probe() dereferences it without a NULL check (kernel oops). autoconf is also what enables bulk endpoint discovery; the capability flags only unlock cases, they don't require the endpoints to exist. """ import argparse import json from contextlib import redirect_stdout import os import pathlib import re import shutil import subprocess import sys import time from pathlib import Path sys.path.append(os.path.dirname(os.path.abspath(__file__))) # PYTHONSAFEPATH drops it VID = 'cafe' PID = '4010' GZ_REF = '0525 a4a0' # copy Gadget Zero's capability profile (ctrl_out+iso+intr) SYS_USB = Path('/sys/bus/usb/devices') DRIVER = Path('/sys/bus/usb/drivers/usbtest') PATTERN_PARAM = Path('/sys/module/usbtest/parameters/pattern') RECOVER_FLASH_TIMEOUT = 90 # bound on the post-hang reflash; typical flash is 10-20s RECOVER_RESET_TIMEOUT = 30 # bound on the post-hang probe reset; ResetTarget measures ~130ms def recovery_steps(flasher_name: str, time_left: float) -> list: """Ordered (kind, bound) recovery attempts that fit in `time_left`. RESET FIRST, reflash second. A probe reset fails the in-flight URB at the source just as a park-flash does, but it is non-destructive -- the firmware under test survives, so the wedge can still be autopsied -- writes no flash, and cannot brick SWD the way a bad park image has on mimxrt1064_evk and max32666fthr (survived a power cycle). Measured 128-129 ms against a full erase+program, and it works on i.MX RT and on DWC2 alike (stm32f407disco, 2026-08-16: `r; g` -> USB disconnect, re-enumerated 325 ms later). The reset also fits budgets a reflash does not: the old gate skipped recovery entirely when RECOVER_FLASH_TIMEOUT did not fit, which left the holder in place for the next job. Whether either worked is decided by wedged_pids(), never by the exit code -- a clean flash only proves the probe wrote the MCU. """ import hil_flash steps = [] reset_fn = getattr(hil_flash, f'reset_{flasher_name.lower()}', None) if getattr(reset_fn, 'no_op', False): reset_fn = None # a stub that returns rc 0 without resetting: do not claim it if reset_fn and time_left >= RECOVER_RESET_TIMEOUT: steps.append(('reset', RECOVER_RESET_TIMEOUT)) if time_left >= RECOVER_FLASH_TIMEOUT: steps.append(('flash', RECOVER_FLASH_TIMEOUT)) return steps HELPER_TIMEOUT = 30 # default bound for sudo helpers (dmesg/modprobe/setpci/tee) # Battery per tier, in run order: control sanity, simple bulk, queued, unaligned, unlink, # halt/toggle, throughput last. TIER_CASES = { 1: [0, 9, 10, 1, 2, 3, 4, 5, 6, 7, 8, 17, 18, 19, 20, 11, 12, 24, 13, 29, 27, 28], 2: [14, 21], 3: [25, 26], 4: [15, 16, 22, 23], } # Per-case testusb parameters (full speed / high speed). All -s/-v values are multiples # of 512 so transfers stay packet-aligned at both speeds: the device streams whole max-size # packets and a non-aligned IN length would babble. 14/21 must never run with defaults # (vary >= length is -EINVAL in the kernel). PARAMS = { 0: ('-c 1', '-c 1'), 9: ('-c 256', '-c 1000'), 10: ('-c 64 -g 16', '-c 256 -g 16'), **{n: ('-c 128 -s 1024 -v 512', '-c 512 -s 1024 -v 512') for n in (1, 2, 3, 4, 17, 18, 19, 20)}, **{n: ('-c 8 -s 1024 -g 8', '-c 32 -s 1024 -g 16') for n in (5, 6, 7, 8)}, **{n: ('-c 64 -s 1024 -g 8', '-c 256 -s 1024 -g 8') for n in (11, 12, 24)}, 13: ('-c 16 -s 512', '-c 64 -s 512'), 29: ('-c 16 -s 512', '-c 64 -s 512'), 27: ('-c 16 -s 1024 -g 32', '-c 128 -s 1024 -g 32'), 28: ('-c 16 -s 1024 -g 32', '-c 128 -s 1024 -g 32'), 14: ('-c 64 -s 512 -v 61', '-c 256 -s 512 -v 61'), 21: ('-c 64 -s 512 -v 61', '-c 256 -s 512 -v 61'), 25: ('-c 32 -s 512', '-c 256 -s 1024'), 26: ('-c 32 -s 512', '-c 256 -s 1024'), **{n: ('-c 16 -s 512 -g 8', '-c 64 -s 1024 -g 8') for n in (15, 16, 22, 23)}, } CASE_NAMES = { 0: 'NOP', 1: 'bulk write', 2: 'bulk read', 3: 'bulk write vary', 4: 'bulk read vary', 5: 'bulk sg write', 6: 'bulk sg read', 7: 'bulk sg write vary', 8: 'bulk sg read vary', 9: 'ch9 subset', 10: 'queued control', 11: 'unlink reads', 12: 'unlink writes', 13: 'ep halt set/clear', 14: 'ctrl_out write/read', 15: 'iso write', 16: 'iso read', 17: 'bulk write unaligned', 18: 'bulk read unaligned', 19: 'bulk write premapped', 20: 'bulk read premapped', 21: 'ctrl_out unaligned', 22: 'iso write unaligned', 23: 'iso read unaligned', 24: 'unlink queued writes', 25: 'int write', 26: 'int read', 27: 'bulk write perf', 28: 'bulk read perf', 29: 'toggle clear', } RE_PASS = re.compile(r'test (\d+),\s*(\d+)\.(\d+) secs') RE_FAIL = re.compile(r'test (\d+) --> (\d+) \((.*)\)') def run(cmd, **kw): # NOT subprocess.run(timeout=): CPython's post-timeout path is an UNBOUNDED wait() that # never returns on a D-state child -- the hang sysfs_write's timeout exists to catch. timeout = kw.pop('timeout', None) data = kw.pop('input', None) # subprocess.run-only kwarg; Popen takes stdin kw.pop('capture_output', None) # ditto: expressed by the PIPEs below kw.setdefault('text', True) kw.setdefault('encoding', 'utf-8') kw.setdefault('errors', 'replace') # strict decode would raise out of _sudo_soft # NO start_new_session: these helpers (dmesg, modprobe, setpci, tee) must stay in our # process group so hil_test's outer killpg reaps them with us. timeout = timeout if timeout is not None else HELPER_TIMEOUT proc = subprocess.Popen(cmd, stdin=subprocess.PIPE if data is not None else None, stdout=subprocess.PIPE, stderr=subprocess.PIPE, **kw) try: out, err = proc.communicate(input=data, timeout=timeout) return subprocess.CompletedProcess(cmd, proc.returncode, out, err) except subprocess.TimeoutExpired: # Under sudo our child is only the wrapper; the root grandchild survives this and # is left for the report and hil_pool_check to name. Close our pipe ends so an # abandoned child costs no fds. try: proc.kill() # same group as us: never killpg, that would kill us too except OSError: pass try: proc.communicate(timeout=5) except subprocess.TimeoutExpired: for pipe in (proc.stdout, proc.stderr, proc.stdin): try: if pipe is not None: pipe.close() except OSError: pass # unkillable: abandon it, the caller reports the timeout raise def sudo(cmd, **kw): if os.geteuid() != 0: cmd = ['sudo', '-n'] + cmd r = run(cmd, **kw) if r.returncode != 0 and 'password is required' in (r.stderr or ''): sys.exit(f'sudo needs a password for: {" ".join(cmd)}\n' 'Run as root, or grant this user NOPASSWD sudo.') return r def sysfs_write(path, data, check=True): # A driver-registry write (new_id/remove_id/bind) blocks in D state when a wedged # device holds its lock: fail fast instead of piling up unkillable writers -- the rig # needs USB recovery first. # # Verified in v6.12.96: unbind_store -> device_driver_detach -> # device_release_driver_internal -> __device_driver_lock (drivers/base/dd.c), which # takes device_lock() -- the UNINTERRUPTIBLE variant, unlike the sysfs read path -- and # ALSO device_lock(parent), because usb_bus_type sets need_parent_lock = true # (drivers/usb/core/driver.c:2048). So one such write against a wedged device blocks # unkillably while holding the HUB's lock: that is the mechanism by which a single # wedged port takes its whole bus down, and why this fails fast instead. try: r = sudo(['tee', str(path)], input=data, timeout=15) except subprocess.TimeoutExpired: sys.exit(f'write "{data}" > {path} blocked >15s: USB subsystem is wedged ' '(a D-state device lock exists). Recover the rig (usb-kernel-recover skill) ' 'before running batteries.') if check and r.returncode != 0: sys.exit(f'write "{data}" > {path} failed: {r.stderr.strip()}') return r.returncode == 0 def _read_sysfs_bounded(path, grace=1.0): """Bounded sysfs attribute read. The value, or None, or hil_util.SYSFS_UNKNOWN. Delegates to hil_util.read_sysfs (imported here, like every helper import in this file) so both properties hold: the strand cap -- find_device re-scans after EVERY case, so a 30-case battery against a wedged peer would otherwise strand dozens of threads and fds -- and UNKNOWN kept distinct from None. Folding UNKNOWN into None made a blinded scan read as "device dropped off the bus", which aborts down a path that skips the HUNG recovery entirely. """ from helper import hil_util return hil_util.read_sysfs(str(path), grace) _DEV_CACHE: dict = {} # serial -> sysname, see find_device def _reread(sysname, serial): """Re-describe an already-resolved device, CONFIRMING its serial. idVendor/idProduct/busnum/devnum/speed are lock-free (sysfs.c:688-705), so they cannot block on a wedged peer -- but every identical board answers them the same, so they prove nothing about identity. `serial` does, at one bounded read: a sysname is a topology path, and after a renumber (controller reset, reboot) it can name a DIFFERENT cafe:4010 board whose verdicts would be filed under this one. Returns None when the serial is gone, mismatched or unconfirmed -- caller falls back to a full scan. """ d = SYS_USB / sysname try: if ((d / 'idVendor').read_text().strip() != VID or (d / 'idProduct').read_text().strip() != PID): return None dev_serial = _read_sysfs_bounded(d / 'serial') if not isinstance(dev_serial, str) or dev_serial.lower() != serial.lower(): return None # gone, mismatched, or unconfirmable -> full scan decides return { 'sysname': sysname, 'serial': dev_serial, 'node': '/dev/bus/usb/%03d/%03d' % (int((d / 'busnum').read_text()), int((d / 'devnum').read_text())), 'speed': (d / 'speed').read_text().strip(), 'tier': int((d / 'bcdDevice').read_text().strip()[-2:], 16), } except (OSError, ValueError): return None def find_device(serial, first=False): """Locate the usbtest device in sysfs, return info dict or None. Cached by serial: this is called after EVERY case, and a full scan pays a bounded but real `serial` read for every cafe:4010 peer on the rig. With another board wedged that cost lands on a HEALTHY battery ~30 times over, truncating it into BUDGET entries. The fast path pays ONE bounded read -- our own device's serial, the only attribute that tells identical boards apart (see _reread). """ if serial: sysname = _DEV_CACHE.get(serial.lower()) if sysname: hit = _reread(sysname, serial) if hit: return hit _DEV_CACHE.pop(serial.lower(), None) matches, inconclusive = [], [] for dev in SYS_USB.iterdir(): try: if (dev / 'idVendor').read_text().strip() != VID or \ (dev / 'idProduct').read_text().strip() != PID: continue # BOUNDED: idVendor/idProduct are cached descriptors, but `serial` is served # under device_lock(), so an unbounded read blocks us in D state on exactly the # DUT whose hang we are here to report, losing every verdict collected so far. dev_serial = _read_sysfs_bounded(dev / 'serial') if dev_serial is not None and not isinstance(dev_serial, str): inconclusive.append(dev.name) # unknown: NOT proof it is not ours continue if dev_serial is None: continue if serial and dev_serial.lower() != serial.lower(): continue matches.append({ 'sysname': dev.name, 'serial': dev_serial, 'node': '/dev/bus/usb/%03d/%03d' % (int((dev / 'busnum').read_text()), int((dev / 'devnum').read_text())), 'speed': (dev / 'speed').read_text().strip(), 'tier': int((dev / 'bcdDevice').read_text().strip()[-2:], 16), }) except (OSError, ValueError): continue if not matches: # "could not tell" is not "gone". The caller aborts the battery on a falsy return # and that path skips the HUNG reflash, so a blinded scan would report the wedge # we exist to recover from as a physical disconnect. return {'inconclusive': inconclusive} if inconclusive else None if serial and len(matches) == 1: _DEV_CACHE[serial.lower()] = matches[0]['sysname'] if len(matches) > 1 and not first: if serial: # Dual-port parts (nanoch32v203, ch32v307) briefly enumerate BOTH ports with # one serial around a variant reflash, and picking one could bind the stale # port -- report ambiguity so the caller retries until it drops. return {'ambiguous': sorted(m['sysname'] for m in matches)} sys.exit(f'multiple {VID}:{PID} devices found, use --serial: ' + ', '.join(m["serial"] for m in matches)) return matches[0] def check_host_compat(dev): """Refuse to run when the DUT's upstream host controller is known-incompatible: the MosChip MCS9990 (9710:9990) outright (buggy FRINDEX silicon: EHCI never schedules int-OUT URBs and mangles unlinked reads - verified A/B 2026-07-09), and the Renesas uPD720201/02 unless it runs firmware >= 2.0.2.6 (see below).""" for attempt in range(3): try: root = Path(f"/sys/bus/usb/devices/usb{int(dev['node'].split('/')[-2])}") drv = (root / '../driver').resolve().name pci = (root / '..').resolve() vid_did = ((pci / 'vendor').read_text().strip(), (pci / 'device').read_text().strip()) break except (OSError, ValueError): # transient sysfs error (racing a re-enumeration): retry so a blip does not # silently pass an incompatible host, then fail open but say so if attempt == 2: print('warning: cannot probe the upstream host controller; ' 'skipping the host compatibility check', file=sys.stderr) return time.sleep(1) if vid_did == ('0x9710', '0x9990'): sys.exit(f'REFUSING to run: DUT is behind a MosChip MCS9990 ({pci.name}), which is ' 'incompatible with usbtest: broken FRINDEX silicon - int-OUT URBs are never ' 'placed in the EHCI periodic schedule and unlinked reads complete as short ' 'transfers (EREMOTEIO). Move the DUT to an xHCI port.') if drv.startswith('xhci') and vid_did in (('0x1912', '0x0014'), ('0x1912', '0x0015')): # The Renesas uPD720201/uPD720202 must run firmware >= 2.0.2.6 (K2026090.mem; # RAM-uploaded, so it reverts to ROM on every power cycle): on ROM firmware its # command ring dies under unlink stress and the hub worker deadlocks holding the # device lock, needing a host power cycle (ch32v307 2026-07-10; ra6m5 test 24, # mimxrt1015 2026-07-11). Both parts expose the FW version at PCI config 0x6c. # Necessary, not sufficient -- batteries have killed the controller on current # firmware too, which per-board skips in the rig config handle. fw = None try: r = _sudo_soft(['setpci', '-s', pci.name, '0x6c.l'], capture_output=True, text=True) if r.returncode == 0: fw = int(r.stdout.strip(), 16) except (OSError, ValueError): pass if fw is None: sys.exit(f'REFUSING to run: cannot read host xHCI Renesas ({pci.name}) firmware ' 'version (setpci missing or not permitted) - usbtest requires verified ' 'firmware >= 0x00202609 (2.0.2.6); on older firmware the command ring ' 'dies under unlink stress. Install pciutils / fix sudo, or load the ' 'firmware and re-check.') if fw < 0x00202609: sys.exit(f'REFUSING to run: host xHCI Renesas ({pci.name}) firmware 0x{fw:08x} ' '< 0x00202609 (2.0.2.6) - its command ring dies under usbtest unlink ' 'stress. Load the latest firmware (K2026090.mem; it is RAM-uploaded and ' 'reverts to ROM on every power cycle).') def bind_usbtest(dev): """Bind the device's interface 0 to the usbtest driver.""" if not DRIVER.exists(): r = _sudo_soft(['modprobe', 'usbtest']) if r.returncode != 0 or not DRIVER.exists(): sys.exit(f'cannot load usbtest module: {r.stderr.strip()}') # always re-register in case a stale dynamic id carries a different profile intf = f'{dev["sysname"]}:1.0' drv = SYS_USB / intf / 'driver' stale_binding = drv.is_symlink() and drv.resolve().name == 'usbtest' sysfs_write(DRIVER / 'remove_id', f'{VID} {PID}', check=False) sysfs_write(DRIVER / 'new_id', f'{VID} {PID} 0 {GZ_REF}') if stale_binding: # it probed against the OLD dynamic id's capability profile; unbind once (the # device is idle here) so the loop below reprobes the fresh one sysfs_write(drv / 'unbind', intf, check=False) deadline = time.monotonic() + 3 while time.monotonic() < deadline: drv = SYS_USB / intf / 'driver' if drv.is_symlink(): if drv.resolve().name == 'usbtest': return # claimed by a foreign driver: steal the interface sysfs_write(drv / 'unbind', intf) sysfs_write(DRIVER / 'bind', intf, check=False) time.sleep(0.2) sys.exit(f'interface {intf} did not bind to usbtest') def set_pattern(value): try: if PATTERN_PARAM.read_text().strip() != str(value): sysfs_write(PATTERN_PARAM, str(value)) except OSError as e: # FileNotFoundError (no param), PermissionError (root-only), ... sys.exit(f'{PATTERN_PARAM} not usable ({e.strerror}): this usbtest module build may lack ' 'the "pattern" param, or it is not readable') def _sudo_soft(cmd, **kw): """sudo() for calls whose failure must never abort the battery: run() re-raises TimeoutExpired, and two of these are evaluated inside run_case's own timeout handler -- a raise there loses the HUNG verdict, the recovery and the JSON report.""" try: return sudo(cmd, **kw) except (OSError, ValueError, subprocess.SubprocessError, SystemExit) as e: # SystemExit too: sudo() sys.exit()s on 'a password is required', unwinding out of # run_case's timeout handler before the HUNG verdict is recorded -- which leaves # unrecovered_hang False and lets the finally run the remove_id/unbind that must # never happen while a D-state device lock is held print(f'{cmd[0]}: {type(e).__name__}: {e}', file=sys.stderr) return subprocess.CompletedProcess(cmd, 1, '', '') def dmesg_tail(): r = _sudo_soft(['dmesg']) lines = [l for l in r.stdout.splitlines() if 'usbtest' in l] return '\n'.join(lines[-8:]) def wedged_pids(devnode): """(pids, complete): pids still in D state on `devnode` after a recovery reflash. Matched by device node rather than by our child's pid because run_case() may wrap testusb in sudo: the Popen pid is then the wrapper and the blocked process is its child. A clean flash only proves the probe wrote the MCU, not that the D-state holder let go -- this is what tells the two apart. FAIL CLOSED. `complete` is False when an entry could be HIDDEN from us, and the caller must then keep treating the hang as unrecovered: the holder is root-owned (run_case uses `sudo -n` whenever the node is not writable) and a hidepid/ProtectProc mount hides exactly that entry. Reporting "no holder" from a scan that could not see it clears unrecovered_hang and lets cleanup run remove_id/unbind against a device whose usbfs lock is still held -- which deadlocks the bus, not just this board. Self-contained: /proc is plain text and this is one pass over it, so importing a helper to do it would only add a failure mode on the recovery path. """ stuck, complete = [], True # A restricted /proc hides other users' entries ENTIRELY -- no entry, so no # PermissionError to catch -- and testusb runs under sudo, so the holder is exactly # what is hidden. Detect the restriction itself rather than its symptom. if os.geteuid() != 0 and not os.access('/proc/1/cmdline', os.R_OK): complete = False for d in pathlib.Path('/proc').glob('[0-9]*'): try: st = (d / 'stat').read_bytes() if st[st.rindex(b')') + 2:st.rindex(b')') + 3] != b'D': continue if devnode.encode() in (d / 'cmdline').read_bytes(): stuck.append(int(d.name)) except PermissionError: complete = False # cannot rule this pid out except (OSError, ValueError): continue # raced with exit return stuck, complete def run_case(num, dev, testusb, quick, timeout): fs_hs = PARAMS[num][0 if dev['speed'] == '12' else 1] if quick: fs_hs = re.sub(r'-c (\d+)', lambda m: f'-c {max(1, int(m.group(1)) // 8)}', fs_hs) cmd = [testusb, '-D', dev['node'], '-t', str(num)] + fs_hs.split() # device nodes are usually opened directly (udev rule); sudo only if not if not os.access(dev['node'], os.W_OK) and os.geteuid() != 0: cmd = ['sudo', '-n'] + cmd result = {'num': num, 'name': CASE_NAMES[num], 'params': fs_hs} # NO start_new_session: testusb must stay in OUR process group so the caller's outer # killpg still reaps it; a sudo-wrapped child is escalated through sudo below instead. # errors='replace': testusb output is not guaranteed UTF-8, and a strict decode would # raise out of here and out of main(), printing no JSON at all (battery '0/30'). p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, encoding='utf-8', errors='replace') try: out, _ = p.communicate(timeout=timeout) except subprocess.TimeoutExpired: # Under sudo we only kill the wrapper; its root-owned testusb keeps the inherited # stdout pipe, so the reap below times out and the overrun is reported as HUNG. # Accepted rather than escalated: the rig's udev rules make the device node # writable, so sudo is the exception, and the harness must never sudo-kill a pid # it cannot prove is its own. try: p.kill() except OSError: pass try: out, _ = p.communicate(timeout=5) except subprocess.TimeoutExpired: # SIGKILL had no effect: the child is in uninterruptible sleep on an in-kernel # usbfs ioctl. Abandon it — waiting or re-signalling can never succeed. result.update(status='HUNG', detail=f'testusb stuck in D state after {timeout}s', dmesg=dmesg_tail()) return result result.update(status='FAIL', detail=f'timeout after {timeout}s', dmesg=dmesg_tail()) return result m = RE_PASS.search(out) if m and int(m.group(1)) == num: secs = float(f'{m.group(2)}.{m.group(3)}') result.update(status='PASS', secs=secs) if num in (27, 28) and secs > 0: opts = dict(zip(fs_hs.split()[::2], fs_hs.split()[1::2])) total = int(opts['-c']) * int(opts['-s']) * int(opts['-g']) result['mbps'] = round(total / secs / 1e6, 2) return result m = RE_FAIL.search(out) if m and int(m.group(1)) == num: result.update(status='FAIL', detail=f'errno {m.group(2)} ({m.group(3)})', dmesg=dmesg_tail()) return result if cmd[0] == 'sudo' and ('password is required' in out or 'a terminal is required' in out): result.update(status='FAIL', detail='sudo needs a password to run testusb: the device node ' 'is not writable') return result # no result line: the kernel returned -EOPNOTSUPP (capability profile or # in-kernel parameter gate) and testusb skipped silently result.update(status='NOTRUN', detail='case gated off: check binding profile/pattern', stderr=out.strip()) return result def main(): p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) p.add_argument('--serial', help='board uid (USB serial string) to select the device') p.add_argument('--tier', type=int, choices=sorted(TIER_CASES), help='override tier (default: from device bcdDevice)') p.add_argument('--tests', help='comma-separated case numbers, overrides tier battery') p.add_argument('--quick', action='store_true', help='divide iteration counts by 8') p.add_argument('--json', action='store_true', help='machine-readable output on stdout') p.add_argument('--keep-binding', action='store_true', help='leave usbtest dynamic id registered') p.add_argument('--testusb', default=None, help='path to testusb binary') p.add_argument('--timeout', type=int, default=120, help='per-case timeout in seconds') p.add_argument('--recover-board', help='board JSON (name + flasher) for the post-hang ' 'reflash recovery; without it a HUNG case leaves the device wedged') p.add_argument('--recover-fw', help='firmware path reflashed by the post-hang recovery') p.add_argument('--outer-timeout', type=int, default=0, help='the caller\'s total bound on this process; a reflash that cannot ' 'finish before it is skipped rather than orphaned mid-flash') p.add_argument('--budget', type=int, default=0, help='stop starting new cases after this many seconds (0 = no limit). ' 'Callers that impose their own outer timeout set this to reserve ' 'the remainder for the post-hang recovery path') args = p.parse_args() t_start = time.monotonic() sys.stdout.reconfigure(line_buffering=True) # per-case results visible when piped/logged testusb = args.testusb or shutil.which('testusb') or os.path.expanduser('~/testusb') if not os.access(testusb, os.X_OK): sys.exit('testusb binary not found: build kernel tools/usb/testusb.c ' 'and install it, or pass --testusb') # retry briefly: after a flash the enumeration may still be settling, and a dual-port # part's stale same-serial node takes a moment to drop off (see find_device) deadline = time.monotonic() + 8 while True: dev = find_device(args.serial) # find_device is THREE-valued: a device, {'ambiguous': [...]}, or # {'inconclusive': [...]} when bounded reads could not rule a device out. Screening # only for 'ambiguous' let the inconclusive marker through as if it were a device, # and the next statement subscripts dev['tier'] -> KeyError, no JSON on stdout, and # hil_test reports "usbtest did not run / 0-30" for a merely-unreadable bus. if dev and not ({'ambiguous', 'inconclusive'} & dev.keys()): break if time.monotonic() > deadline: if dev and 'ambiguous' in dev: sys.exit(f"multiple devices with serial {args.serial}: {', '.join(dev['ambiguous'])} " '— stale enumeration from another port? replug or retry') if dev and 'inconclusive' in dev: from helper import hil_util as _hu sys.exit(f"cannot tell whether {VID}:{PID} is present: bounded sysfs reads " f"did not answer for {', '.join(dev['inconclusive'])}" f"{_hu.sysfs_blind_note()}") sys.exit(f'no {VID}:{PID} device' + (f' with serial {args.serial}' if args.serial else '')) time.sleep(0.5) # a stale/foreign device advertising an out-of-range tier must not silently run an # empty battery ('0/0 passed' would read as green in CI) tier = args.tier or dev['tier'] if not 1 <= tier <= max(TIER_CASES): sys.exit(f"device advertises tier {tier} (bcdDevice ...{tier:02x}); reflash a usbtest build " f"or pass --tier 1..{max(TIER_CASES)} — refusing to run an unknown/empty battery") if args.tests: cases = [] for tok in args.tests.split(','): tok = tok.strip() if not tok.isdecimal() or int(tok) not in PARAMS: # isdecimal rejects unicode digits sys.exit(f'--tests: {tok!r} is not a known case number (valid 0..{max(PARAMS)})') cases.append(int(tok)) else: cases = [n for t in range(1, tier + 1) for n in TIER_CASES[t]] info = f"device {dev['serial']} {dev['node']} speed={dev['speed']} tier={tier}" if not args.json: print(info) # before touching the device: an incompatible host exits here, before any bind check_host_compat(dev) results = [] unrecovered_hang = False try: bind_usbtest(dev) set_pattern(0) # tier 1 firmware sources zeros; also required by perf cases 27/28 abort_reason = None # set on any early exit; drives the BUDGET back-fill below for idx, num in enumerate(cases): # Only a HUNG case aborts the battery; an ordinary case timeout is a FAIL and # the loop continues, each burning --timeout+5s, so without this the run can # still be in the case loop when the outer timeout SIGKILLs it before it emits # JSON. Checked before dispatch: worst overshoot is one case. if args.budget and time.monotonic() - t_start > args.budget: abort_reason = f'battery budget {args.budget}s exhausted' break results.append(run_case(num, dev, testusb, args.quick, args.timeout)) r = results[-1] if not args.json: extra = f" {r.get('secs', '')}s" if r['status'] == 'PASS' else f" {r.get('detail', '')}" extra += f" {r['mbps']} MB/s" if 'mbps' in r else '' print(f"test {num:2d} {r['name']:22s} {r['status']:6s}{extra}") if r['status'] == 'HUNG': abort_reason = 'battery aborted on a kernel-side hang' # Reflash, NEVER a root-port cycle: resetting the MCU through the DUT's own # debug probe fails the in-flight URB at the source, so the ioctl returns, # the queued kill lands and the cleanup below is lock-safe -- and it reaches # exactly one board, where a root-port cycle bounces every fixture under the # port (and could never remove power anyway; see usb-kernel-recover). # Deliberately not gated on a hub-worker check: our own stuck testusb is # what drives a hub worker into usb_lock_device(), so a pre-check reads # wedged by construction. # # Assume unrecovered until proven otherwise, so any early exit from this # block reaches the finally with the flag set instead of running the # remove_id/unbind that must not happen while a device lock is held. unrecovered_hang = True print('aborting battery: kernel-side hang, device wedged mid-transfer', file=sys.stderr) if not (args.recover_board and args.recover_fw): print('no --recover-board/--recover-fw: the device stays wedged and ' 'cleanup is skipped', file=sys.stderr) break # The reflash is bounded to RECOVER_FLASH_TIMEOUT and skipped when the # caller's outer bound cannot contain it: the flasher runs in its own # session, so an outer killpg mid-flash would ORPHAN it on the probe. Gate # each step on the time actually LEFT -- reserving for the worst case up # front skipped recovery for nearly every real hang, since the hang-prone # cases run late in the tier order. def _time_left(): if not args.outer_timeout: return float('inf') # what still runs after a step: run_cmd's post-kill reap (10s), # the settle (5s), the sudo-escalated descendant reap run_case may # have just paid (up to 7s) and the JSON write return args.outer_timeout - (time.monotonic() - t_start) - 35 if _time_left() < RECOVER_RESET_TIMEOUT: print('insufficient time before the outer bound for even a bounded ' 'reset; the device stays wedged and cleanup is skipped', file=sys.stderr) break try: board = json.loads(args.recover_board) bname, fname = board['name'], board['flasher']['name'] import hil_flash # deferred: stdlib-only unless recovery actually runs flash_fn = getattr(hil_flash, f'flash_{fname.lower()}') reset_fn = getattr(hil_flash, f'reset_{fname.lower()}', None) except Exception as e: # malformed/short json, import failure, unknown flasher print(f'reflash recovery unavailable ({e})', file=sys.stderr) break # DELIVERY must be convoy-safe or the recovery makes things worse: our own # testusb is D-state on this DUT's node, so a flasher that enumerates by # OPENING usbfs nodes blocks on it, survives SIGKILL and is abandoned -- # a SECOND stray, the budget spent, the device still wedged. On 2026-08-12 # a vid_pid-pinned openocd was the only flasher that still reached its # probe; JLinkExe's ShowEmuList returned zero. See hil_flash.convoy_safe. if not hil_flash.convoy_safe(board['flasher']): print(f'{fname} is not convoy-safe for delivery (it enumerates by ' f'opening usbfs nodes, and this DUT has a D-state holder on ' f'its own node): skipping the reflash rather than adding a ' f'second stray. Pin the roster entry with vid_pid on an ' f'openocd flasher to enable recovery for this board.', file=sys.stderr) break # RESET FIRST (see recovery_steps). Non-destructive, ~130 ms, and it # clears the wedge by the same mechanism as the reflash. wedged_pids is the # arbiter: reset_esptool is a stub that returns rc 0 without resetting # anything, so an exit code here proves nothing. steps = recovery_steps(fname, _time_left()) if reset_fn and any(k == 'reset' for k, _ in steps): print(f'auto-recovering: resetting {bname} via {fname} probe ' f'(non-destructive; reflash only if this does not clear it)', file=sys.stderr) try: with redirect_stdout(sys.stderr): reset_fn(board, timeout=RECOVER_RESET_TIMEOUT) except TypeError: with redirect_stdout(sys.stderr): reset_fn(board) # older primitives take no bound except Exception as e: print(f'probe reset raised: {e}; falling through to the reflash', file=sys.stderr) time.sleep(5) # let the freed ioctl unwind stuck, complete = wedged_pids(dev['node']) if complete and not stuck: print('probe reset cleared the wedge; skipping the reflash ' '(firmware under test left intact for autopsy)', file=sys.stderr) unrecovered_hang = False break if _time_left() < RECOVER_FLASH_TIMEOUT: print('reset did not clear it and no budget left for a reflash; ' 'the device stays wedged', file=sys.stderr) break print(f'auto-recovering: reflashing {bname} via ' f'{fname} (see .claude/skills/usb-kernel-recover). ' f'Unbudgeted by flash_permit, like the root-cycle it replaced: the ' f'per-controller semaphores live in hil_test\'s process.', file=sys.stderr) # run_cmd bounds the flash; its banners go to stdout, which in --json mode # carries the result object -- keep them off it. A raising flasher (missing # serial node, unwritable CWD) must not cost the battery its JSON report. try: with redirect_stdout(sys.stderr): ret = flash_fn(board, args.recover_fw, timeout=RECOVER_FLASH_TIMEOUT) except Exception as e: print(f'reflash raised: {e}; the device may still be wedged', file=sys.stderr) break if ret.returncode != 0: # a wedged RP DAP answers nothing and the probe has no reset line; # POR it via the Rescue DP and retry once, exactly as the normal # flash path does (no-op for every other board/failure) out_txt = ret.stdout if isinstance(ret.stdout, str) else '' # inside the redirect like its siblings (hil_test slices the result # object from the first '{' on stdout), and only if a POR + retry # still fits before the outer kill rescued = False try: if _time_left() >= 2 * RECOVER_FLASH_TIMEOUT: with redirect_stdout(sys.stderr): rescued = hil_flash.rescue_openocd( board, out_txt, timeout=RECOVER_FLASH_TIMEOUT) if rescued: print('DAP wedged; rescued via Rescue DP, retrying reflash', file=sys.stderr) with redirect_stdout(sys.stderr): ret = flash_fn(board, args.recover_fw, timeout=RECOVER_FLASH_TIMEOUT) except Exception as e: # guarded like the first flash: a raise here would unwind past the # BUDGET back-fill and the JSON print print(f'rescue/retry raised: {e}', file=sys.stderr) if ret.returncode != 0: print(f'reflash failed (rc {ret.returncode}); the device may still ' f'be wedged', file=sys.stderr) # settle even on a non-zero exit: the reset may have landed before the # flasher failed, and the freed ioctl needs a moment to unwind before # wedged_pids samples time.sleep(5) # Authoritative either way: a clean flash only proves the probe wrote the # MCU, not that the D-state holder let go. stuck, complete = wedged_pids(dev['node']) if stuck: print(f'{dev["sysname"]}: pid(s) {stuck} still in D state on ' f'{dev["node"]} — the device lock was never released', file=sys.stderr) # No hub-worker verdict here: our own testusb still holds the DUT's # device lock, which is what drives a hub worker into usb_lock_device() # -- any verdict from here is confounded by construction. elif not complete: print('cannot confirm recovery: /proc is only partly readable, so a ' 'hidden D-state holder cannot be ruled out', file=sys.stderr) else: unrecovered_hang = False break # re-resolve: a mid-battery re-enumeration changes the devnum and so the node # path. Match on the concrete serial (not args.serial, which may be None) so # this can never retarget to another device sharing the VID:PID. # first=False: the ambiguity guard exists because ONE serial can match two # sysfs nodes on the dual-port WCH parts, and `dev = live` below makes any # mistake stick for the rest of the battery -- including wedged_pids() then # scanning the wrong node and clearing unrecovered_hang on a device it never # checked. Ambiguous comes back as {'ambiguous': [...]}, handled below. live = find_device(dev['serial']) if live and live.get('ambiguous'): # two nodes now answer to one serial (the dual-port WCH parts do this # around a re-enumeration). Picking either would file the rest of the # battery's verdicts under a device we cannot identify, so stop here and # keep the recovery in play rather than guess. abort_reason = (f'serial {dev["serial"]} matches more than one device ' f'({", ".join(live["ambiguous"])}) after case {num}') unrecovered_hang = True break if live and live.get('inconclusive'): # bounded reads stopped answering, so we cannot say the device left -- # treat it as the wedge it probably is, which keeps the HUNG reflash and # the lock-safe cleanup in play from helper import hil_util as _hu abort_reason = ('cannot tell whether the device is still present: bounded ' 'sysfs reads stopped answering' + _hu.sysfs_blind_note()) unrecovered_hang = True break if not live: # no second entry for `num`: run_case already recorded it, and a duplicate # inflates the denominator (31/30) and reports a PASSing case as failed abort_reason = f'device dropped off the bus after case {num}' break dev = live if abort_reason and all(c in {r['num'] for r in results} for c in cases) \ and 'dropped off the bus' in abort_reason and results: # nothing left to back-fill (the drop happened during/after the LAST case), # so the run would report a clean pass; the case it died on is not a pass if results[-1].get('status') == 'PASS': # only a PASS: a real FAIL/NOTRUN verdict names the actual regression # (errno, dmesg) and must not be overwritten by the drop message results[-1] = dict(results[-1], status='FAIL', detail=abort_reason) if abort_reason: # One BUDGET entry per case never dispatched, on EVERY abort path: a shrunken # denominator (4/5 instead of 4/30) hides that most of the battery never # executed and makes a regression in the skipped range read as "not the # problem". ran = {r['num'] for r in results} results += [{'num': n, 'status': 'BUDGET', 'detail': f'not run: {abort_reason}'} for n in cases if n not in ran] finally: # best-effort cleanup: a sudo/sysfs failure here (sudo() may sys.exit) must not replace # an exception propagating out of the try body with a less useful one try: if unrecovered_hang: # testusb still holds the device lock in a usbfs ioctl: remove_id/unbind # would join the convoy and deadlock the bus (see usb-kernel-recover) print('skipping cleanup after unrecovered hang: ask the operator for a full PVE host ' 'power cycle (a VM reboot is not reliable — hubs latch up across the PCIe reset)', file=sys.stderr) elif not args.keep_binding: sysfs_write(DRIVER / 'remove_id', f'{VID} {PID}', check=False) # release every claimed interface: another device sharing the VID:PID # (stale example firmware) may have been grabbed on probe and would stay # bound to usbtest until re-plugged, hijacking the next test's device for intf in DRIVER.glob('*:*'): sysfs_write(DRIVER / 'unbind', intf.name, check=False) except SystemExit: pass # BUDGET, not NOTRUN: NOTRUN is taken, for a case the KERNEL gated off (-EOPNOTSUPP, # see run_case) -- a real result that must stay in `failed` and keep its case number. # BUDGET keeps the denominator honest without lying about the numerator: naming cases # that never executed as failures sends a maintainer bisecting one of them. notrun = [r for r in results if r['status'] == 'BUDGET'] failed = [r for r in results if r['status'] not in ('PASS', 'BUDGET')] ran = len(results) if args.json: # `wedged` is the verdict this process ALREADY computed; without it the caller had # to infer one from 'HUNG' in our stdout, which misses a recovery that ran and # failed, the inconclusive abort (no case reaches status HUNG), and any battery # killed before it printed. print(json.dumps({'serial': dev['serial'], 'speed': dev['speed'], 'tier': tier, 'passed': ran - len(failed) - len(notrun), 'failed': len(failed), 'notrun': len(notrun), 'wedged': bool(unrecovered_hang), 'cases': results}, indent=2)) else: print(f"{ran - len(failed) - len(notrun)}/{ran} passed" + (f", {len(notrun)} not run" if notrun else "")) for r in failed: print(f" FAILED test {r['num']}: {r.get('detail', '')}") if r.get('dmesg'): print(' ' + r['dmesg'].replace('\n', '\n ')) # NOTRUN counts toward the exit status even though it is reported separately: a # standalone run whose cases were all skipped has NOT passed, and returning 0 hands a # false success to any script driving this directly. return len(failed) + len(notrun) if __name__ == '__main__': sys.exit(main())