mirror of
https://github.com/hathach/tinyusb.git
synced 2026-08-18 11:02:16 +00:00
test/hil: usbtest.py runner + HIL integration
Binds the kernel usbtest driver (gadget-zero profile), runs the tier-based battery, auto-recovers kernel-side hangs, and skips cases the host controller cannot run (MosChip MCS9990 EHCI int-OUT). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HeF2gZ1M7GWkz6Av4BpKPg
This commit is contained in:
hathach
@ -32,7 +32,9 @@
|
||||
"name": "lpcxpresso43s67",
|
||||
"uid": "08F000044528BAAA8D858F58C50700F5",
|
||||
"tests": {
|
||||
"device": true, "host": false, "dual": false
|
||||
"device": true, "host": false, "dual": false,
|
||||
"skip": ["device/usbtest"],
|
||||
"comment": "usbtest skipped: ip3511 HS wedges from the first control case (1/30); needs on-rig debugging"
|
||||
},
|
||||
"flasher": {
|
||||
"name": "jlink",
|
||||
|
||||
@ -28,6 +28,8 @@
|
||||
# libmtp9 - pymtp ctypes load (device/mtp); Debian 13 uses libmtp9t64
|
||||
# alsa-utils - arecord (device/audio_test_freertos)
|
||||
# iperf - throughput tests (device/net_lwip_*)
|
||||
# - 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 :
|
||||
@ -68,17 +70,28 @@ STATUS_SKIPPED = "\033[33mSkipped\033[0m"
|
||||
# A missing binary is reported as skipped too.
|
||||
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):
|
||||
super().__init__(msg)
|
||||
self.metric = metric
|
||||
|
||||
|
||||
verbose = False
|
||||
test_only = []
|
||||
board_test = {}
|
||||
build_dir = 'cmake-build'
|
||||
skip_flash = False
|
||||
print_lock = None
|
||||
usbtest_lock = None # serializes the usbtest batteries across the board worker pool
|
||||
|
||||
|
||||
def init_worker(lock):
|
||||
global print_lock
|
||||
def init_worker(lock, ut_lock):
|
||||
global print_lock, usbtest_lock
|
||||
print_lock = lock
|
||||
usbtest_lock = ut_lock
|
||||
|
||||
|
||||
def log_line(msg: str) -> None:
|
||||
@ -144,7 +157,7 @@ class HilConfig(TypedDict):
|
||||
boards: list[Board]
|
||||
|
||||
CMD_TIMEOUT = int(os.getenv('HIL_CMD_TIMEOUT', '180'))
|
||||
POOL_TIMEOUT = int(os.getenv('HIL_POOL_TIMEOUT', '3000'))
|
||||
POOL_TIMEOUT = int(os.getenv('HIL_POOL_TIMEOUT', '4200')) # usbtest batteries are serialized fleet-wide, lengthening the tail
|
||||
SERIAL_READ_TIMEOUT = float(os.getenv('HIL_SERIAL_READ_TIMEOUT', '5'))
|
||||
SERIAL_WRITE_TIMEOUT = float(os.getenv('HIL_SERIAL_WRITE_TIMEOUT', '10'))
|
||||
|
||||
@ -1479,6 +1492,62 @@ def test_device_hid_generic_inout(board):
|
||||
h.close()
|
||||
|
||||
|
||||
def test_device_usbtest(board):
|
||||
# Run the Linux testusb tier-4 battery (test/hil/usbtest.py) against the enumerated cafe:4010
|
||||
# device; surface the pass count in the report cell ("✅ 30/30", or "❌ 29/30" on a partial).
|
||||
uid = board['uid']
|
||||
|
||||
def usbtest_enumerated():
|
||||
# match VID:PID too, not just the serial: right after flashing, the previous example's
|
||||
# enumeration (same serial, different PID) can linger and would fail usbtest.py's lookup
|
||||
for f in glob.glob('/sys/bus/usb/devices/*/serial'):
|
||||
d = os.path.dirname(f)
|
||||
try:
|
||||
if (open(f).read().strip().lower() == uid.lower()
|
||||
and open(os.path.join(d, 'idVendor')).read().strip() == 'cafe'
|
||||
and open(os.path.join(d, 'idProduct')).read().strip() == '4010'):
|
||||
return True
|
||||
except OSError:
|
||||
pass
|
||||
return False
|
||||
|
||||
end = time.time() + ENUM_TIMEOUT
|
||||
while time.time() < end and not usbtest_enumerated():
|
||||
time.sleep(0.2)
|
||||
# settle: right after flashing the enumeration can bounce once (and on dual-port parts like
|
||||
# CH32V307 the other port's stale usbtest node — same serial and PID — lingers a moment);
|
||||
# running testusb into that gap sees the device drop mid-case
|
||||
time.sleep(3)
|
||||
|
||||
# --keep-binding leaves the usbtest dynamic id registered: the cleanup path unbinds every
|
||||
# claimed interface, which has wedged the host xHCI (usb_hcd_alloc_bandwidth) on this rig.
|
||||
# Boards test in a worker pool, but the batteries must run one at a time: each one saturates
|
||||
# the host controller (bulk perf, iso streams, unlink storms), and several at once have
|
||||
# hard-frozen the CI rig (fatal PCIe error on its VFIO-passed xHCI).
|
||||
script = Path(__file__).resolve().parent / 'usbtest.py'
|
||||
cmd = f'python3 "{script}" --serial "{uid}" --json --keep-binding --timeout 60'
|
||||
if usbtest_lock is not None:
|
||||
with usbtest_lock:
|
||||
r = run_cmd(cmd, timeout=200)
|
||||
else:
|
||||
r = run_cmd(cmd, timeout=200)
|
||||
out = cmd_stdout_text(r.stdout)
|
||||
brace = out.find('{')
|
||||
try:
|
||||
data = json.loads(out[brace:])
|
||||
passed, failed = int(data['passed']), int(data['failed'])
|
||||
except (ValueError, KeyError, json.JSONDecodeError):
|
||||
raise AssertionError(f'usbtest did not run: {compact_output(out) or cmd_stdout_text(r.stderr)}')
|
||||
|
||||
skipped = int(data.get('skipped', 0)) # host-controller limitation (see usbtest.py host_broken_cases)
|
||||
total = passed + failed
|
||||
if failed == 0 and total > 0:
|
||||
return f'{REPORT_CELL["pass"]} {passed}/{total}' + (f' +{skipped}skip' if skipped else '')
|
||||
bad = [c.get('num') for c in data.get('cases', []) if c.get('status') not in ('PASS', 'SKIP')]
|
||||
raise TestFail(f'usbtest {passed}/{total} (cases failed: {bad})',
|
||||
metric=f'{REPORT_CELL["fail"]} {passed}/{total}')
|
||||
|
||||
|
||||
# -------------------------------------------------------------
|
||||
# Main
|
||||
# -------------------------------------------------------------
|
||||
@ -1502,6 +1571,7 @@ device_tests = [
|
||||
'device/printer_to_cdc',
|
||||
'device/midi_test',
|
||||
'device/mtp',
|
||||
'device/usbtest', # cafe:4010, unique PID; runs the Linux testusb tier-4 battery via usbtest.py
|
||||
# 'device/net_lwip_webserver', # disabled for PR #3605: USB net iface enum is flaky on the CI HIL host
|
||||
]
|
||||
|
||||
@ -1532,7 +1602,7 @@ def find_firmware(variant: str, example: str):
|
||||
return None
|
||||
|
||||
|
||||
def test_example(board: Board, variant: str, example: str) -> tuple[int, str]:
|
||||
def test_example(board: Board, variant: str, example: str) -> tuple[int, str, str | None]:
|
||||
"""
|
||||
Test example firmware
|
||||
:param board: board dict
|
||||
@ -1592,6 +1662,8 @@ def test_example(board: Board, variant: str, example: str) -> tuple[int, str]:
|
||||
last_detail = compact_output(attempt_out.getvalue())
|
||||
if i == max_retry - 1:
|
||||
err_count += 1
|
||||
# a failing test may still carry a metric to show in its cell (e.g. "❌ 29/30")
|
||||
metric = getattr(e, 'metric', None)
|
||||
msg = f'{test_name} {STATUS_FAILED}: {e}'
|
||||
if last_detail:
|
||||
msg += f' {last_detail}'
|
||||
@ -1756,10 +1828,19 @@ def render_matrix(rows_all: list) -> str:
|
||||
sep = '| ' + '-' * board_w + ' | ' + ' | '.join(':' + '-' * (w - 2) + ':' for w in col_w) + ' |'
|
||||
body = [line(lbl, [cell(cells, c) for c in columns]) for lbl, cells in rows_all]
|
||||
|
||||
# tally run cells (blank/not-run cells are absent from the dicts); a metric string counts as pass
|
||||
failed = sum(v == 'fail' for _, cells in rows_all for v in cells.values())
|
||||
skipped = sum(v == 'skip' for _, cells in rows_all for v in cells.values())
|
||||
passed = sum(v not in ('fail', 'skip') for _, cells in rows_all for v in cells.values())
|
||||
# tally run cells (blank/not-run cells are absent from the dicts). A cell is a bare status
|
||||
# ('pass'/'fail'/'skip') or a metric string that carries its own icon (e.g. "❌ 29/30" is a
|
||||
# fail, "✅ 30/30" / "✅ CDC …" a pass), 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**')
|
||||
|
||||
@ -1872,7 +1953,7 @@ def main() -> None:
|
||||
for f in (REPORT_JSON, REPORT_MD):
|
||||
(report_dir / f).unlink(missing_ok=True)
|
||||
|
||||
with Pool(processes=os.cpu_count() or 1, initializer=init_worker, initargs=(Lock(),)) as pool:
|
||||
with Pool(processes=os.cpu_count() or 1, initializer=init_worker, initargs=(Lock(), Lock())) as pool:
|
||||
async_ret = pool.map_async(test_board, config_boards)
|
||||
try:
|
||||
mret = async_ret.get(timeout=POOL_TIMEOUT)
|
||||
|
||||
@ -494,10 +494,32 @@
|
||||
"args": ""
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "ch32v307v_r1_1v0",
|
||||
"uid": "DE6B3E263B3857CAFFFFFFFF",
|
||||
"toolchain": "riscv-gcc",
|
||||
"variant": [
|
||||
{"name": "ch32v307v_r1_1v0-usbhs", "defines": ["SPEED=high"]},
|
||||
{"name": "ch32v307v_r1_1v0-usbfs", "defines": ["SPEED=full"]}
|
||||
],
|
||||
"tests": {
|
||||
"device": true,
|
||||
"host": false,
|
||||
"dual": false
|
||||
},
|
||||
"flasher": {
|
||||
"name": "openocd_wch",
|
||||
"uid": "BC5DA47360D0",
|
||||
"args": ""
|
||||
}
|
||||
}
|
||||
],
|
||||
"boards-skip": [
|
||||
{
|
||||
"name": "ch582m_evt",
|
||||
"uid": "D443627B5450",
|
||||
"toolchain": "riscv-gcc",
|
||||
"comment": "unplugged: fixture (board + WCH-Link) failed to re-enumerate after the 2026-07-06 rig reboot; replug to re-enable",
|
||||
"tests": {
|
||||
"device": true,
|
||||
"host": false,
|
||||
@ -508,9 +530,21 @@
|
||||
"uid": "7FD88F0604B5",
|
||||
"args": ""
|
||||
}
|
||||
}
|
||||
],
|
||||
"boards-skip": [
|
||||
},
|
||||
{
|
||||
"name": "nrf54lm20dk",
|
||||
"uid": "899C3DE5B0F4D5CA",
|
||||
"tests": {
|
||||
"device": true,
|
||||
"host": false,
|
||||
"dual": false
|
||||
},
|
||||
"flasher": {
|
||||
"name": "jlink",
|
||||
"uid": "1051856258",
|
||||
"args": "-device NRF54LM20A_M33"
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "stm32f769disco",
|
||||
"uid": "21002F000F51363531383437",
|
||||
|
||||
413
test/hil/usbtest.py
Executable file
413
test/hil/usbtest.py
Executable file
@ -0,0 +1,413 @@
|
||||
#!/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
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
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')
|
||||
USB_RECOVER = Path(__file__).resolve().parents[2] / '.claude/skills/usb-recover/scripts/usb_recover.sh'
|
||||
PATTERN_PARAM = Path('/sys/module/usbtest/parameters/pattern')
|
||||
|
||||
# Battery per tier, in run order: control sanity first, then 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):
|
||||
kw.setdefault('capture_output', True)
|
||||
kw.setdefault('text', True)
|
||||
return subprocess.run(cmd, **kw)
|
||||
|
||||
|
||||
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):
|
||||
r = sudo(['tee', str(path)], input=data)
|
||||
if check and r.returncode != 0:
|
||||
sys.exit(f'write "{data}" > {path} failed: {r.stderr.strip()}')
|
||||
return r.returncode == 0
|
||||
|
||||
|
||||
def find_device(serial, first=False):
|
||||
"""Locate the usbtest device in sysfs, return info dict or None."""
|
||||
matches = []
|
||||
for dev in SYS_USB.iterdir():
|
||||
try:
|
||||
if (dev / 'idVendor').read_text().strip() != VID or \
|
||||
(dev / 'idProduct').read_text().strip() != PID:
|
||||
continue
|
||||
dev_serial = (dev / 'serial').read_text().strip()
|
||||
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:
|
||||
return None
|
||||
if len(matches) > 1 and not first:
|
||||
if serial:
|
||||
# Dual-port parts (nanoch32v203 fsdev/usbfs, ch32v307 usbhs/usbfs) briefly enumerate
|
||||
# BOTH ports with the same serial around a variant reflash; picking one arbitrarily
|
||||
# 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 host_broken_cases(dev):
|
||||
"""Cases the DUT's upstream host controller cannot run: {case: reason}.
|
||||
The MosChip MCS9990 (9710:9990) EHCI cannot run interrupt-OUT: its FRINDEX
|
||||
register is buggy silicon (the kernel probes it with "applying MosChip
|
||||
frame-index workaround") and ehci-hcd never keeps the int-OUT QH in the
|
||||
hardware periodic schedule, so every int-OUT URB times out regardless of
|
||||
bInterval/mps/size while the device sits armed. Verified A/B 2026-07-09,
|
||||
same board+hub: EHCI FAIL (QH absent from the debugfs periodic schedule the
|
||||
whole hang), OHCI companion PASS, xHCI fine; int-IN unaffected. Skip with a
|
||||
visible SKIP so the battery self-heals once the DUT tree is back on an xHCI."""
|
||||
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())
|
||||
except (OSError, ValueError):
|
||||
return {}
|
||||
if drv.startswith('ehci') and vid_did == ('0x9710', '0x9990'):
|
||||
return {25: 'host EHCI (MosChip MCS9990) loses interrupt-OUT completions'}
|
||||
return {}
|
||||
|
||||
|
||||
def bind_usbtest(dev):
|
||||
"""Bind the device's interface 0 to the usbtest driver."""
|
||||
if not DRIVER.exists():
|
||||
r = sudo(['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:
|
||||
# bound before the re-registration: that probe captured the OLD dynamic id's capability
|
||||
# profile; unbind once (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 dmesg_tail():
|
||||
r = sudo(['dmesg'])
|
||||
lines = [l for l in r.stdout.splitlines() if 'usbtest' in l]
|
||||
return '\n'.join(lines[-8:])
|
||||
|
||||
|
||||
def pci_addr_of_bus(busnum):
|
||||
"""Return the PCI B:D.F backing a USB bus, or None for a non-PCI (SoC/platform) controller."""
|
||||
m = re.search(r'([0-9a-f]{4}:[0-9a-f]{2}:[0-9a-f]{2}\.[0-9])/usb\d+$',
|
||||
os.path.realpath(f'/sys/bus/usb/devices/usb{int(busnum)}'))
|
||||
return m.group(1) if m else None
|
||||
|
||||
|
||||
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}
|
||||
|
||||
p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
|
||||
try:
|
||||
out, _ = p.communicate(timeout=timeout)
|
||||
except subprocess.TimeoutExpired:
|
||||
p.kill()
|
||||
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 (device stopped responding mid-transfer).
|
||||
# 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')
|
||||
args = p.parse_args()
|
||||
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: right after a flash the enumeration may still be settling, and on dual-port
|
||||
# parts the other port'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)
|
||||
if dev and 'ambiguous' not in dev:
|
||||
break
|
||||
if time.monotonic() > deadline:
|
||||
if dev:
|
||||
sys.exit(f"multiple devices with serial {args.serial}: {', '.join(dev['ambiguous'])} "
|
||||
'— stale enumeration from another port? replug or retry')
|
||||
sys.exit(f'no {VID}:{PID} device' + (f' with serial {args.serial}' if args.serial else ''))
|
||||
time.sleep(0.5)
|
||||
|
||||
# tier drives which cases run; 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)
|
||||
|
||||
results = []
|
||||
unrecovered_hang = False
|
||||
try:
|
||||
bind_usbtest(dev)
|
||||
set_pattern(0) # tier 1 firmware sources zeros; also required by perf cases 27/28
|
||||
|
||||
broken = host_broken_cases(dev)
|
||||
for num in cases:
|
||||
if num in broken:
|
||||
results.append({'num': num, 'name': CASE_NAMES[num], 'status': 'SKIP',
|
||||
'detail': broken[num]})
|
||||
if not args.json:
|
||||
print(f"test {num:2d} {CASE_NAMES[num]:22s} SKIP {broken[num]}")
|
||||
continue
|
||||
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':
|
||||
pci = pci_addr_of_bus(dev['node'].split('/')[-2])
|
||||
if pci:
|
||||
print(f'aborting battery: kernel-side hang, device wedged mid-transfer.\n'
|
||||
f'auto-recovering: sudo {USB_RECOVER} pci-reset {pci} '
|
||||
f'(see .claude/skills/usb-recover)', file=sys.stderr)
|
||||
# FLR frees the D-state ioctl without the device lock; must run BEFORE
|
||||
# any unbind/remove_id, which would deadlock the bus otherwise
|
||||
if sudo([str(USB_RECOVER), 'pci-reset', pci]).returncode != 0:
|
||||
unrecovered_hang = True
|
||||
time.sleep(5) # let the bus re-enumerate before cleanup touches sysfs
|
||||
else:
|
||||
unrecovered_hang = True
|
||||
print('aborting battery: kernel-side hang, and the controller has no PCI address '
|
||||
'for FLR recovery — manual intervention (reboot) required', file=sys.stderr)
|
||||
break
|
||||
# re-resolve: after a mid-battery re-enumeration the devnum (and thus the node
|
||||
# path) changes; keep testing the live node instead of the stale one. Match on the
|
||||
# concrete serial (not args.serial, which may be None) so this can never retarget to
|
||||
# a different device that happens to share the VID:PID.
|
||||
live = find_device(dev['serial'], first=True)
|
||||
if not live:
|
||||
results.append({'num': num, 'status': 'FAIL',
|
||||
'detail': f'device dropped off the bus after case {num}'})
|
||||
break
|
||||
dev = live
|
||||
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 is still stuck in a usbfs ioctl holding the device lock; remove_id/unbind
|
||||
# would join the convoy and deadlock the bus (see usb-recover skill) — leave it be
|
||||
print('skipping cleanup after unrecovered hang: reboot required to release the bus',
|
||||
file=sys.stderr)
|
||||
elif not args.keep_binding:
|
||||
sysfs_write(DRIVER / 'remove_id', f'{VID} {PID}', check=False)
|
||||
# release every claimed interface: other devices sharing the VID:PID (stale example
|
||||
# firmware on a test rig) may have been grabbed on probe and would otherwise 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
|
||||
|
||||
failed = [r for r in results if r['status'] not in ('PASS', 'SKIP')]
|
||||
skipped = [r for r in results if r['status'] == 'SKIP']
|
||||
ran = len(results) - len(skipped)
|
||||
if args.json:
|
||||
print(json.dumps({'serial': dev['serial'], 'speed': dev['speed'], 'tier': tier,
|
||||
'passed': ran - len(failed), 'failed': len(failed),
|
||||
'skipped': len(skipped), 'cases': results}, indent=2))
|
||||
else:
|
||||
note = f", {len(skipped)} skipped (host limitation)" if skipped else ''
|
||||
print(f"{ran - len(failed)}/{ran} passed{note}")
|
||||
for r in failed:
|
||||
print(f" FAILED test {r['num']}: {r.get('detail', '')}")
|
||||
if r.get('dmesg'):
|
||||
print(' ' + r['dmesg'].replace('\n', '\n '))
|
||||
return len(failed)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user