usbtest skill: scripts for bring-up and failed cases (#4019) - kernel_src.py prints a testusb case's kernel source and its completion waits for a tag or uname -r - run_case.py runs chosen cases on one roster board: refuses before touching hardware, then lock, flash, run under a budget, park, release; SIGTERM kills its children and still releases the lock - usbtest.py: a case testusb never ran is FAIL, duplicate --tests are refused, an unrecovered hang exits nonzero - hints in CLAUDE.md, the hil skill and docs/porting.rst
diff --git a/.claude/skills/hil/SKILL.md b/.claude/skills/hil/SKILL.md index 1cca161..6ae20c2 100644 --- a/.claude/skills/hil/SKILL.md +++ b/.claude/skills/hil/SKILL.md
@@ -23,6 +23,7 @@ The `ci` rig also hosts a GitHub Actions runner that flashes boards and runs HIL as part of CI. Hardware access is arbitrated **per board** with kernel flocks in `/tmp/tinyusb-hil-locks/` — do NOT stop the runner service. - `hil_test.py` self-locks each board for its flash+test (holder reason `hil_test.py`). A locked board fails immediately (`<board> Failed: board locked: {holder info}`) without flashing — in CI, re-run the failed job later; if your `hold` is refused with reason `hil_test.py`, a CI job is mid-test — wait a few minutes and retry rather than forcing. +- For usbtest bring-up or case diagnosis, load the `usbtest` skill before a manual case run; its `run_case.py` takes and releases the lock itself. This skill still owns board locks and rig selection. - For hardware work outside `hil_test.py` (JLink/GDB, manual flashing, `usbtest.py`, serial poking), hold the lock first and release it when done — release is mandatory cleanup; the auto-release on holder death is a backstop, not the plan: ```bash @@ -288,7 +289,8 @@ already begins with `--accumulate` and restricts each board to its failed tests. A hand-scoped `-b <board>` retry MUST pass `--accumulate` too: a fresh run unlinks the report, replacing the whole-fleet table with a one-row table. A usbtest battery that produced per-case verdicts is not -auto-retried; its result already stands. If a board or fixture stops enumerating, or a tool of +auto-retried; its result already stands, and its FAIL, NOTRUN or HUNG cases go to the `usbtest` +skill for diagnosis. If a board or fixture stops enumerating, or a tool of yours hangs in D state, that is a wedge: save `dmesg | tail -50` as an artifact beside the report (never into the report's rows) and name the board in `wedged`, without recovering it. If a retry is still not enough, an interactive
diff --git a/.claude/skills/usbtest/SKILL.md b/.claude/skills/usbtest/SKILL.md index 36871be..020aa7f 100644 --- a/.claude/skills/usbtest/SKILL.md +++ b/.claude/skills/usbtest/SKILL.md
@@ -1,210 +1,197 @@ --- name: usbtest -description: Use when running, debugging, or porting the Linux usbtest/testusb battery (examples/device/usbtest, cafe:4010) — device "did not bind", SET_CONFIGURATION fails, a case fails with errno 110/32/5/71, toggle-clear/halt/unlink/iso failures, iso packets dropped, or a new MCU/DCD needs the full 30/30 sign-off. Needs a Linux PC as the link's host driving TinyUSB in device role — it exercises the DCD, not the TinyUSB host stack. +description: Use when bringing up examples/device/usbtest (cafe:4010) on a new MCU or DCD, or when a usbtest case fails — a red usbtest cell in a HIL report, testusb errno 110/32/5/71, NOTRUN or HUNG cases, device "did not bind", SET_CONFIGURATION fails, halt/toggle-clear/unlink/interrupt/isochronous failures. Load it before reproducing or debugging the case. A Linux PC hosts the link and drives TinyUSB in device role: it exercises the DCD, not the TinyUSB host stack. --- -# usbtest — porting & debugging the Linux kernel USB battery +# usbtest — bring-up and failed cases of the Linux kernel USB battery -## Overview +`examples/device/usbtest` is the device-side peer of the kernel's `usbtest.ko`/`testusb` +(Gadget Zero source/sink): 30 cases over EP0, bulk, interrupt and isochronous, including halt, +data-toggle and unlink storms. The example README describes every case and tier; the host runner +is `test/hil/usbtest.py`. -`examples/device/usbtest` is the device-side peer of the Linux kernel's `usbtest.ko`/`testusb` -(gadget-zero source/sink protocol): 30 cases over bulk, EP0, interrupt, and isochronous, including -halt, data-toggle, and unlink storms. It is the most adversarial exerciser a DCD gets — every port -so far surfaced at least one real driver bug. Host runner: `test/hil/usbtest.py`; HIL integration -runs it per board and reports `✅ 30/30` cells. +**The battery tests the DCD, not the firmware.** A failed case points at the DCD path it +exercises (map below). Reproduce that one case, root-cause it on hardware before changing +anything (agentrc's `hw-debugger`), one variable at a time. A fix is proven when the case passes +and the full battery still passes across reflash cycles. -**Core principle: the battery is a DCD test, not a firmware test.** When a case fails, suspect the -DCD path it exercises (table below), reproduce that one case, and root-cause on hardware before -changing anything (agentrc's `hw-debugger`). One variable at a time; a fix is proven by -the failing case passing *and* the full battery still at 30/30 across reflash cycles. +## Choose the path -## Run +| Situation | Do | +|--------------------------------------------------|-------------------------------------------------------------------------------------| +| Routine full battery on rig boards | The HIL contract (`hil` skill): `hil_test.py -b <board> -t device/usbtest <config>` | +| Chosen cases on a board you have not taken | `scripts/run_case.py` (below) | +| Chosen cases inside a lock you hold, firmware on | `python3 test/hil/usbtest.py --serial <uid> --tests 13,29 --json` | +| What a case does, or whether its hang can clear | `scripts/kernel_src.py` (below), then read the functions the case calls | +| New MCU or DCD | Bring-up ladder | + +Build first, through the build contract; `--variants` gives each roster variant its own +`cmake-build-<variant>` dir, the one HIL and `run_case.py` flash from. A board not yet in the +roster builds without `--variants` and is flashed by hand until it is added: ```bash -# build through the build contract; descriptor sizes auto-adapt per MCU via the example's -# own src/usb_descriptors.h + src/tusb_config.h. No -D: --variants supplies each roster variant's -# defines into its own cmake-build-<variant> (hil skill, Prerequisites), and an extra one would -# stick to the dirs HIL flashes from. A board off the roster drops --variants. -# board_test is the park firmware hil_test.py flashes after the battery. python3 .claude/skills/build/scripts/check_build.py --board <board> -e device/usbtest -e device/board_test --shared --variants <this host's config> - -# rig board, full battery: the HIL harness self-locks (no pre-hold), flashes through the -# roster's probe, budgets the battery and enables hang recovery where the board allows. -python3 test/hil/hil_test.py -b <board> -t device/usbtest <this host's config> - -# one case by hand: the harness re-parks the board afterwards, so hold it, flash usbtest with -# its probe pinned, wait ~3-5 s for enumeration, run, release. From the board's entry in this -# host's HIL config json: <probe-uid>/<args> are its flasher "uid"/"args", <uid> its own "uid", -# <variant> the variant under test, or the board's name when its entry has no "variant" list. -# A non-jlink flasher: run the command flash_<flasher>() in test/hil/hil_flash.py builds for the -# usbtest image its FLASHER_SUFFIX entry picks. -python3 test/hil/helper/hil_lock.py hold <board> --reason "usbtest case 29" -JLinkExe -USB <probe-uid> <args> -if swd -JTAGConf -1,-1 -speed auto -NoGui 1 -ExitOnError 1 \ - -CommandFile cmake-build/cmake-build-<variant>/device/usbtest/usbtest.jlink -python3 test/hil/usbtest.py --serial <uid> --tests 29 -python3 test/hil/helper/hil_lock.py release <board> ``` -A bench with a single J-Link attached flashes with `ninja -C cmake-build/cmake-build-<variant> -usbtest-jlink`; Espressif boards flash with `idf.py` (CLAUDE.md, ESP-IDF). - -- The run registers `cafe 4010` with the usbtest module once per rig (Gadget Zero's profile) and - leaves the id and the binding in place: an unbind has wedged host xHCIs - (`usb_hcd_alloc_bandwidth`), and the next example enumerates under its own PID. -- CI (`hil_test.py`) additionally passes `--budget` and, when the board's recovery flasher is - convoy-safe and flashing is on, `--recover-board`/`--recover-fw`: a HUNG case (the killed - testusb not reaped within 30 s; a peer's held lock reaps later and counts as a timeout) aborts - the battery and RESETS the DUT through its roster probe (non-destructive, ~130 ms), or reflashes - it where the flasher has no reset (esptool); the child reaping afterwards is what clears - `wedged` (see usb-kernel-recover). Manual runs without those flags leave a HUNG device wedged — - expected; reset or reflash it yourself. -- A non-root caller whose device node is not writable runs testusb under `sudo -n`, and the - child it can kill is only the wrapper. A case is HUNG when the read after the kill still times - out after 5 s; it still attempts the recovery step when one is available, but cannot confirm - the test process was reaped, so `wedged` stays set. -- Always settle a few seconds after flashing — enumeration can bounce once; testusb into the gap sees - the device drop mid-case. -- For manual work on a CI rig: hold the board lock before touching hardware and release it - after — never stop the actions runner. It keeps running; the per-board flock is what arbitrates (see the `hil` skill). - Never start a battery by hand next to a running one: `hil_test.py` budgets 2 concurrent batteries - per host controller (`HIL_USBTEST_PARALLEL`). The width itself is a profiled throughput/bandwidth - trade, not a safety ceiling (the concurrency note above `FLASH_PARALLEL` in `hil_lock.py`) — but - a battery outside the budget is a real hazard: unbudgeted concurrent batteries have hard-frozen - the rig with a fatal PCIe error on a VFIO-passed xHCI, and a marginal DUT port bouncing under - concurrent batteries has killed a uPD720201 outright, which lowering the widths does not fix - (that note records every such death). - -## Repair a wrong profile - -The listing (`/sys/bus/usb/drivers/usbtest/new_id`) shows only `cafe 4010`, never the profile behind -it; a bound interface keeps the profile it was probed with. Symptoms of a wrong one (say the -user-mode profile `0525 a4a4`): cases 14/21, 25/26 and 15/16/22/23 come back NOTRUN while -the bulk cases pass, and `dmesg` names the probe (`Linux gadget zero` is the right one). -Repair is a module reload on a reserved, idle rig; `remove_id` alone leaves every bound -interface on the old profile. - -1. Reserve the whole fleet as the `hil` skill's rig-wide rule says (it owns that command and - its per-host config trap); go on only once it holds, and confirm no `testusb`/`usbtest.py` - is running. -2. `sudo modprobe -r usbtest`, then `sudo modprobe usbtest`. A refused unload means an interface - is still in use: find that holder and wait; never force it. -3. Register the correct entry: the next run of `usbtest.py` or `hil_test.py` does it. -4. Verify `dmesg` shows a `Linux gadget zero` probe for the next enumerated board, then release. - -## Porting ladder — new MCU/DCD to 30/30 - -1. **Tier 1 (bulk)**: set `USBTEST_TIER 1`, get enumeration + cases 0,9,10 (EP0) + 1–8,17–20,27,28 - solid. EP0 correctness first — everything else reports through it. -2. **Tier 2 (ctrl_out 14/21)**, **tier 3 (interrupt 25/26)**, **tier 4 (iso 15/16/22/23)** — raise - the tier only when the layer below is clean; run the *full* battery after each layer. -3. **Fit the endpoints**: tier 4 needs 6 endpoints + EP0. Small parts need per-MCU mps/epbuf - overrides in the example's own `src/usb_descriptors.h` (`USBTEST_INT/ISO_EP_MPS_FS`) and - `src/tusb_config.h` (`CFG_TUD_VENDOR_TX_EPSIZE`) — follow the existing CH32/LPC11 patterns. - Parts that can't fit go in `skip.txt`. -4. **Sign-off = reliability, not one pass**: 3–10 full flash→battery cycles. One 30/30 proves - nothing on a flaky bring-up; deterministic partial counts (e.g. exactly 1-in-8 lost) are a - signature, not noise — chase them. -5. Register the board in `test/hil/tinyusb.json` so the HIL suite runs it. - -## Case → DCD subsystem map - -| Failing case(s) | Exercises | First suspect | -|---|---|---| -| 9, 10 | EP0 control storms | EP0 state machine, ZLP/status stage, control starvation under load | -| 1–8, 17–20, 27, 28 | bulk source/sink, sg, perf | FIFO handling, multi-packet, ZLP tolerance | -| 11, 12, 24 | URB unlink mid-transfer | abort/close paths leaving state half-armed | -| 13 | set/clear halt | stall must kill the transfer; halt on armed IN must flush the TX FIFO | -| **29** | clear-halt on an **armed, un-halted** ep | **the classic**: `dcd_edpt_clear_stall` resets toggle but disarms the queued receive → NAKs forever, errno 110. Fix: reset toggle to DATA0 *and* re-arm/preserve the pending transfer. Found independently on rp2040, fsdev, ch32_usbhs, rusb2 | -| 14, 21 | vendor EP0 write/readback | multi-packet control-OUT chunking, DCP flow control | -| 25, 26 | interrupt src/sink | usually free once bulk works | -| 15, 16, 22, 23 | isochronous | see iso rules below | - -## Iso rules (most-violated contract) - -- **DATA0-only in BOTH directions** at FS — never run bulk-style toggle logic on an iso endpoint - (manual-toggle parts: skip the ISR toggle flip for iso IN *and* the toggle-mismatch drop for iso - OUT). Symptom of violating it: exactly every-other packet lost. -- **No handshake** — iso never NAKs/STALLs; parts with response fields use their "no response" - encoding (e.g. NYET on WCH). -- `dcd_edpt_iso_alloc`/`iso_activate` **must not be stubs returning false** — usbd fails the - interface open and the kernel logs "did not bind"/SET_CONFIG times out. If a DCD refuses iso - "because the hardware can't", **verify against the datasheet — the manual outranks the code - comment** (two "no iso support" claims in this tree were false, incl. a per-endpoint exception - the RM documents for one endpoint number only). -- A multi-packet iso IN submit is legal: the DCD streams it one packet per frame, refilling in the - ISR. Slow cores may need double-buffered iso to make the frame deadline. - -## Debug ladder (escalate in order) - -| errno | Meaning | -|-------|--------------------------------------------------------------| -| 110 | timeout — endpoint NAKing forever / device wedged | -| 32 | EPIPE — unexpected STALL | -| 5 | EIO — iso packet errors (check `dmesg`: "N errors out of M") | -| 71 | EPROTO — device answered wrong / too slow (after HC retries) | - -**Step 0 — read what the case actually does.** The kernel module is ground truth; -the table above is a summary. Do this before theorising, and always before deciding -whether a hung case is recoverable. Fetch the upstream version matching the rig's -kernel (`uname -r`; the distro's own source when its patches matter). The sed reads an -upstream or Debian 13+ (`6.12.48+deb13-amd64`) `uname -r`. An ABI name (Debian ≤12 -`6.1.0-18-amd64`, stock Ubuntu `6.8.0-45-generic`) maps to vx.y, so read the real version from -`/proc/version` (Debian) or `/proc/version_signature` (Ubuntu); a Fedora (`6.14.0-0.rc3…`) or -Ubuntu-mainline (`6.12.0-061200rc3-…`) rc kernel hides its `-rcN`. Name either tag by hand: - ```bash -curl --fail -sSO "https://git.kernel.org/pub/scm/linux/kernel/git/stable/linux.git/plain/drivers/usb/misc/usbtest.c?h=v$(uname -r | sed -E 's/^([0-9.]+(-rc[0-9]+)?).*/\1/; s/\.0(-|$)/\1/')" # run on the rig; x.y.0[-rcN] is tagged vx.y[-rcN] -# case N lives under `case N:` in the kernel's usbtest_do_ioctl() -# (drivers/usb/misc/usbtest.c); kernel tools/usb/testusb.c maps the flags: -# -c = param.iterations, -s = param.length, -g = param.sglen (NOT what they read like) +# lock, flash usbtest from the roster, wait for cafe:4010, run, park on board_test, release +python3 .claude/skills/usbtest/scripts/run_case.py --config <this host's config> --board <board> --tests 13,29 --after park ``` -- **Real traffic and pass criteria.** Case 24 at `-c 256 -s 1024 -g 8` is 256 rounds - of 8 bulk-OUT URBs, unlinking `urbs[num-4]`/`urbs[num-2]` and requiring - `-ECONNRESET` on those two plus normal completion on the other 6 — not the - "256 URBs" the flags suggest. -- **Whether the wait is bounded** — decisive for recovery. `simple_io` uses - `wait_for_completion_timeout` (:481); the unlink paths use a bare - `wait_for_completion` (:1502, :1615). A device stalling there wedges the ioctl in - **D state permanently** — it holds the device lock, so nothing recovers it - (usb-kernel-recover, "The terminal case"). Knowing this first stops you burning - the rig on attempts that cannot work. -- **Which DCD path is implicated**, precisely rather than by category. +`--after leave` keeps usbtest running for a debug session. `--variant` is required when the board +has several. It refuses with exit 2 and the reason before touching hardware, including while any +battery or another `run_case.py` runs on the host; pass `--allow-concurrent` only after checking +no battery shares the board's host controller. A wedged device is never parked. The last stdout +line is its JSON verdict; each of its `cases` carries `usbtest.py`'s `detail`, and `stderr` and +`dmesg` when it captured them. -1. `usbtest.py` per-case output + its captured `dmesg` (`TEST n` markers bracket each case). -2. **usbmon** (`usb-kernel-debug` skill): URB-level ground truth. **It cannot show data toggles or NAKs** — - a toggle desync and a dead endpoint look identical (Submits without Completes); distinguish - device-side with GDB. -3. **Target-side diagnosis**: `target-debug` (`esp-target-debug` for Espressif) on the original failing case. -4. **Cross-check the reference manual** (`read-doc` skill) before changing any register-level code — - per CLAUDE.md, and because comments/assumptions in DCDs have been wrong about hardware caps. -5. Check the vendor's **silicon errata** early for timing/DMA hangs (an unimplemented erratum - workaround caused a case-10 hang on one port). +## Rig hazards -## Traps that pass gcc/desk review but fail elsewhere +- **Batteries are budgeted per host controller.** `hil_test.py` runs at most 2 per controller + (`HIL_USBTEST_PARALLEL`); those permits live inside its process, so a battery started outside it + is not counted. Unbudgeted batteries on one controller have frozen the rig, and a marginal DUT + port bouncing under concurrent batteries has killed a uPD720201 (the note above + `FLASH_PARALLEL` in `hil_lock.py`). Never start a battery outside `hil_test.py` on a controller + another battery is using. +- **A wedged peer stalls every testusb on the host.** `testusb` opens every usbfs node while + scanning, even with `-D` (`tools/usb/testusb.c` find_testdev), and opening a node takes its + device lock. Once any device holds its lock for good, each new case blocks there in D state, + and `usbtest.py` blames the device under test. When testusb runs without `sudo -n`, a hold that + ends within the 30 s watch is a FAIL `timeout after Ns (the kill landed Ns late; a peer held the + node)` that stops the battery; one that outlasts the watch becomes HUNG. Check for other D-state + `testusb` processes on the host before trusting a HUNG verdict on a healthy board. +- **The id and binding stay.** `usbtest.py` registers `cafe 4010` once and never unbinds or + removes it: those writes take the uninterruptible device lock. Recovery of a HUNG case is one + step: a probe reset, or a reflash where the flasher has none (esptool); never a root-port + cycle (`usb-kernel-recover`). Only the killed testusb reaping afterwards clears `wedged`, so + under `sudo -n` (node not writable, the child is the wrapper) it stays set. A manual run + without `--recover-board` leaves a HUNG device wedged. `wedged` is also set with no HUNG case + when the serial matches two devices or cannot be read after a case; stderr names which + (`reported wedged: ...`). -- `TUD_OPT_HIGH_SPEED` is a **compile-time capability, not the live speed**: the FS config - descriptor (and OTHER_SPEED) must use FS-legal sizes (int ≤ 64, iso IN+OUT ≤ 1023 B/frame) even - on HS builds — use separate `_FS`/`_HS` descriptor macros. -- Unused `static inline` helpers: clang `-Wunused-function` and IAR `Pe177` error where gcc stays - quiet → `TU_ATTR_UNUSED`. -- A symbol referenced only inside naked asm is invisible to LTO and gets dropped in `-flto` make - builds → keep a `TU_ATTR_USED` C reference to it. -- Nested USB IRQs on cores with hardware context stacks (QingKe HWSTK): plain - `__attribute__((interrupt))` corrupts the return — use naked handlers relying on the HW stack. -- Dedicated USB RAM budgets (PMA/USB-RAM) differ per part *and* per build system section placement: - check the link map, not just that it builds. +## Failed case -## Red flags — stop and re-examine +**Step 0: read the case in the rig's kernel.** The flags do not mean what they look like, and +whether a hang can clear depends on the wait the case reaches, not on the rig: -- "One pass = done" → run reflash cycles. -- "The DCD comment says the hardware can't" → open the datasheet. -- "usbmon shows no toggle problem" → usbmon can't see toggles. -- "It works on gcc" → clang/IAR/LTO/make still pending. -- "Fixed iso IN" → apply the same exemption to iso OUT (toggle logic is symmetric). -- A clean single-board run does not validate concurrent/fleet behavior — a fleet run puts up to 2 - batteries per host controller (`HIL_USBTEST_PARALLEL`) plus concurrent flashes on the same hub - uplinks, which one board never exercises. -- Reasoning about a case from its name or table row → open `usbtest.c` (step 0). The - flags don't mean what they look like, and recoverability is a property of that - case's wait, not of the rig. +```bash +python3 .claude/skills/usbtest/scripts/kernel_src.py --release "$(ssh <rig> uname -r)" --case 24 +``` + +It prints the `case N:` block of `usbtest_do_ioctl()` and every completion wait with its function. +A Debian release maps only to an upstream candidate, which the distro may have patched. +`testusb -c` is iterations, `-s` length, `-g` sglen, `-v` vary; the runner's per-speed values are +its `PARAMS` table. In v6.12, `test_ctrl_queue` (case 10), `unlink1` (11, 12), `unlink_queued` (24) +and `test_queue` (15, 16, 22, 23, 27, 28) wait with no timeout while holding the device lock, so a +device that stops answering there leaves `testusb` in D state for good; scatter-gather (5-8) runs +under a timer. + +| Verdict or errno | Meaning | +|--------------------|--------------------------------------------------------------------------------------------| +| 110 | Timeout: an endpoint NAKs forever or the device wedged | +| 32 | EPIPE: unexpected STALL | +| 5 | EIO: isochronous packet errors; dmesg says "N errors out of M" | +| 71 | EPROTO: the device answered wrong or too slowly after host retries | +| NOTRUN | testusb opened the device and the kernel skipped the case: profile or parameter gate | +| FAIL "did not run" | testusb never reached the ioctl (open or usage error): read the case's captured `stderr` | +| HUNG | testusb not reaped 35 s after SIGKILL; under `sudo -n` 5 s and unconfirmed (wrapper only) | +| BUDGET | never dispatched: the battery stopped first; its detail names why | + +| Failing case(s) | Exercises | First suspect | +|--------------------|--------------------------------------------|----------------------------------------------------------------------------| +| 9, 10 | EP0 control, queued control | EP0 state machine, ZLP and status stage, control starvation under load | +| 1-8, 17-20, 27, 28 | bulk source/sink, sg, perf | FIFO handling, multi-packet transfers, ZLP tolerance | +| 11, 12, 24 | URB unlink mid-transfer | abort and close paths that leave an endpoint half-armed | +| 13 | set/clear halt | a stall must kill the armed transfer and flush a loaded IN FIFO | +| 29 | clear-halt on an armed, un-halted bulk OUT | toggle reset that also disarms the queued receive: NAKs forever, errno 110 | +| 14, 21 | vendor EP0 write and readback | multi-packet control-OUT data stages | +| 25, 26 | interrupt source/sink | usually clean once bulk works | +| 15, 16, 22, 23 | isochronous | the isochronous rules below | + +Case 29 (`test_toggle_sync`: clear halt, write, clear halt, write) is the most common DCD bug here: +reset the toggle to DATA0 **and** keep the pending transfer armed (fixed that way in rp2040 +ad7acc849, fsdev 046463687, ch32_usbhs d63a45509). + +Escalate in order: +1. The case's `detail` and captured `dmesg` in the JSON verdict (`TEST n` lines bracket each case). +2. usbmon (`usb-kernel-debug`): URB-level truth. It cannot show data toggles or NAKs, so a toggle + desync and a dead endpoint look the same (Submits without Completes); tell them apart on the + target. +3. Target-side evidence: `target-debug` (`esp-target-debug` on Espressif) on the failing case. +4. The reference manual and errata (`read-doc`) before changing any register-level code: DCD + comments have been wrong about what the hardware can do. + +## Isochronous rules + +From USB 2.0 §8.5.5 (Calibre book 775, p. 229-230): +- **No handshake, no retry.** An isochronous endpoint never NAKs or STALLs; parts with a response + field use their "no response" encoding. +- **No toggle sequencing.** A full-speed device sends only DATA0 and should accept DATA0 or DATA1: + skip bulk-style toggle logic on isochronous endpoints in both directions (no toggle flip on IN, + no toggle-mismatch drop on OUT). Symptom of breaking it: every other packet lost. +- **Full-speed sizes:** isochronous up to 1023 bytes per endpoint (§5.6.3), interrupt up to 64 + (§5.7.3). `TUD_OPT_HIGH_SPEED` is a build capability, not the live speed, so the full-speed and + other-speed descriptors need full-speed sizes even on a high-speed build (`_FS`/`_HS` macros in + the example's `src/usb_descriptors.h`). +- On ports that define `TUP_DCD_EDPT_ISO_ALLOC`, `dcd_edpt_iso_alloc`/`dcd_edpt_iso_activate` must + work: a stub returning false fails the vendor interface open, SET_CONFIGURATION is refused and + the runner reports "did not bind". Before accepting "the hardware has no isochronous", check the + reference manual: two such claims in this tree were false. +- A multi-packet isochronous IN submit is legal: the DCD sends one packet per frame and refills in + its ISR. Slow cores may need double buffering to meet the frame deadline. + +## Bring-up ladder + +1. **Tier 1, bulk and EP0** (`USBTEST_TIER 1`): enumeration, then cases 0, 9, 10 first (everything + else reports through EP0), then 1-8, 11-13, 17-20, 24, 27-29 (`TIER_CASES` in `usbtest.py`). +2. **Tier 2** control-OUT (14, 21), **tier 3** interrupt (25, 26), **tier 4** isochronous + (15, 16, 22, 23): raise the tier only when the layer below is clean, and run the full battery + after each. +3. **Fit the endpoints.** Tier 4 needs six endpoints plus EP0. Small parts take per-MCU sizes in + the example's `src/usb_descriptors.h` (`USBTEST_INT_EP_MPS_FS`, `USBTEST_ISO_EP_MPS_FS`) and + `src/tusb_config.h` (`CFG_TUD_VENDOR_TX_EPSIZE`), as the CH32 and LPC11 entries do. A part whose + DCD cannot serve a tier lowers its default `USBTEST_TIER` there, with the reason (RA2A1 is + tier 3); one that cannot fit at all goes in `skip.txt`. +4. **Sign-off is every case of the justified tier, across 3-10 flash-and-run cycles.** The device + advertises its tier in `bcdDevice`, so the battery is 30 cases only at tier 4. One pass proves + nothing on a flaky bring-up; a deterministic partial count (exactly 1 in 8 lost) is a signature + to chase, not noise. +5. Add the board to the HIL roster (`test/hil/*.json`) when roster edits are in the task's scope; + otherwise hand it to the owner. Automated PR repair never touches the rosters. + +## Wrong profile + +`/sys/bus/usb/drivers/usbtest/new_id` lists only `cafe 4010`, never the profile behind it, and a +bound interface keeps the profile it was probed with. The right one probes as `Linux gadget zero` +in dmesg; the user-mode profile (`0525 a4a4`) passes bulk but reports 14, 21, 25, 26 and 15, 16, +22, 23 as NOTRUN. Repair is a module reload on a reserved, idle rig: + +1. Reserve the whole fleet as the `hil` skill's rig-wide rule says, and confirm no `testusb` or + `usbtest.py` is running: `modprobe -r` detaches every bound interface and blocks, unkillably, + behind any case still in flight. It is never refused for an interface in use. +2. `sudo modprobe -r usbtest && sudo modprobe usbtest`. +3. The next `usbtest.py` or `hil_test.py` run registers the id again. +4. Confirm a `Linux gadget zero` probe for the next board, then release. + +## Traps that pass gcc and desk review + +Carried over from earlier ports, not re-verified: +- clang `-Wunused-function` and IAR `Pe177` reject an unused `static inline` that gcc accepts: + `TU_ATTR_UNUSED`. +- A symbol referenced only from naked asm is dropped by `-flto` make builds: keep a + `TU_ATTR_USED` C reference. +- Nested USB IRQs on cores with a hardware context stack (QingKe HWSTK): a plain + `__attribute__((interrupt))` corrupts the return; use naked handlers. +- Dedicated USB RAM (PMA, USB-RAM) budgets differ per part and per build system's section + placement: check the link map, not only that it builds. + +## Red flags + +- "One pass means done": run reflash cycles. +- "The DCD comment says the hardware can't": open the reference manual. +- "usbmon shows no toggle problem": usbmon cannot see toggles. +- "Fixed isochronous IN": apply the same exemption to OUT. +- "It works on gcc": clang, IAR, LTO and make builds are still pending. +- "A clean single-board run": fleet runs put two batteries per controller plus concurrent flashes + on shared hub uplinks. +- Reasoning about a case from its name or a table row: run step 0.
diff --git a/.claude/skills/usbtest/scripts/kernel_src.py b/.claude/skills/usbtest/scripts/kernel_src.py new file mode 100644 index 0000000..b10db2f --- /dev/null +++ b/.claude/skills/usbtest/scripts/kernel_src.py
@@ -0,0 +1,200 @@ +#!/usr/bin/env python3 +# SPDX-License-Identifier: MIT +"""Fetch the Linux usbtest sources for one kernel version and show what a case does. + + kernel_src.py (--tag vX.Y[.Z][-rcN] | --release RELEASE) [--case N] [--cache DIR] + +--release takes `uname -r` of the rig's kernel in one of two forms only: + upstream X.Y.Z or X.Y.0-rcN -> vX.Y.Z, vX.Y (for X.Y.0), vX.Y-rcN + Debian 13+ X.Y.Z+debN-<flavour> -> vX.Y.Z, an UPSTREAM CANDIDATE: Debian may patch + the files; compare its source package to be sure +Anything else (Debian <= 12 and Ubuntu ABI names, Fedora, Ubuntu mainline, local suffixes) is +refused: pass --tag after reading the real version from /proc/version or the distro package. + +Fetches drivers/usb/misc/usbtest.c and tools/usb/testusb.c from the kernel.org stable tree into +the cache (default $XDG_CACHE_HOME/tinyusb-usbtest/<tag>) and prints their paths. --case N also +prints `case N:` of usbtest_do_ioctl() with line numbers, and every completion wait in usbtest.c +with its enclosing function and whether the API takes a timeout. It does not say which waits a +case reaches or whether a hang there recovers: read the functions the case calls. + +Exit 0 ok, 1 fetch or extraction failed, 2 usage error or a refused release. +""" +import argparse +import http.client +import os +import re +import sys +import urllib.error +import urllib.parse +import urllib.request +from pathlib import Path + +BASE_URL = os.environ.get('KERNEL_SRC_BASE_URL', + 'https://git.kernel.org/pub/scm/linux/kernel/git/stable/linux.git/plain') +FILES = ('drivers/usb/misc/usbtest.c', 'tools/usb/testusb.c') +FETCH_TIMEOUT = 60 + +RE_TAG = re.compile(r'v\d[\w.+-]*') # shape only: the fetch decides whether it exists +RE_UPSTREAM = re.compile(r'(\d+)\.(\d+)\.(\d+)(-rc\d+)?') +RE_DEBIAN = re.compile(r'(\d+)\.(\d+)\.(\d+)\+deb(\d+)(-[a-z0-9]+)+') +WAIT_APIS = {'wait_for_completion': 'no timeout', + 'wait_for_completion_timeout': 'timeout', + 'wait_for_completion_interruptible': 'no timeout, interruptible', + 'wait_for_completion_killable': 'no timeout, killable', + 'usb_sg_wait': 'no timeout of its own'} + + +def usage(msg): + print(f'error: {msg}', file=sys.stderr) + sys.exit(2) + + +def fail(msg): + print(f'error: {msg}', file=sys.stderr) + sys.exit(1) + + +def tag_for_release(release): + """(tag, note) for a `uname -r` in a known form, else None.""" + m = RE_UPSTREAM.fullmatch(release) + if m: + major, minor, patch, rc = m.groups() + if rc and patch != '0': + return None # an rc is always cut from X.Y.0 + base = f'v{major}.{minor}' if patch == '0' else f'v{major}.{minor}.{patch}' + return base + (rc or ''), 'upstream' + m = RE_DEBIAN.fullmatch(release) + if m and int(m.group(4)) >= 13: + major, minor, patch = m.group(1), m.group(2), m.group(3) + base = f'v{major}.{minor}' if patch == '0' else f'v{major}.{minor}.{patch}' + return base, f'upstream candidate for Debian {release}: Debian may patch these files' + return None + + +def fetch(tag, cache): + out = cache / tag + try: + out.mkdir(parents=True, exist_ok=True) + except OSError as e: + fail(f'cannot create the cache {out}: {e}') + paths = [] + for rel in FILES: + dest = out / Path(rel).name + if not dest.is_file() or dest.stat().st_size == 0: + url = f'{BASE_URL}/{rel}?h={urllib.parse.quote(tag)}' + # git.kernel.org answers 403 to Python's default User-Agent + req = urllib.request.Request(url, headers={'User-Agent': 'tinyusb-usbtest-kernel-src'}) + try: + with urllib.request.urlopen(req, timeout=FETCH_TIMEOUT) as r: + data = r.read() + except (urllib.error.URLError, http.client.HTTPException, OSError) as e: + fail(f'cannot fetch {rel} at {tag} ({url}): {e}') + if not data: + fail(f'{rel} at {tag} came back empty ({url})') + tmp = dest.with_suffix('.part') + try: + tmp.write_bytes(data) + tmp.replace(dest) + except OSError as e: + fail(f'cannot write {dest}: {e}') + paths.append(dest) + return paths + + +def function_span(lines, name): + """(first, last) 0-based line indexes of the definition of `name`, or None.""" + starts = [i for i, l in enumerate(lines) + if re.match(rf'(\w[\w\s\*]*\s)?\**{name}\s*\(', l) and not l.rstrip().endswith(';')] + if len(starts) != 1: + return None + for j in range(starts[0], len(lines)): + if lines[j] == '}': + return starts[0], j + return None + + +def case_block(lines, num): + """Numbered lines of `case num:` in usbtest_do_ioctl(); SystemExit on anything ambiguous.""" + span = function_span(lines, 'usbtest_do_ioctl') + if span is None: + fail('usbtest_do_ioctl() not found exactly once in usbtest.c') + first, last = span + # the ioctl's own switch labels sit at one tab; nested switches are indented deeper + labels = [i for i in range(first, last) if re.match(r'\tcase \d+:', lines[i])] + hits = [i for i in labels if re.match(rf'\tcase {num}:', lines[i])] + if len(hits) != 1: + fail(f'case {num}: found {len(hits)} top-level labels in usbtest_do_ioctl(), expected 1') + start = hits[0] + # fall-through labels share the body below them: the block ends at the first label that + # follows some body line, else at the switch's closing brace + def is_code(j): + text = lines[j].strip() + return text and j not in labels and not text.startswith(('/*', '*', '//')) + end = next((i for i in labels if i > start and any(map(is_code, range(start + 1, i)))), None) + if end is None: + end = next(i for i in range(start + 1, last + 1) if lines[i] == '\t}') + return [(i + 1, lines[i]) for i in range(start, end)] + + +def wait_sites(lines): + """(line, api, note, function) for every completion wait in the file.""" + # a definition header starts at column 0 and names the function right before its '(' -- + # also the kernel's split style, where the name opens its own line after the return type + funcs = [(i, m.group(1)) for i, l in enumerate(lines) + if l[:1].isalpha() and not l.rstrip().endswith(';') + and (m := re.search(r'(\w+)\s*\(', l))] + sites = [] + api_re = re.compile(r'\b(' + '|'.join(sorted(WAIT_APIS, key=len, reverse=True)) + r')\s*\(') + for i, l in enumerate(lines): + m = api_re.search(l) + if not m or l.lstrip().startswith(('*', '/*', '//')): + continue + owner = [name for start, name in funcs if start <= i] + sites.append((i + 1, m.group(1), WAIT_APIS[m.group(1)], owner[-1] if owner else '?')) + return sites + + +def main(): + p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + g = p.add_mutually_exclusive_group(required=True) + g.add_argument('--tag', help='kernel.org tag, e.g. v6.12.107 or v6.13-rc3') + g.add_argument('--release', help='`uname -r` of the kernel under test') + p.add_argument('--case', type=int, help='print this usbtest case and the wait sites') + p.add_argument('--cache', type=Path, + default=Path(os.environ.get('XDG_CACHE_HOME') or Path.home() / '.cache') / 'tinyusb-usbtest', + help='where fetched files are kept (default: %(default)s)') + args = p.parse_args() + + if args.tag: + if not RE_TAG.fullmatch(args.tag): + usage(f'--tag {args.tag!r} is not a kernel.org tag (vX.Y, vX.Y.Z or vX.Y-rcN)') + tag, note = args.tag, 'given tag' + else: + resolved = tag_for_release(args.release) + if resolved is None: + usage(f'--release {args.release!r} is not an upstream or Debian 13+ release name; ' + 'read the real version (/proc/version, /proc/version_signature or the ' + 'distro package) and pass --tag') + tag, note = resolved + if args.case is not None and not 0 <= args.case <= 99: + usage(f'--case {args.case} is out of range') + + usbtest_c, testusb_c = fetch(tag, args.cache) + print(f'tag {tag} ({note})') + print(f'usbtest.c {usbtest_c}') + print(f'testusb.c {testusb_c}') + if args.case is None: + return 0 + + lines = usbtest_c.read_text(errors='replace').splitlines() + print(f'\n== case {args.case} in usbtest_do_ioctl()') + for n, text in case_block(lines, args.case): + print(f'{n:5d} {text}') + print('\n== completion waits in usbtest.c (API only; read what the case calls)') + for n, api, what, func in wait_sites(lines): + print(f'{n:5d} {func}: {api} ({what})') + return 0 + + +if __name__ == '__main__': + sys.exit(main())
diff --git a/.claude/skills/usbtest/scripts/run_case.py b/.claude/skills/usbtest/scripts/run_case.py new file mode 100644 index 0000000..24971ad --- /dev/null +++ b/.claude/skills/usbtest/scripts/run_case.py
@@ -0,0 +1,385 @@ +#!/usr/bin/env python3 +# SPDX-License-Identifier: MIT +"""Run chosen usbtest cases on one roster board: lock, flash, enumerate, run, park or leave, release. + + run_case.py --config CONFIG --board BOARD --tests N[,N...] --after park|leave + [--variant NAME] [--timeout S] [--allow-concurrent] + +For a session that has NOT taken the board. Inside a lock you already hold, with the firmware +already flashed, run `test/hil/usbtest.py --serial <uid> --tests N --json` directly instead. + +Refuses before touching hardware (exit 2) when: the board or variant is unknown or ambiguous, +the board sits in boards-skip, does not run device/usbtest or lacks a uid or flasher name in the +roster, the usbtest (or, for --after park, the board_test) firmware is not built, the board lock +is held or unusable, or a testusb, usbtest.py, hil_test.py or other run_case.py process is alive +on this host, checked before and again after taking the lock. That check is host-wide and does +not see a hil_test.py started after it, nor join hil_test.py's per-controller battery permits: +--allow-concurrent skips it only once you have established that no battery shares this board's +host controller. + +Flashes with the roster's flasher from cmake-build/cmake-build-<variant> (the build skill's +--shared --variants layout), waits for cafe:4010 with the board's serial, and runs usbtest.py +with post-hang recovery when the board's recovery flasher can deliver it. --after park flashes +device/board_test afterwards unless the device is wedged or the verdict is incomplete. + +stdout ends with one JSON line: {"pass", "board", "variant", "tests", "cases", "wedged", +"boardState", "error"}, each case with usbtest.py's num, name, status, detail and, when it captured +them, stderr and dmesg. Exit 0 every case passed, 1 a case failed or the run failed after the +lock was taken, 2 refused before any hardware action. A SIGTERM (`hil_lock.py release` sends one +to the lock holder) kills the running flasher or usbtest.py, reports boardState unknown with a +verdict line, then releases the lock. +""" +import argparse +import json +import os +import signal +import sys +import tempfile +import time +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[4] +HIL_DIR = ROOT / 'test' / 'hil' +sys.path.insert(0, str(HIL_DIR)) +sys.path.insert(0, str(ROOT / 'tools')) +import ci_select # noqa: E402 +import hil_flash # noqa: E402 +import usbtest # noqa: E402 +from helper import hil_lock, hil_report, hil_util # noqa: E402 + +ENUM_TIMEOUT = 8 # hil_test.py ENUM_TIMEOUT +SETTLE = 3 # hil_test.py USBTEST_SETTLE: enumeration can bounce once after a flash +PEERS = ('testusb', 'usbtest.py', 'hil_test.py', 'run_case.py') +HELPER_S = usbtest.HELPER_TIMEOUT + 5 # a sudo helper at its bound plus usbtest.run()'s reap +# usbtest.py's start before case 1, its steps at their bounds: the 8 s device wait, 3 s of +# host-compat retries, setpci, modprobe, the id-registration lock, the new_id and pattern writes +# (sysfs_write: 15 s tee + 5 s reap) and the 3 s bind. An estimate, not a bound: the battery +# budget starts before it, so a slower start costs cases (reported BUDGET), not a kill. +SETUP_S = 8 + 3 + 2 * HELPER_S + usbtest.REGISTER_LOCK_TIMEOUT + 2 * (15 + 5) + 3 +# each case: its timeout, the 5 s post-kill reap, the dmesg read, and the find_device between cases +CASE_EXTRA_S = 5 + HELPER_S + 5 +PROC = Path('/proc') + + +class Refused(Exception): + """Stop before any hardware action.""" + + +class Terminated(BaseException): + """SIGTERM: a BaseException, like KeyboardInterrupt, so run_cmd kills its child on the way out + and no `except Exception` on the path (flash()'s) mistakes it for a failed step.""" + + +def on_sigterm(signum, frame): + signal.signal(signal.SIGTERM, signal.SIG_IGN) # a second one must not cut the cleanup short + kill_children() + raise Terminated() + + +def kill_children(): + """SIGKILL every child and its session: run_cmd's cleanup misses a child forked before its + try, e.g. when the signal lands inside Popen, which does not kill a child it has started.""" + me = os.getpid() + for d in PROC.glob('[0-9]*'): + try: + if int((d / 'stat').read_text().rsplit(')', 1)[1].split()[1]) != me: + continue + except (OSError, ValueError, IndexError): + continue + pid = int(d.name) # unreaped, so neither this pid nor its group can be reused + for kill in (os.killpg, os.kill): # kill: a child not yet in its own session + try: + kill(pid, signal.SIGKILL) + except OSError: + pass + + +def resolve(config, board_name, variant): + """(board, variant name) from the roster, or Refused naming the alternatives.""" + try: + cfg = json.loads(Path(config).read_text()) + except (OSError, ValueError) as e: + raise Refused(f'cannot read {config}: {e}') + listed = [b['name'] for b in cfg.get('boards', [])] + if listed.count(board_name) > 1: + raise Refused(f'{board_name} appears {listed.count(board_name)} times in {config}') + boards = {b['name']: b for b in cfg.get('boards', [])} + if board_name not in boards: + if any(b['name'] == board_name for b in cfg.get('boards-skip', [])): + raise Refused(f'{board_name} is in boards-skip of {config}') + raise Refused(f'{board_name} is not a board in {config}') + board = boards[board_name] + if 'device/usbtest' not in ci_select.board_tests(board): + raise Refused(f'{board_name} does not run device/usbtest in {config} (its "tests" entry); ' + f'its device port may not reach this host') + flasher = board.get('flasher') + if not (isinstance(board.get('uid'), str) and board['uid'] and isinstance(flasher, dict) + and isinstance(flasher.get('name'), str) and flasher['name']): + raise Refused(f'{board_name} in {config} needs a "uid" and a "flasher" with a "name"') + try: + names = [v['name'] for v in hil_report.board_variants(board)] + except ValueError as e: + raise Refused(f'{board_name}: {e}') + if len(set(names)) != len(names): + raise Refused(f'{board_name} lists a variant name twice: {", ".join(names)}') + if variant is None: + if len(names) > 1: + raise Refused(f'{board_name} has variants {", ".join(names)}: pass --variant') + variant = names[0] + elif variant not in names: + raise Refused(f'{board_name} has no variant {variant}; it has {", ".join(names)}') + return board, variant + + +def lineage(): + """This process and its ancestors: a wrapper such as sudo or timeout names run_case.py too.""" + pids, pid = set(), os.getpid() + while pid > 1 and pid not in pids: + pids.add(pid) + try: + pid = int((PROC / str(pid) / 'stat').read_text().rsplit(')', 1)[1].split()[1]) + except (OSError, ValueError, IndexError): + break + return pids + + +def live_peers(): + """([(pid, name)], complete) for processes that run or drive a battery on this host.""" + found, complete = [], True + own = lineage() + # hidepid hides other users' processes without an error, and testusb may run under sudo: + # detect the restriction itself + if os.geteuid() != 0 and not os.access(PROC / '1' / 'cmdline', os.R_OK): + complete = False + for d in PROC.glob('[0-9]*'): + if int(d.name) in own: + continue + try: + argv = (d / 'cmdline').read_bytes().split(b'\0') + except (FileNotFoundError, ProcessLookupError): + continue # exited while we looked + except OSError: + complete = False # could be a battery we cannot see + continue + hit = argv_peer([a.decode(errors='replace') for a in argv]) + if hit: + found.append((int(d.name), hit)) + return found, complete + + +def argv_peer(argv): + """The PEERS name argv runs: its program past wrappers (sudo, timeout, env, nice) and their + options, numbers and VAR=value, or for python the script past its options; -c/-m run none.""" + rest = iter(argv) + prog = next((a for a in rest if not (os.path.basename(a) in ('sudo', 'timeout', 'env', 'nice') + or a.startswith('-') or '=' in a + or a.rstrip('smhd').replace('.', '', 1).isdigit())), '') + if os.path.basename(prog).rstrip('0123456789.') == 'python': + for a in rest: + if a in ('-c', '-m'): + return None + if a in ('-W', '-X'): # the python options that take a separate value + next(rest, None) + elif not a.startswith('-'): + prog = a + break + else: + return None + name = os.path.basename(prog) + return name if name in PEERS else None + + +def check_peers(): + peers, complete = live_peers() + if peers: + raise Refused('a battery may be running on this host: ' + + ', '.join(f'{name} (pid {pid})' for pid, name in peers)) + if not complete: + raise Refused('cannot read every process on this host, so a running battery ' + 'cannot be ruled out') + + +def firmware(variant, example, flasher): + fw = hil_flash.find_firmware(variant, example, flasher=flasher) + if fw is None: + raise Refused(f'no {example} build for {variant} (flasher {flasher}): build it with the ' + f'build skill\'s --shared --variants <config>') + return str(fw) + + +def flash(board, fw): + try: + ret = hil_flash.flash_primitive(board['flasher']['name'])(board, fw) + except Exception as e: # a raising flasher is a failed flash, not a crash holding the lock + return f'flash raised {type(e).__name__}: {e}' + if ret.returncode != 0: + return f'flash failed (rc {ret.returncode}): {hil_util.cmd_stdout_text(ret.stdout)[-300:]}' + return None + + +def enumerated(uid): + deadline = time.monotonic() + ENUM_TIMEOUT + while time.monotonic() < deadline: + if hil_util.usb_scan(vid_pid=(usbtest.VID, usbtest.PID), serial=uid): + time.sleep(SETTLE) + return True + time.sleep(0.2) + return False + + +def battery(board, fw, tests, timeout): + """(usbtest.py's JSON verdict or None when it printed none, its stderr, a note when the + bound killed it).""" + cmd = [sys.executable, str(HIL_DIR / 'usbtest.py'), '--serial', board['uid'], '--json', + '--tests', ','.join(map(str, tests)), '--timeout', str(timeout)] + rec = hil_flash.recover_flasher(board) + recovery = hil_flash.convoy_safe(rec) + if recovery: + cmd += ['--recover-board', json.dumps({'name': board['name'], 'flasher': rec}), + '--recover-fw', fw] + # usbtest.py stops dispatching cases at its budget, as under hil_test.py, so the kill needs + # room past it only for the one case already started, then usbtest.recovery_reserve() when + # a recovery can run, else the WEDGE_CONFIRM_S watch + budget = SETUP_S + len(tests) * (timeout + CASE_EXTRA_S) + cmd += ['--budget', str(budget)] + bound = budget + timeout + CASE_EXTRA_S + ( + usbtest.recovery_reserve(rec) if recovery else usbtest.WEDGE_CONFIRM_S) + r = hil_util.run_cmd(cmd, timeout=bound, split_stderr=True, quiet=True) + out, err = hil_util.cmd_stdout_text(r.stdout), hil_util.cmd_stdout_text(r.stderr) + if r.returncode == 124: # whatever it printed, a killed battery has no verdict + return None, err, f'usbtest.py killed at its {bound}s bound (rc 124)' + brace = out.find('{') + try: + return (json.loads(out[brace:]) if brace >= 0 else None), err, '' + except ValueError: + return None, err, '' + + +def main(): + p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + p.add_argument('--config', required=True, help="this host's HIL config json (hil skill)") + p.add_argument('--board', required=True) + p.add_argument('--tests', required=True, help='comma-separated usbtest case numbers') + p.add_argument('--after', required=True, choices=('park', 'leave'), + help='park: flash device/board_test afterwards; leave: keep usbtest running') + p.add_argument('--variant', help='roster variant, required when the board has several') + p.add_argument('--timeout', type=int, default=60, help='per-case timeout in seconds (default 60, as HIL)') + p.add_argument('--allow-concurrent', action='store_true', + help='skip the live-battery check; see the description first') + args = p.parse_args() + + report = {'pass': False, 'board': args.board, 'variant': args.variant, 'tests': args.tests, + 'cases': [], 'wedged': False, 'boardState': 'untouched', 'error': ''} + + def finish(code, error=''): + signal.signal(signal.SIGTERM, signal.SIG_IGN) # one verdict line; the release still follows + report['error'] = error + if error: + print(f'error: {error}', file=sys.stderr) + print(json.dumps(report)) + return code + + try: + tests = [int(t) for t in args.tests.split(',')] + except ValueError: + return finish(2, f'--tests {args.tests!r} is not a list of case numbers') + if len(set(tests)) != len(tests) or not all(t in usbtest.PARAMS for t in tests): + return finish(2, f'--tests {args.tests!r}: every case once, each one of {sorted(usbtest.PARAMS)}') + if args.timeout <= 0: + return finish(2, '--timeout must be positive') + + previous = signal.signal(signal.SIGTERM, on_sigterm) + try: + return lock_and_run(args, tests, report, finish) + finally: + signal.signal(signal.SIGTERM, previous) + + +def lock_and_run(args, tests, report, finish): + cwd = os.getcwd() # before the lock: nothing after it may fail outside the cleanup + try: + board, variant = resolve(args.config, args.board, args.variant) + report['variant'] = variant + flasher = board['flasher']['name'] + fw = firmware(variant, 'device/usbtest', flasher) + park_fw = firmware(variant, 'device/board_test', flasher) if args.after == 'park' else None + if not args.allow_concurrent: + check_peers() + try: + lock = hil_lock.acquire_board_lock(board['name'], reason=f'run_case.py usbtest {args.tests}') + except RuntimeError as e: + raise Refused(f'{board["name"]}: {e}') + if lock is None: + raise Refused(f'{board["name"]}: board lock unavailable or bypassed (HIL_NO_BOARD_LOCK)') + except Refused as e: + return finish(2, str(e)) + except Terminated: + return finish(2, 'terminated (SIGTERM) before any hardware action') + + try: + if not args.allow_concurrent: + try: + check_peers() # again: a peer may have started before the lock was taken + except Refused as e: + return finish(2, str(e)) + # flash_jlink writes its command file into the cwd: keep it out of the checkout. Cleanup + # runs after run_locked printed the verdict, so its errors must not add a second one. + with tempfile.TemporaryDirectory(prefix='run_case-', ignore_cleanup_errors=True) as workdir: + os.chdir(workdir) + return run_locked(board, fw, park_fw, tests, args.timeout, report, finish) + except Terminated: + report['pass'] = False + report['boardState'] = 'unknown: terminated (SIGTERM) mid-run' + return finish(1, 'terminated (SIGTERM), e.g. by hil_lock.py release') + except Exception as e: # any failure after the lock still ends with a verdict line + return finish(1, f'{type(e).__name__}: {e}') + finally: + try: + os.chdir(cwd) + finally: + release(lock) + + +def release(lock): + hil_lock.clear_record(lock) + lock.close() + + +def run_locked(board, fw, park_fw, tests, timeout, report, finish): + err = flash(board, fw) + report['boardState'] = 'flash failed' if err else 'usbtest firmware' + if err: + return finish(1, err) + if not enumerated(board['uid']): + return finish(1, f'no {usbtest.VID}:{usbtest.PID} device with serial {board["uid"]} ' + f'{ENUM_TIMEOUT}s after flashing') + data, stderr, killed = battery(board, fw, tests, timeout) + if stderr.strip(): + print(stderr.rstrip(), file=sys.stderr) + if data is None: + report['boardState'] = 'usbtest firmware, verdict incomplete' + return finish(1, killed or 'usbtest.py printed no verdict') + report['cases'] = [{k: c[k] for k in ('num', 'name', 'status', 'detail', 'stderr', 'dmesg') if k in c} + for c in data.get('cases', [])] + report['wedged'] = bool(data.get('wedged')) + report['pass'] = (not report['wedged'] and data.get('failed') == 0 + and data.get('notrun') == 0 and len(report['cases']) == len(tests)) + for c in report['cases']: + print(f"case {c['num']:2d} {c.get('name', ''):22s} {c['status']:6s} {c.get('detail', '')}") + for k in ('stderr', 'dmesg'): + if c.get(k): + print(' ' + c[k].replace('\n', '\n ')) + if report['wedged']: + report['boardState'] = 'wedged: recover it (usb-kernel-recover)' + return finish(1) + if park_fw: + err = flash(board, park_fw) + report['boardState'] = f'park failed: {err}' if err else 'parked on board_test' + if err: + report['pass'] = False + return finish(1, err) + return finish(0 if report['pass'] else 1) + + +if __name__ == '__main__': + sys.exit(main())
diff --git a/.claude/test/test_usbtest_kernel_src.py b/.claude/test/test_usbtest_kernel_src.py new file mode 100644 index 0000000..d1cbaa6 --- /dev/null +++ b/.claude/test/test_usbtest_kernel_src.py
@@ -0,0 +1,269 @@ +"""kernel_src.py against a local HTTP server standing in for git.kernel.org and a trimmed +usbtest.c fixture. Plumbing only: the real fetch is proved by a run against kernel.org.""" +import http.client +import http.server +import importlib.util +import io +import os +import subprocess +import sys +import tempfile +import threading +import unittest +from contextlib import redirect_stderr +from pathlib import Path +from unittest import mock + +REPO = Path(__file__).resolve().parents[2] +SCRIPT = REPO / '.claude' / 'skills' / 'usbtest' / 'scripts' / 'kernel_src.py' +spec = importlib.util.spec_from_file_location('kernel_src', SCRIPT) +kernel_src = importlib.util.module_from_spec(spec) +spec.loader.exec_module(kernel_src) + +# The shapes the parser relies on, as usbtest.c writes them: a split-style header, a nested +# switch whose labels are indented deeper, fall-through labels, and each wait API. +USBTEST_C = '''\ +static int simple_io(struct usbtest_dev *tdev, struct urb *urb) +{ +\tif (!wait_for_completion_timeout(&completion, expire)) { +\t} +} + +static int +test_queue(struct usbtest_dev *dev, struct usbtest_param_32 *param) +{ +\t/* wait_for_completion(&commented) is not a call site */ +\twait_for_completion(&context.done); +} + +static int ch9_postconfig(struct usbtest_dev *dev) +{ +\tswitch (i) { +\t\tcase 5:\t\t/* nested label, not a test case */ +\t\t\tbreak; +\t} +} + +static long +usbtest_do_ioctl(struct usb_interface *intf, struct usbtest_param_32 *param) +{ +\tint\tretval = -EOPNOTSUPP; + +\tswitch (param->test_num) { + +\tcase 0: +\t\tretval = 0; +\t\tbreak; + +\tcase 5: +\t\tswitch (x) { +\t\tcase 7: +\t\t\tbreak; +\t\t} +\t\tretval = perform(dev); +\t\tbreak; +\tcase 15: +\t/* shared with 16 */ +\tcase 16: +\t\tretval = test_queue(dev, param); +\t\tbreak; +\t} +\treturn retval; +} +''' +TESTUSB_C = 'int main(void) { return 0; }\n' + + +class Server: + """Serves <root>/<path> for GET <path>?h=<tag>, counting requests.""" + def __init__(self, root): + self.hits = [] + hits = self.hits + + class Handler(http.server.SimpleHTTPRequestHandler): + def __init__(self, *a, **kw): + super().__init__(*a, directory=str(root), **kw) + + def do_GET(self): + hits.append(self.path) + super().do_GET() + + def log_message(self, *a): + pass + + self.httpd = http.server.ThreadingHTTPServer(('127.0.0.1', 0), Handler) + self.url = f'http://127.0.0.1:{self.httpd.server_address[1]}' + threading.Thread(target=self.httpd.serve_forever, kwargs={'poll_interval': 0.05}, daemon=True).start() + + def close(self): + self.httpd.shutdown() + self.httpd.server_close() + + +class ReleaseToTag(unittest.TestCase): + def test_upstream_forms(self): + for release, tag in [('6.12.107', 'v6.12.107'), ('6.13.0', 'v6.13'), + ('6.12.0-rc3', 'v6.12-rc3')]: + self.assertEqual(kernel_src.tag_for_release(release), (tag, 'upstream'), release) + + def test_debian_13_is_only_a_candidate(self): + for release in ('6.12.107+deb13-amd64', '6.12.107+deb13-cloud-amd64', '6.12.107+deb14-arm64'): + tag, note = kernel_src.tag_for_release(release) + self.assertEqual(tag, 'v6.12.107') + self.assertIn('upstream candidate', note) + + def test_refused_forms(self): + for release in ('6.1.0-18-amd64', # Debian <= 12 ABI name + '6.1.123+deb12-amd64', # Debian 12 with the new suffix + '6.8.0-45-generic', # Ubuntu ABI name + '6.14.0-0.rc3.29.fc42.x86_64', # Fedora rc + '6.12.0-061200rc3-generic', # Ubuntu mainline rc + '6.12.48-rc1', # no rc of a stable point release + '6.12.48-arch1-1', '6.12', ''): + self.assertIsNone(kernel_src.tag_for_release(release), release) + + +class CaseBlock(unittest.TestCase): + lines = USBTEST_C.splitlines() + + def texts(self, num): + return [text.strip() for _, text in kernel_src.case_block(self.lines, num)] + + def test_block_runs_to_the_next_top_level_label(self): + block = self.texts(5) + self.assertEqual(block[0], 'case 5:') + self.assertIn('case 7:', block) # nested label stays inside the block + self.assertEqual(block[-1], 'break;') + self.assertNotIn('case 15:', block) + + def test_fall_through_label_carries_the_shared_body(self): + shared = ['retval = test_queue(dev, param);', 'break;'] + self.assertEqual(self.texts(15), ['case 15:', '/* shared with 16 */', 'case 16:', *shared]) + self.assertEqual(self.texts(16), ['case 16:', *shared]) # last case: up to the switch end + + def test_line_numbers_are_one_based(self): + n, text = kernel_src.case_block(self.lines, 0)[0] + self.assertEqual(self.lines[n - 1], text) + + def test_missing_nested_or_duplicate_label_fails(self): + with self.assertRaises(SystemExit) as e: + kernel_src.case_block(self.lines, 7) # exists only nested + self.assertEqual(e.exception.code, 1) + dup = USBTEST_C.replace('\tcase 0:', '\tcase 0:\n\tcase 0:') + with self.assertRaises(SystemExit): + kernel_src.case_block(dup.splitlines(), 0) + with self.assertRaises(SystemExit): + kernel_src.case_block(USBTEST_C.replace('usbtest_do_ioctl', 'other').splitlines(), 0) + + +class WaitSites(unittest.TestCase): + def test_each_call_with_its_function_and_api(self): + sites = kernel_src.wait_sites(USBTEST_C.splitlines()) + self.assertEqual([(api, what, func) for _, api, what, func in sites], + [('wait_for_completion_timeout', 'timeout', 'simple_io'), + ('wait_for_completion', 'no timeout', 'test_queue')]) + + +class Cli(unittest.TestCase): + def setUp(self): + td = tempfile.TemporaryDirectory() + self.addCleanup(td.cleanup) + self.tmp = Path(td.name) + src = self.tmp / 'srv' + for rel, text in (('drivers/usb/misc/usbtest.c', USBTEST_C), ('tools/usb/testusb.c', TESTUSB_C)): + (src / rel).parent.mkdir(parents=True, exist_ok=True) + (src / rel).write_text(text) + self.src = src + self.server = Server(src) + self.addCleanup(self.server.close) + + def run_script(self, *args): + env = dict(os.environ, KERNEL_SRC_BASE_URL=self.server.url, PYTHONDONTWRITEBYTECODE='1') + return subprocess.run([sys.executable, str(SCRIPT), '--cache', str(self.tmp / 'cache'), *args], + capture_output=True, text=True, env=env, timeout=60) + + def test_fetch_prints_case_and_waits(self): + r = self.run_script('--release', '6.12.107+deb13-amd64', '--case', '16') + self.assertEqual(r.returncode, 0, r.stderr) + self.assertIn('tag v6.12.107 (upstream candidate', r.stdout) + lines = USBTEST_C.splitlines() + first = lines.index('\tcase 16:') + 1 + block = [f'{n:5d} {lines[n - 1]}' for n in range(first, first + 3)] + waits = [' 3 simple_io: wait_for_completion_timeout (timeout)', + ' 11 test_queue: wait_for_completion (no timeout)'] + self.assertEqual(r.stdout.splitlines()[3:], + ['', '== case 16 in usbtest_do_ioctl()', *block, '', + '== completion waits in usbtest.c (API only; read what the case calls)', *waits]) + self.assertEqual((self.tmp / 'cache' / 'v6.12.107' / 'usbtest.c').read_text(), USBTEST_C) + self.assertEqual(sorted(self.server.hits), + ['/drivers/usb/misc/usbtest.c?h=v6.12.107', '/tools/usb/testusb.c?h=v6.12.107']) + + def test_cache_is_reused(self): + self.assertEqual(self.run_script('--tag', 'v6.12.107').returncode, 0) + self.assertEqual(self.run_script('--tag', 'v6.12.107').returncode, 0) + self.assertEqual(len(self.server.hits), 2) + + def test_fetch_failures_exit_1_and_leave_no_file(self): + (self.src / 'tools/usb/testusb.c').unlink() + r = self.run_script('--tag', 'v6.12.107') + self.assertEqual(r.returncode, 1) + self.assertIn('cannot fetch tools/usb/testusb.c', r.stderr) + self.assertFalse((self.tmp / 'cache' / 'v6.12.107' / 'testusb.c').exists()) + (self.src / 'tools/usb/testusb.c').write_text('') + r = self.run_script('--tag', 'v6.12.107') + self.assertEqual(r.returncode, 1) + self.assertIn('came back empty', r.stderr) + + def test_a_truncated_response_exits_1(self): + class Truncated: + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + def read(self): + raise http.client.IncompleteRead(b'part', 100) + + err = io.StringIO() + with mock.patch.object(kernel_src.urllib.request, 'urlopen', lambda req, timeout: Truncated()), \ + redirect_stderr(err), self.assertRaises(SystemExit) as e: + kernel_src.fetch('v6.12.107', self.tmp / 'cache') + self.assertEqual(e.exception.code, 1) + self.assertIn('error: cannot fetch drivers/usb/misc/usbtest.c', err.getvalue()) + + def test_cache_write_errors_exit_1(self): + (self.tmp / 'file').write_text('') + r = self.run_script('--tag', 'v6.12.107', '--cache', str(self.tmp / 'file')) + self.assertEqual(r.returncode, 1, r.stderr) + self.assertIn('error: cannot create the cache', r.stderr) + (self.tmp / 'cache' / 'v6.12.107' / 'usbtest.part').mkdir(parents=True) # write_bytes fails + r = self.run_script('--tag', 'v6.12.107') + self.assertEqual(r.returncode, 1, r.stderr) + self.assertIn('error: cannot write', r.stderr) + self.assertNotIn('Traceback', r.stderr) + + def test_usage_errors_exit_2_before_any_fetch(self): + for args in (['--release', '6.8.0-45-generic'], ['--tag', '6.12.107'], ['--tag', 'v6/../x'], + ['--tag', 'v6.12.107', '--case', '100'], []): + r = self.run_script(*args) + self.assertEqual(r.returncode, 2, args) + self.assertEqual(self.server.hits, []) + + def test_any_tag_shape_is_fetched_and_a_missing_tag_fails_there(self): + self.server.close() + self.server = Server(self.tmp / 'empty') # nothing served: every tag is missing + self.addCleanup(self.server.close) + (self.tmp / 'empty').mkdir() + r = self.run_script('--tag', 'v6.12.0') + self.assertEqual(r.returncode, 1) + self.assertEqual(self.server.hits, ['/drivers/usb/misc/usbtest.c?h=v6.12.0']) + + def test_unknown_case_exits_1(self): + r = self.run_script('--tag', 'v6.12.107', '--case', '7') + self.assertEqual(r.returncode, 1) + self.assertIn('case 7: found 0 top-level labels', r.stderr) + + +if __name__ == '__main__': + unittest.main()
diff --git a/.claude/test/test_usbtest_run_case.py b/.claude/test/test_usbtest_run_case.py new file mode 100644 index 0000000..de08560 --- /dev/null +++ b/.claude/test/test_usbtest_run_case.py
@@ -0,0 +1,580 @@ +"""run_case.py with every hardware call stubbed: roster resolution, each refusal before +hardware, the flash -> enumerate -> battery -> park order, and that the board lock is released +on every path. Plumbing only: the real chain is proved by a run on a rig board.""" +import importlib.util +import io +import json +import os +import signal +import subprocess +import sys +import tempfile +import time +import unittest +from contextlib import redirect_stderr, redirect_stdout +from pathlib import Path +from unittest import mock + +REPO = Path(__file__).resolve().parents[2] +SCRIPT = REPO / '.claude' / 'skills' / 'usbtest' / 'scripts' / 'run_case.py' +spec = importlib.util.spec_from_file_location('run_case', SCRIPT) +run_case = importlib.util.module_from_spec(spec) +spec.loader.exec_module(run_case) + +ROSTER = { + 'boards': [ + {'name': 'solo', 'uid': 'UID1', 'flasher': {'name': 'jlink', 'uid': 'P1', 'args': '-device X'}, + 'tests': {'device': True, 'host': False}}, + {'name': 'duo', 'uid': 'UID2', 'flasher': {'name': 'openocd', 'uid': 'P2', 'args': ''}, + 'variant': [{'name': 'duo-a', 'flags': ''}, {'name': 'duo-b', 'flags': ''}], + 'tests': {'only': ['device/board_test', 'device/usbtest']}}, + {'name': 'hostonly', 'uid': 'UID4', 'flasher': {'name': 'openocd'}, 'tests': {'device': False, 'host': True}}, + {'name': 'skips', 'uid': 'UID5', 'flasher': {'name': 'jlink'}, + 'tests': {'device': True, 'skip': ['device/usbtest']}}, + {'name': 'notlisted', 'uid': 'UID6', 'flasher': {'name': 'jlink'}, 'tests': {'only': ['device/cdc_msc']}}, + {'name': 'untested', 'uid': 'UID7', 'flasher': {'name': 'jlink'}}, + {'name': 'malformed', 'uid': 'UID8', 'flasher': {'name': 'jlink'}, 'tests': {'device': True}, + 'variant': [{'name': ''}]}, + {'name': 'noflasher', 'uid': 'UID9', 'tests': {'only': ['device/usbtest']}}, + {'name': 'nouid', 'flasher': {'name': 'jlink'}, 'tests': {'only': ['device/usbtest']}}, + ], + 'boards-skip': [{'name': 'parked', 'uid': 'UID3', 'flasher': {'name': 'jlink'}}], +} +PASS_JSON = {'passed': 1, 'failed': 0, 'notrun': 0, 'wedged': False, + 'cases': [{'num': 29, 'name': 'toggle clear', 'status': 'PASS'}]} + + +class Lock: + def __init__(self): + self.closed = False + self.cleared = False + self.sigterm = None # the SIGTERM disposition when the lock was released + + def close(self): + self.closed = True + + +class Rig: + """Stubs for everything run_case.py touches, recording the order of hardware actions.""" + + def __init__(self, test): + self.calls = [] + self.firmware = {'device/usbtest': '/fw/usbtest.elf', 'device/board_test': '/fw/board_test.elf'} + self.flash_rc = 0 + self.enumerates = True + self.verdict = (PASS_JSON, '') + self.killed = '' + self.lock = Lock() + self.lock_error = None + self.peers = ([], True) + self.peer_scans = [] # per-scan live_peers results, then peers + self.flash_rcs = [] # per-flash return codes, then flash_rc + tmp = tempfile.TemporaryDirectory() + test.addCleanup(tmp.cleanup) + self.config = Path(tmp.name) / 'rig.json' + self.config.write_text(json.dumps(ROSTER)) + self.cwds = [] + + def flash_fn(board, fw): + self.calls.append(('flash', board['name'], fw)) + self.cwds.append(os.getcwd()) + rc = self.flash_rcs.pop(0) if self.flash_rcs else self.flash_rc + if isinstance(rc, BaseException): + raise rc + return subprocess.CompletedProcess('flash', rc, 'probe says no', '') + + def acquire(name, reason): + if isinstance(self.lock_error, BaseException): + raise self.lock_error + if self.lock_error: + raise RuntimeError(self.lock_error) + self.calls.append(('lock', name)) + return self.lock + + def clear(fh): + fh.cleared = True + fh.sigterm = signal.getsignal(signal.SIGTERM) + + def battery(board, fw, tests, timeout): + self.calls.append(('battery', tuple(tests), fw)) + if isinstance(self.verdict, BaseException): + raise self.verdict + return (*self.verdict, self.killed) + + for obj, name, value in ( + (run_case.hil_flash, 'find_firmware', + lambda variant, example, flasher=None: self.firmware.get(example) and Path(self.firmware[example])), + (run_case.hil_flash, 'flash_primitive', lambda name: flash_fn), + (run_case.hil_lock, 'acquire_board_lock', acquire), + (run_case.hil_lock, 'clear_record', clear), + (run_case, 'enumerated', lambda uid: self.calls.append(('enumerated', uid)) or self.enumerates), + (run_case, 'battery', battery), + (run_case, 'live_peers', lambda: self.peer_scans.pop(0) if self.peer_scans else self.peers)): + patcher = mock.patch.object(obj, name, value) + patcher.start() + test.addCleanup(patcher.stop) + + def run(self, *argv): + out, err = io.StringIO(), io.StringIO() + with mock.patch.object(sys, 'argv', ['run_case.py', '--config', str(self.config), *argv]), \ + redirect_stdout(out), redirect_stderr(err): + rc = run_case.main() + self.stdout = out.getvalue() + return rc, json.loads(self.stdout.splitlines()[-1]), err.getvalue() + + +class Refusals(unittest.TestCase): + """Each exits 2 before any lock or hardware action.""" + + def refused(self, rig, *argv, says): + rc, report, err = rig.run(*argv) + self.assertEqual(rc, 2, err) + self.assertIn(says, report['error']) + self.assertEqual([c for c in rig.calls if c[0] != 'lock'], []) + self.assertEqual(report['boardState'], 'untouched') + return rig + + def test_board_and_variant_resolution(self): + for argv, says in ( + (['--board', 'nosuch'], 'nosuch is not a board'), + (['--board', 'parked'], 'parked is in boards-skip'), + (['--board', 'hostonly'], 'hostonly does not run device/usbtest'), + (['--board', 'skips'], 'skips does not run device/usbtest'), + (['--board', 'notlisted'], 'notlisted does not run device/usbtest'), + (['--board', 'untested'], 'untested does not run device/usbtest'), + (['--board', 'malformed'], 'malformed: board malformed variant'), + (['--board', 'noflasher'], 'noflasher in'), + (['--board', 'nouid'], 'nouid in'), + (['--board', 'duo'], 'duo has variants duo-a, duo-b: pass --variant'), + (['--board', 'duo', '--variant', 'duo-c'], 'duo has no variant duo-c'), + (['--board', 'solo', '--variant', 'duo-a'], 'solo has no variant duo-a')): + rig = Rig(self) + self.refused(rig, *argv, '--tests', '29', '--after', 'leave', says=says) + self.assertEqual(rig.calls, []) + + def test_bad_case_lists(self): + for tests in ('29,29', '30', 'x', '29,'): + self.refused(Rig(self), '--board', 'solo', '--tests', tests, '--after', 'leave', + says='--tests') + + def test_duplicate_roster_names(self): + roster = json.loads(json.dumps(ROSTER)) + roster['boards'].append(dict(roster['boards'][0])) + roster['boards'][1]['variant'].append({'name': 'duo-a'}) + for board, says in (('solo', 'solo appears 2 times'), ('duo', 'lists a variant name twice')): + rig = Rig(self) + rig.config.write_text(json.dumps(roster)) + self.refused(rig, '--board', board, '--variant', 'duo-a', '--tests', '29', '--after', 'leave', says=says) + + def test_missing_firmware(self): + rig = Rig(self) + del rig.firmware['device/usbtest'] + self.refused(rig, '--board', 'solo', '--tests', '29', '--after', 'leave', says='no device/usbtest build for solo') + rig = Rig(self) + del rig.firmware['device/board_test'] + self.refused(rig, '--board', 'solo', '--tests', '29', '--after', 'park', says='no device/board_test build') + + def test_park_firmware_is_not_needed_to_leave(self): + rig = Rig(self) + del rig.firmware['device/board_test'] + self.assertEqual(rig.run('--board', 'solo', '--tests', '29', '--after', 'leave')[0], 0) + + def test_a_live_battery_or_an_unreadable_proc(self): + rig = Rig(self) + rig.peers = ([(4242, 'hil_test.py')], True) + self.refused(rig, '--board', 'solo', '--tests', '29', '--after', 'leave', says='hil_test.py (pid 4242)') + rig = Rig(self) + rig.peers = ([], False) + self.refused(rig, '--board', 'solo', '--tests', '29', '--after', 'leave', says='cannot read every process') + + def test_a_peer_that_started_before_the_lock(self): + rig = Rig(self) + rig.peer_scans = [([], True), ([(4343, 'run_case.py')], True)] + self.refused(rig, '--board', 'solo', '--tests', '29', '--after', 'leave', says='run_case.py (pid 4343)') + self.assertEqual(rig.calls, [('lock', 'solo')]) + self.assertTrue(rig.lock.cleared and rig.lock.closed) + + def test_allow_concurrent_skips_only_the_peer_check(self): + rig = Rig(self) + rig.peers = ([(4242, 'testusb')], False) + rc, _, _ = rig.run('--board', 'solo', '--tests', '29', '--after', 'leave', '--allow-concurrent') + self.assertEqual(rc, 0) + + def test_lock_held_or_unavailable(self): + rig = Rig(self) + rig.lock_error = 'board locked: {"pid": 1, "reason": "hil_test.py"}' + self.refused(rig, '--board', 'solo', '--tests', '29', '--after', 'leave', says='board locked') + rig = Rig(self) + rig.lock = None + self.refused(rig, '--board', 'solo', '--tests', '29', '--after', 'leave', says='lock unavailable') + + +class Chain(unittest.TestCase): + def released(self, rig): + self.assertTrue(rig.lock.cleared and rig.lock.closed) + + def test_pass_then_park(self): + rig = Rig(self) + cwd = os.getcwd() + rc, report, _ = rig.run('--board', 'solo', '--tests', '29', '--after', 'park') + self.assertEqual(rc, 0) + self.assertEqual(rig.calls, [('lock', 'solo'), ('flash', 'solo', '/fw/usbtest.elf'), + ('enumerated', 'UID1'), ('battery', (29,), '/fw/usbtest.elf'), + ('flash', 'solo', '/fw/board_test.elf')]) + self.assertEqual((report['pass'], report['variant'], report['boardState']), + (True, 'solo', 'parked on board_test')) + self.released(rig) + self.assertEqual(os.getcwd(), cwd) + self.assertTrue(all(c != cwd and not os.path.exists(c) for c in rig.cwds)) # temp cwd, removed + + def test_leave_keeps_usbtest_running(self): + rig = Rig(self) + rc, report, _ = rig.run('--board', 'duo', '--variant', 'duo-b', '--tests', '13,29', '--after', 'leave') + self.assertEqual(rig.calls[-1], ('battery', (13, 29), '/fw/usbtest.elf')) + self.assertEqual(report['boardState'], 'usbtest firmware') + self.assertEqual(rc, 1) # the stub verdict holds one case for two requested + + def test_a_failed_case_still_parks(self): + rig = Rig(self) + rig.verdict = ({**PASS_JSON, 'failed': 1, 'cases': [{'num': 29, 'status': 'FAIL', 'detail': 'errno 110'}]}, '') + rc, report, _ = rig.run('--board', 'solo', '--tests', '29', '--after', 'park') + self.assertEqual((rc, report['pass'], report['boardState']), (1, False, 'parked on board_test')) + self.released(rig) + + def test_wedged_is_never_parked(self): + rig = Rig(self) + rig.verdict = ({**PASS_JSON, 'wedged': True}, 'aborting battery') + rc, report, _ = rig.run('--board', 'solo', '--tests', '29', '--after', 'park') + self.assertEqual((rc, report['wedged']), (1, True)) + self.assertTrue(report['boardState'].startswith('wedged')) + self.assertEqual(sum(c[0] == 'flash' for c in rig.calls), 1) + self.released(rig) + + def test_no_verdict_is_never_parked(self): + rig = Rig(self) + rig.verdict = (None, 'killed') + rc, report, _ = rig.run('--board', 'solo', '--tests', '29', '--after', 'park') + self.assertEqual((rc, report['boardState'], report['error']), + (1, 'usbtest firmware, verdict incomplete', 'usbtest.py printed no verdict')) + self.assertEqual(sum(c[0] == 'flash' for c in rig.calls), 1) + self.released(rig) + rig.killed = 'usbtest.py killed at its 999s bound (rc 124)' + rc, report, _ = rig.run('--board', 'solo', '--tests', '29', '--after', 'park') + self.assertEqual((rc, report['error']), (1, rig.killed)) + + def test_flash_failure_or_no_device_stops_there(self): + rig = Rig(self) + rig.flash_rc = 1 + rc, report, _ = rig.run('--board', 'solo', '--tests', '29', '--after', 'park') + self.assertEqual((rc, report['boardState']), (1, 'flash failed')) + self.assertIn('probe says no', report['error']) + self.assertNotIn('enumerated', [c[0] for c in rig.calls]) + self.released(rig) + rig = Rig(self) + rig.enumerates = False + rc, report, _ = rig.run('--board', 'solo', '--tests', '29', '--after', 'park') + self.assertEqual(rc, 1) + self.assertIn('no cafe:4010 device with serial UID1', report['error']) + self.assertNotIn('battery', [c[0] for c in rig.calls]) + self.released(rig) + + def test_captured_stderr_and_dmesg_are_reported(self): + rig = Rig(self) + cases = [{'num': 13, 'name': 'halt', 'status': 'FAIL', 'detail': 'errno 32', 'dmesg': 'usbtest a\nusbtest b'}, + {'num': 29, 'status': 'FAIL', 'detail': 'testusb did not run the case (rc 1)', + 'stderr': 'no such device', 'secs': 1.0}] + rig.verdict = ({**PASS_JSON, 'failed': 2, 'cases': cases}, '') + rc, report, _ = rig.run('--board', 'solo', '--tests', '13,29', '--after', 'leave') + self.assertEqual(rc, 1) + self.assertEqual(report['cases'], [{k: v for k, v in c.items() if k != 'secs'} for c in cases]) + self.assertIn(' usbtest a\n usbtest b\n', rig.stdout) + self.assertIn(' no such device\n', rig.stdout) + + def test_case_0_keeps_its_number(self): + rig = Rig(self) + rig.verdict = ({**PASS_JSON, 'cases': [{'num': 0, 'name': 'NOP', 'status': 'PASS'}]}, '') + rc, report, _ = rig.run('--board', 'solo', '--tests', '0', '--after', 'leave') + self.assertEqual((rc, report['cases']), (0, [{'num': 0, 'name': 'NOP', 'status': 'PASS'}])) + + def test_a_failed_park_fails_the_run(self): + rig = Rig(self) + rig.flash_rcs = [0, 1] + rc, report, _ = rig.run('--board', 'solo', '--tests', '29', '--after', 'park') + self.assertEqual((rc, report['pass']), (1, False)) + self.assertTrue(report['boardState'].startswith('park failed')) + self.assertIn('probe says no', report['error']) + self.released(rig) + + def test_an_error_after_the_lock_still_reports(self): + rig = Rig(self) + rig.verdict = RuntimeError('usb_scan blew up') + rc, report, _ = rig.run('--board', 'solo', '--tests', '29', '--after', 'park') + self.assertEqual((rc, report['error']), (1, 'RuntimeError: usb_scan blew up')) + self.released(rig) + + def test_a_setup_error_after_the_lock_still_reports(self): + rig = Rig(self) + with mock.patch.object(run_case.tempfile, 'TemporaryDirectory', side_effect=OSError('disk full')): + rc, report, _ = rig.run('--board', 'solo', '--tests', '29', '--after', 'park') + self.assertEqual((rc, report['error'], report['boardState']), (1, 'OSError: disk full', 'untouched')) + self.assertEqual(rig.calls, [('lock', 'solo')]) + self.released(rig) + + def test_a_failed_cwd_restore_still_releases_the_lock(self): + rig = Rig(self) + real = os.chdir + + def chdir(d): # into the temp dir works; back to the checkout fails + if not d.startswith(tempfile.gettempdir()): + raise OSError('gone') + real(d) + + with mock.patch.object(run_case.os, 'chdir', side_effect=chdir): + with self.assertRaises(OSError): + rig.run('--board', 'solo', '--tests', '29', '--after', 'park') + real(REPO) + self.released(rig) + + def test_an_exception_still_releases_the_lock(self): + rig = Rig(self) + rig.verdict = KeyboardInterrupt() + with self.assertRaises(KeyboardInterrupt): + rig.run('--board', 'solo', '--tests', '29', '--after', 'park') + self.released(rig) + + +class Sigterm(unittest.TestCase): + """What `hil_lock.py release` sends the lock holder, raised in-process as on_sigterm would.""" + + def setUp(self): + self.addCleanup(signal.signal, signal.SIGTERM, signal.getsignal(signal.SIGTERM)) + self.previous = lambda signum, frame: None + signal.signal(signal.SIGTERM, self.previous) + patcher = mock.patch.object(run_case, 'kill_children') # the real one SIGKILLs our children + self.kill_children = patcher.start() + self.addCleanup(patcher.stop) + + def terminated_mid_run(self, rig): + cwd = os.getcwd() + rc, report, _ = rig.run('--board', 'solo', '--tests', '29', '--after', 'park') + self.assertEqual((rc, report['pass'], report['boardState']), + (1, False, 'unknown: terminated (SIGTERM) mid-run')) + self.assertIn('terminated (SIGTERM)', report['error']) + self.assertTrue(rig.lock.cleared and rig.lock.closed) + self.assertEqual(os.getcwd(), cwd) + self.assertIs(signal.getsignal(signal.SIGTERM), self.previous) + + def test_during_the_battery(self): + rig = Rig(self) + rig.verdict = run_case.Terminated() + self.terminated_mid_run(rig) + self.assertEqual(sum(c[0] == 'flash' for c in rig.calls), 1) # never parked + + def test_during_the_flash_is_not_a_failed_flash(self): + rig = Rig(self) + rig.flash_rcs = [run_case.Terminated()] + self.terminated_mid_run(rig) + self.assertNotIn('enumerated', [c[0] for c in rig.calls]) + + def test_before_the_lock_is_a_refusal(self): + rig = Rig(self) + rig.lock_error = run_case.Terminated() + rc, report, _ = rig.run('--board', 'solo', '--tests', '29', '--after', 'park') + self.assertEqual((rc, report['boardState'], report['error']), + (2, 'untouched', 'terminated (SIGTERM) before any hardware action')) + self.assertEqual(rig.calls, []) + self.assertIs(signal.getsignal(signal.SIGTERM), self.previous) + + def test_handler_installed_for_the_run_then_restored(self): + rig = Rig(self) + during = [] + battery = run_case.battery + with mock.patch.object(run_case, 'battery', + lambda *a: during.append(signal.getsignal(signal.SIGTERM)) or battery(*a)): + self.assertEqual(rig.run('--board', 'solo', '--tests', '29', '--after', 'park')[0], 0) + self.assertEqual(during, [run_case.on_sigterm]) + self.assertIs(rig.lock.sigterm, signal.SIG_IGN) # after the verdict line, before the release + self.assertIs(signal.getsignal(signal.SIGTERM), self.previous) + + def test_a_second_sigterm_is_ignored(self): + with self.assertRaises(run_case.Terminated): + run_case.on_sigterm(signal.SIGTERM, None) + self.assertIs(signal.getsignal(signal.SIGTERM), signal.SIG_IGN) + self.kill_children.assert_called_once_with() + + +class Battery(unittest.TestCase): + """The usbtest.py command and its bound.""" + + def command(self, flasher): + seen = {} + + def run_cmd(cmd, timeout, split_stderr, quiet): + seen.update(cmd=cmd, timeout=timeout) + return subprocess.CompletedProcess(cmd, 0, 'banner\n' + json.dumps(PASS_JSON), 'note') + + board = {'name': 'b', 'uid': 'UID', 'flasher': flasher} + with mock.patch.object(run_case.hil_util, 'run_cmd', run_cmd): + data, err, killed = run_case.battery(board, '/fw/u.elf', [13, 29], 60) + self.assertEqual((data, err, killed), (PASS_JSON, 'note', '')) + return seen + + def test_recovery_only_when_the_flasher_can_deliver_it(self): + with mock.patch.object(run_case.hil_flash, 'convoy_safe', lambda f: True): + seen = self.command({'name': 'openocd', 'args': '', 'uid': 'P'}) + self.assertIn('--recover-fw', seen['cmd']) + self.assertEqual(seen['cmd'][seen['cmd'].index('--tests') + 1], '13,29') + per_case = 60 + 5 + (run_case.usbtest.HELPER_TIMEOUT + 5) + 5 + budget = run_case.SETUP_S + 2 * per_case + self.assertEqual(seen['cmd'][seen['cmd'].index('--budget') + 1], str(budget)) + self.assertEqual(seen['timeout'], budget + per_case + + run_case.usbtest.recovery_reserve({'name': 'openocd', 'args': ''})) + with mock.patch.object(run_case.hil_flash, 'convoy_safe', lambda f: False): + seen = self.command({'name': 'jlink', 'args': '', 'uid': 'P'}) + self.assertNotIn('--recover-board', seen['cmd']) + self.assertEqual(seen['timeout'], budget + per_case + run_case.usbtest.WEDGE_CONFIRM_S) + + def test_a_kill_at_the_bound_is_named(self): + board = {'name': 'b', 'uid': 'UID', 'flasher': {'name': 'jlink', 'args': '', 'uid': 'P'}} + for out in ('partial', json.dumps(PASS_JSON)): # a verdict printed before the kill counts for nothing + killed = lambda cmd, timeout, split_stderr, quiet: subprocess.CompletedProcess(cmd, 124, out, '') + with mock.patch.object(run_case.hil_util, 'run_cmd', killed), \ + mock.patch.object(run_case.hil_flash, 'convoy_safe', lambda f: False): + data, _, note = run_case.battery(board, '/fw/u.elf', [29], 60) + self.assertIsNone(data) + self.assertRegex(note, r'^usbtest.py killed at its \d+s bound \(rc 124\)$') + + +class OnlyThese: + """/proc narrowed to the given pids, so the real kill_children cannot reach anything else.""" + + def __init__(self, *pids): + self.pids = pids + + def glob(self, pattern): + return [Path('/proc') / str(pid) for pid in self.pids] + + +def gone(pid): + """pid has exited and been reaped, or is a zombie waiting for its new parent to reap it.""" + try: + return Path(f'/proc/{pid}/stat').read_text().rsplit(')', 1)[1].split()[0] == 'Z' + except OSError: + return True + + +class KillChildren(unittest.TestCase): + def spawn(self, script, **kw): + p = subprocess.Popen(['sh', '-c', script], stdout=subprocess.PIPE, text=True, **kw) + self.addCleanup(p.wait) + self.addCleanup(p.stdout.close) + self.addCleanup(p.kill) # skips a child already reaped, whose pid may be reused + return p + + def echoed(self, p): + """The pid of the background process p's script echoes.""" + pid = int(p.stdout.readline()) + self.assertGreater(pid, 1) # kill(0) or kill(-1) would hit far more than this test + pidfd = os.pidfd_open(pid) # pins this process: once reaped, a signal fails, never hits a reused pid + self.addCleanup(os.close, pidfd) + self.addCleanup(self.kill, pidfd) + return pid + + @staticmethod + def kill(pidfd): + try: + signal.pidfd_send_signal(pidfd, signal.SIGKILL) + except ProcessLookupError: + pass + + def wait_gone(self, pid): + for _ in range(50): + if gone(pid): + return True + time.sleep(0.1) + return False + + def test_children_and_the_session_die_others_live(self): + leader = self.spawn('sleep 30 >/dev/null & echo $!; wait', start_new_session=True) + grandchild = self.echoed(leader) + plain = self.spawn('exec sleep 30') + helper = self.spawn('sleep 30 >/dev/null & echo $!') + stranger = self.echoed(helper) + helper.wait(timeout=5) # stranger is now nobody's child of ours + with mock.patch.object(run_case, 'PROC', OnlyThese(leader.pid, grandchild, plain.pid, stranger)): + run_case.kill_children() + self.assertEqual((leader.wait(timeout=5), plain.wait(timeout=5)), (-9, -9)) + self.assertTrue(self.wait_gone(grandchild)) + self.assertFalse(gone(stranger)) + + def test_a_permission_error_is_swallowed(self): + with tempfile.TemporaryDirectory() as d: + (Path(d) / '31').mkdir() + (Path(d) / '31' / 'stat').write_text(f'31 (a) b) S {os.getpid()} 31 0\n') + denied = mock.Mock(side_effect=PermissionError) + with mock.patch.object(run_case, 'PROC', Path(d)), \ + mock.patch.object(run_case.os, 'killpg', denied), \ + mock.patch.object(run_case.os, 'kill', denied): + run_case.kill_children() + self.assertEqual(denied.call_args_list, [mock.call(31, signal.SIGKILL)] * 2) + + +class ArgvPeer(unittest.TestCase): + def test_a_peer_is_the_program_or_the_script_python_runs(self): + for argv, peer in ( + (['/home/u/testusb', '-D', 'x'], 'testusb'), + (['sudo', '-n', 'testusb', '-t', '9'], 'testusb'), + (['python3', 'test/hil/usbtest.py'], 'usbtest.py'), + (['sudo', '-n', 'python3', '/r/test/hil/hil_test.py', 'c.json'], 'hil_test.py'), + (['timeout', '900', 'python3', 'run_case.py'], 'run_case.py'), + (['python3', '-W', 'ignore', '-u', 'test/hil/usbtest.py', '--json'], 'usbtest.py'), + (['git', 'diff', '--', 'test/hil/usbtest.py'], None), + (['vim', '-o', 'a', 'b', 'run_case.py'], None), + (['sed', '-n', '1p', 'run_case.py'], None), + (['python3', 'other.py', 'usbtest.py'], None), + (['python3', '-c', 'pass', 'usbtest.py'], None), + (['python3'], None), + ([''], None)): + with self.subTest(argv=argv): + self.assertEqual(run_case.argv_peer(argv), peer) + + +class LivePeers(unittest.TestCase): + def test_matches_battery_processes_by_argv(self): + with tempfile.TemporaryDirectory() as d: + for pid, argv in ((1, ['/sbin/init']), (11, ['sudo', '-n', '/home/u/testusb', '-D', 'x']), + (12, ['python3', 'test/hil/usbtest.py', '--json']), + (13, ['/usr/bin/python3', '/r/test/hil/hil_test.py', 'c.json']), + (14, ['python3', 'run_case.py']), (15, ['bash'])): + (Path(d) / str(pid)).mkdir() + (Path(d) / str(pid) / 'cmdline').write_bytes(b'\0'.join(a.encode() for a in argv) + b'\0') + with mock.patch.object(run_case, 'PROC', Path(d)): + peers, complete = run_case.live_peers() + self.assertEqual((sorted(peers), complete), + ([(11, 'testusb'), (12, 'usbtest.py'), (13, 'hil_test.py'), (14, 'run_case.py')], + True)) + (Path(d) / '16').mkdir() # exited: no cmdline left + self.assertTrue(run_case.live_peers()[1]) + (Path(d) / '17' / 'cmdline').mkdir(parents=True) # unreadable for another reason + self.assertFalse(run_case.live_peers()[1]) + + def test_its_own_wrappers_are_not_peers(self): + me = os.getpid() + with tempfile.TemporaryDirectory() as d: + for pid, ppid, argv in ((me, 21, ['python3', 'run_case.py']), + (21, 1, ['timeout', '900', 'python3', 'run_case.py']), + (22, 1, ['python3', 'run_case.py'])): + (Path(d) / str(pid)).mkdir() + (Path(d) / str(pid) / 'stat').write_text(f'{pid} (a) b) S {ppid} {pid} 0\n') + (Path(d) / str(pid) / 'cmdline').write_bytes(b'\0'.join(a.encode() for a in argv) + b'\0') + with mock.patch.object(run_case, 'PROC', Path(d)): + self.assertEqual(run_case.live_peers()[0], [(22, 'run_case.py')]) + + def test_a_restricted_proc_is_incomplete(self): + with tempfile.TemporaryDirectory() as d, mock.patch.object(run_case, 'PROC', Path(d)), \ + mock.patch.object(run_case.os, 'geteuid', lambda: 1000): + self.assertEqual(run_case.live_peers(), ([], False)) # hidepid: not even pid 1 + + +if __name__ == '__main__': + unittest.main()
diff --git a/CLAUDE.md b/CLAUDE.md index ef47112..c4e685f 100644 --- a/CLAUDE.md +++ b/CLAUDE.md
@@ -17,6 +17,7 @@ - Skill script tests live in `.claude/test/test_*.py`. - Hardware diagnosis and repair use agentrc's `hw-debugger`; independent hardware claims and committed-fix validation use `hw-validator`; routine HIL runs use `hil-operator`, which follows the HIL contract below — all orchestrated by `chief`. - Driver audit: the `code-audit` workflow with `dirs` (e.g. `src/portable/<vendor>/<driver>`) and `dimensions`: `correctness: transfer state machines, endpoint bookkeeping, completion and error paths`; `ISR safety: work deferred to task context, shared-state races, register access ordering`; `register use vs datasheet and MCU errata: cross-check the reference manual AND errata sheets via the read-doc skill; if the skill is unavailable treat the document as absent (low confidence, never a web/filesystem substitute); a missing erratum workaround is a finding`; `style: repo conventions (TU_ASSERT, no dynamic allocation, include order, naming)`. +- usbtest battery: load `usbtest` before bringing up `examples/device/usbtest` on a new MCU or DCD, or before debugging a failed usbtest case (a red usbtest cell, a testusb errno, NOTRUN or HUNG); it maps the case to the DCD path and runs chosen cases on one board. - Static analysis: the `pvs-studio` agent with rules `.PVS-Studio/.pvsconfig` (never add suppressions) on a board's examples build; `raspberry_pi_pico` mirrors CI, `stm32f407disco` is fastest. ## Build and Validate @@ -27,6 +28,7 @@ - ESP-IDF: `. "$IDF_PATH/export.sh"` before anything Espressif; verification still goes through the build contract, with `idf.py -DBOARD=<board> flash monitor` in the example reserved for interactive flash and monitor. - Before submitting, and again after a rebase: `pre-commit run --all-files` (includes unit tests). - For code changes: build the full example set for boards that exercise the changed modules. Add fuzz/HIL coverage for parsers or protocol state machines. +- For DCD bring-up or a behavior change (`src/portable/<vendor>/<driver>`), use the usbtest battery for hardware sign-off: run every case of the justified tier. If no suitable rig board is available, report hardware validation pending. - After board/dependency changes, regenerate docs with `build-doc`. - Before committing code changes, verify size impact with `code-size`.
diff --git a/docs/porting.rst b/docs/porting.rst index 9b60ec5..643df0a 100644 --- a/docs/porting.rst +++ b/docs/porting.rst
@@ -240,6 +240,11 @@ Stalling is one way an endpoint can indicate failure such as when an unsupported command is transmitted. The pair of ``dcd_edpt_stall()``, ``dcd_edpt_clear_stall()`` help manage the stall state of all endpoints. +Validate with usbtest +--------------------- + +Once ``cdc_msc`` enumerates, build ``examples/device/usbtest`` and run the Linux kernel ``usbtest`` battery against it from a Linux host. It stresses the device controller driver harder than a class example, and its tiers add vendor control, interrupt and isochronous coverage as the port supports them. See the `example README <https://github.com/hathach/tinyusb/tree/master/examples/device/usbtest>`_ for the tiers, the cases and how to run it. + Woohoo! -------
diff --git a/test/hil/test/test_ci_select.py b/test/hil/test/test_ci_select.py index 86a9c02..b02ba7c 100644 --- a/test/hil/test/test_ci_select.py +++ b/test/hil/test/test_ci_select.py
@@ -1219,6 +1219,7 @@ 'test/hil/test/test_hil_reset_order.py', 'test/hil/test/test_hil_rtt.py', 'test/hil/test/test_hil_usbtest_id.py', + 'test/hil/test/test_hil_usbtest_verdict.py', 'test/hil/test/test_hil_util.py', 'test/hil/test/test_membrowse_cli.py', 'test/hil/test/usbtest_harness.py',
diff --git a/test/hil/test/test_hil_usbtest_verdict.py b/test/hil/test/test_hil_usbtest_verdict.py new file mode 100644 index 0000000..e2d4fc6 --- /dev/null +++ b/test/hil/test/test_hil_usbtest_verdict.py
@@ -0,0 +1,142 @@ +#!/usr/bin/env python3 +# SPDX-License-Identifier: MIT +"""usbtest's per-case verdict and the battery's exit status. run_case runs a fake testusb +that prints kernel tools/usb/testusb.c's own strings; main() runs with stubbed discovery. +The strings are checked against the rig's testusb by a real battery run.""" +import io +import json +import os +import stat +import sys +import types +import unittest +from contextlib import redirect_stderr, redirect_stdout +from tempfile import TemporaryDirectory + +TEST_DIR = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, os.path.dirname(TEST_DIR)) +import usbtest_harness # noqa: E402 - stubs serial before hil_test is imported +import usbtest # noqa: E402 + +NODE = '/dev/bus/usb/001/005' +# testusb -D <node> -t <n>: stderr first, then stdout. FAKE_OUT is a format string over the +# node and case; '%s' is the device name as testusb prints it. +FAKE_TESTUSB = f'''#!{sys.executable} +import os, sys +node = sys.argv[sys.argv.index('-D') + 1] +case = int(sys.argv[sys.argv.index('-t') + 1]) +sys.stdout.write(os.environ['FAKE_OUT'].replace('%s', node).replace('%d', str(case))) +sys.exit(int(os.environ.get('FAKE_RC', '0'))) +''' +SPEED = '%s: %s may see only control tests\nhigh speed\t%s\t0\n' # testusb.c main, handle_testdev + + +class RunCase(unittest.TestCase): + def setUp(self): + td = TemporaryDirectory() + self.addCleanup(td.cleanup) + self.testusb = os.path.join(td.name, 'testusb') + with open(self.testusb, 'w') as f: + f.write(FAKE_TESTUSB) + os.chmod(self.testusb, stat.S_IRWXU) + usbtest_harness.patch(self, usbtest, 'dmesg_tail', lambda: '') + usbtest_harness.patch(self, usbtest.os, 'access', lambda path, mode: True) # no sudo wrap + self.addCleanup(os.environ.pop, 'FAKE_OUT', None) + self.addCleanup(os.environ.pop, 'FAKE_RC', None) + + def case(self, out, num=1, rc=0): + os.environ['FAKE_OUT'], os.environ['FAKE_RC'] = out, str(rc) + return usbtest.run_case(num, {'node': NODE, 'speed': '480'}, self.testusb, False, 30) + + def test_pass_line(self): + r = self.case(SPEED + '%s test %d, 3.004567 secs\n') # "%s test %d, %4d.%.06d secs" + self.assertEqual((r['status'], r['secs']), ('PASS', 3.004567)) + + def test_perf_case_reports_throughput(self): + r = self.case(SPEED + '%s test %d, 2.000000 secs\n', num=27) + self.assertEqual(r['status'], 'PASS') + self.assertEqual(r['mbps'], round(128 * 1024 * 32 / 2 / 1e6, 2)) # HS -c 128 -s 1024 -g 32 + + def test_error_line(self): + r = self.case(SPEED + '%s test %d --> 110 (Connection timed out)\n') # "%s test %d --> %d (%s)" + self.assertEqual((r['status'], r['detail']), ('FAIL', 'errno 110 (Connection timed out)')) + + def test_another_cases_result_line_is_not_this_ones(self): + r = self.case(SPEED + '%s test 2, 1.000000 secs\n', num=1) + self.assertEqual(r['status'], 'NOTRUN') + + def test_opened_but_no_result_is_a_kernel_skip(self): + r = self.case(SPEED) # -EOPNOTSUPP: testusb continues silently + self.assertEqual(r['status'], 'NOTRUN') + + def test_other_nodes_it_could_not_open_do_not_change_a_skip(self): + r = self.case('/dev/bus/usb/002/001: Permission denied\n' + SPEED) # ftw's perror(name) + self.assertEqual(r['status'], 'NOTRUN') + + def test_never_opened_the_node_is_a_failure_not_a_skip(self): + for out, rc in (("can't open dev file r/w: Permission denied\n", 0), + ("must specify '-a' or '-D dev', or DEVICE=/dev/bus/usb/BBB/DDD in env\n", 1), + ('USB device files are missing\n', 255), + ('', 0)): + r = self.case(out, rc=rc) + self.assertEqual(r['status'], 'FAIL', out) + self.assertEqual(r['detail'], f'testusb did not run the case (rc {rc})') + self.assertEqual(r['stderr'], out.strip()) + + def test_speed_line_for_another_node_does_not_count(self): + r = self.case('high speed\t/dev/bus/usb/001/006\t0\n') + self.assertEqual(r['status'], 'FAIL') + + +class ExitStatus(unittest.TestCase): + """What main() returns and says when a battery ends without a clean verdict.""" + + def main(self, *extra, find=None, status='PASS', stranded=False): + usbtest_harness.stub_device(self, usbtest, lambda num, d, tu, quick, timeout: + {'num': num, 'name': 'x', 'params': '', 'status': status, 'secs': 0.1}) + usbtest_harness.patch(self, usbtest, 'register_usbtest_id', lambda: None) + usbtest_harness.patch(self, usbtest, 'bind_usbtest', lambda d: None) + if find: + usbtest_harness.patch(self, usbtest, 'find_device', find) + if stranded: + hu = types.SimpleNamespace(path_stranded=lambda p: True) + usbtest_harness.patch(self, usbtest, '_hu', lambda: hu) + usbtest_harness.argv(self, *extra) + out, err = usbtest_harness.Out(), io.StringIO() + with redirect_stdout(out), redirect_stderr(err): + rc = usbtest.main() + text = out.getvalue() + return rc, json.loads(text[text.find('{'):]), err.getvalue() + + def lookups(self, *answers): + seq = iter(answers) + return lambda serial, first=False: next(seq) + + def test_ambiguous_device_after_the_last_case_is_not_a_success(self): + rc, data, _ = self.main('--tests', '0', find=self.lookups( + dict(usbtest_harness.DEV), {'ambiguous': ['1-1', '1-2']})) + self.assertEqual((rc, data['passed'], data['wedged']), (1, 1, True)) + + def test_unreadable_device_after_the_last_case_is_not_a_success(self): + rc, data, _ = self.main('--tests', '0', stranded=True, + find=self.lookups(dict(usbtest_harness.DEV), None)) + self.assertEqual((rc, data['wedged']), (1, True)) + + def test_clean_run_still_exits_0(self): + rc, data, err = self.main('--tests', '0') + self.assertEqual((rc, data['wedged'], err), (0, False, '')) + + def test_failed_case_points_at_the_skill_once(self): + for status in ('FAIL', 'NOTRUN'): + rc, data, err = self.main('--tests', '0,1', status=status) + self.assertEqual((rc, data['failed']), (2, 2)) + self.assertEqual(err.count('usbtest skill'), 1, status) + + def test_duplicate_case_numbers_are_refused(self): + with self.assertRaises(SystemExit) as e: + self.main('--tests', '0,1,0') + self.assertIn('case 0 is listed twice', str(e.exception.code)) + + +if __name__ == '__main__': + unittest.main()
diff --git a/test/hil/usbtest.py b/test/hil/usbtest.py index e230614..701cf86 100755 --- a/test/hil/usbtest.py +++ b/test/hil/usbtest.py
@@ -519,6 +519,14 @@ 'is not writable') return result + # testusb prints '<speed> speed\t<node>\t<ifnum>' once it has opened the node + # (tools/usb/testusb.c handle_testdev). Without it testusb never reached the ioctl: an + # open failure, a usage error. Not NOTRUN, which sends the reader to the binding profile. + if not re.search(rf'^\S+ speed\t{re.escape(dev["node"])}\t\d+$', out, re.M): + result.update(status='FAIL', detail=f'testusb did not run the case (rc {p.returncode})', + stderr=out.strip()) + 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', @@ -585,6 +593,9 @@ 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)})') + # results are keyed by case number: an unrun duplicate would vanish from the report + if int(tok) in cases: + sys.exit(f'--tests: case {int(tok)} is listed twice') cases.append(int(tok)) else: cases = [n for t in range(1, tier + 1) for n in TIER_CASES[t]] @@ -781,10 +792,14 @@ print(f" FAILED test {r['num']}: {r.get('detail', '')}") if r.get('dmesg'): print(' ' + r['dmesg'].replace('\n', '\n ')) + if failed: + print('diagnose failed cases with the usbtest skill, "Failed case" ' + '(.claude/skills/usbtest/SKILL.md)', file=sys.stderr) # 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) + # false success to any script driving this directly. So does a device reported wedged + # after its last case passed. + return len(failed) + len(notrun) or int(unrecovered_hang) if __name__ == '__main__':