mirror/tinyusb
3
0
Fork 0
mirror of https://github.com/hathach/tinyusb.git synced 2026-08-18 02:53:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

hil: split hil_test.py into hil_lock/hil_flash, add pool_check, update rig probes (#3794)

Browse Source 
test/hil: add board-pool health check, split hil_test into focused modules (#3794)

Add test/hil/hil_pool_check.py: per-board rig health scan — probe presence,
light-example flash (dfu_runtime; device_info + serial check for host-only
boards), uid re-enumeration, safe recovery (probe authorized-toggle, board
reset), verified board_test re-park, USB topology report, and a markdown
summary table. Missing firmware is built on the spot (tools/build.py, idf.py
for espressif, one get_deps retry); row statuses: ok, flash-failed, failed,
locked. Board locks are always respected, never bypassed.

Refactor hil_test.py into hil_lock.py (flock protocol, controller permits,
hold/release/status CLI; replaces board_lock.py) and hil_flash.py (flashers,
find_firmware, run_cmd). Update WCH probe uids and the board roster in
tinyusb.json; add the hil-pool-check skill.
This commit is contained in:
Ha Thach
2026-07-29 17:29:59 +07:00
committed by GitHub
parent 538ec3e332
commit e88fc441dd
21 changed files with 2477 additions and 783 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
.claude/agents/hil-operator.md
View File

@ -7,7 +7,7 @@ model: sonnet
You operate physical USB test hardware. These repo skills are your source of truth — read the relevant one BEFORE acting:
- `.claude/skills/hil/SKILL.md` — run `hostname` first (host `ci` = local mode with `test/hil/tinyusb.json`; host `htpc` = local `local.json` or remote via `test/hil/hil_ci.sh`); the board lock protocol; exact `hil_test.py` invocations.
- `.claude/skills/hil/SKILL.md` — run `hostname` first (host `ci` = local mode with `test/hil/tinyusb.json`; host `tusb` = local mode with `test/hil/hfp.json`; any other host (dev PC) = local `local.json` or remote via `test/hil/hil_ci.sh`); the board lock protocol; exact `hil_test.py` invocations.
- `.claude/skills/usb-kernel-recover/SKILL.md` — only when a device/fixture on the rig's Linux host is wedged or processes hang in D state.
- `.claude/skills/usb-kernel-debug/SKILL.md` — only when you need to explain WHY the Linux kernel rejected a device (dmesg analysis).
@ -18,12 +18,12 @@ The GitHub Actions runner keeps running during your work. Per-board flock locks
- `python3 test/hil/hil_test.py ...` runs: do NOT pre-hold those boards — `hil_test.py` self-locks each board for its flash+test and would fail fast with `board locked` against your own hold.
- ANY other hardware action (JLinkExe/openocd/GDB, manual flash, usbtest.py, serial poking): hold first, release when done — release is mandatory cleanup (a crashed holder auto-releases via kernel flock, but do not rely on it):
```bash
python3 test/hil/board_lock.py hold <board...> --reason "<task>"
python3 test/hil/hil_lock.py hold <board...> --reason "<task>"
# ... hardware work ...
python3 test/hil/board_lock.py release <board...>
python3 test/hil/hil_lock.py release <board...>
```
- Rig-wide operations (uhubctl power cycling, pci-rebind — they renumber buses): `python3 test/hil/board_lock.py hold --all --reason "<why>"` first.
- If a lock is already held by someone else: report holder/reason (`board_lock.py status`) — never force, never kill the holder. If the holder's reason is `hil_test.py`, that is a concurrent CI job mid-test on the board: waiting a few minutes and retrying once is appropriate when your task allows; otherwise return the holder info so the orchestrator can ask the user.
- Rig-wide operations (uhubctl power cycling, pci-rebind — they renumber buses): `python3 test/hil/hil_lock.py hold --all --reason "<why>"` first.
- If a lock is already held by someone else: report holder/reason (`hil_lock.py status`) — never force, never kill the holder. If the holder's reason is `hil_test.py`, that is a concurrent CI job mid-test on the board: waiting a few minutes and retrying once is appropriate when your task allows; otherwise return the holder info so the orchestrator can ask the user.
- You cannot ask the user anything. Bypassing a lock (`HIL_NO_BOARD_LOCK=1`, or proceeding with manual hardware work despite a held lock) is allowed ONLY when your prompt explicitly states the user authorized forcing.
## Hard rules

2
.claude/agents/target-debugger.md
View File

@ -46,7 +46,7 @@ the next technique you would try.
## Lock discipline
- Hold the board lock for the WHOLE session (`board_lock.py hold <board>
- Hold the board lock for the WHOLE session (`hil_lock.py hold <board>
--reason "target debug: <bug>"`). Multi-hour holds are fine; never stop the
actions-runner. Locks held by others: report holder/reason, never force
unless your prompt states the user authorized it.

2
.claude/skills/etm-trace/SKILL.md
View File

@ -43,7 +43,7 @@ this skill for exact counts, coverage, or instruction-by-instruction history.
capture script uses automation port **19201**, never an interactive Ozone's
19200.
- Hold the board lock (see the `hil` skill):
`python3 test/hil/board_lock.py hold <board> --reason "etm capture"`.
`python3 test/hil/hil_lock.py hold <board> --reason "etm capture"`.
- Committed `hw/bsp/**/ozone/*.jdebug` are the maintainer's interactive
projects — automation never opens them (Ozone rewrites project files); the
script generates a throwaway project.

63
.claude/skills/hil-pool-check/SKILL.md Normal file
View File

@ -0,0 +1,63 @@
---
name: hil-pool-check
description: Use when asked for a pool check or board/probe health scan on a TinyUSB HIL rig, when probes or boards are offline or fail to flash, after rig maintenance, reboot, or re-cabling, or before starting a HIL test campaign.
---
# HIL Pool Check (board/probe health)
Health-scan the HIL board pool with `test/hil/hil_pool_check.py`: per board it checks the flash
probe is on the USB bus, flashes a light example (`device/dfu_runtime`; host-only boards get
`host/device_info`, verified by serial output), waits for the board's uid to re-enumerate,
applies safe per-device recovery (probe authorized-toggle, board reset), re-parks with
`board_test`, and prints a summary table plus a USB topology report. Flags and details: `--help`
and the module docstring.
**REQUIRED BACKGROUND:** the `hil` skill owns config-by-hostname selection (run `hostname`
first) and the board-lock protocol. Locked boards are reported 🔒 locked and skipped — never
waited on, never bypassed; a needed build peeks the lock first. A CI worker reaching a board the
pool check holds fails it as "board locked" — prefer running between CI runs.
## A "pool check" means the full check
A request for a "pool check" means the DEFAULT full check below. Use `--scan-only` only when the
user explicitly asks for a quick look, or when you have VERIFIED a CI sweep is mid-run right now
(`python3 test/hil/hil_lock.py status` shows `hil_test.py` holders) — "CI might be running" is not
that predicate: the full check is already lock-safe (CI-held boards report 🔒 locked and are never
touched), so an unconfirmed suspicion is no reason to downgrade. In either scan case say which
mode ran and why; never silently substitute the scan for the full check.
```bash
python3 test/hil/hil_pool_check.py # full check: ~10 s + ~1-2 s/board with firmware built;
# first run on an unbuilt tree takes minutes (it builds)
python3 test/hil/hil_pool_check.py --scan-only # USB presence only, <1 s, no locks/flashing/building
python3 test/hil/hil_pool_check.py -b BOARD [-b …] # subset; may name boards-skip (parked) entries
# from a dev PC, against the ci rig (bash -lc: flashers like STM32_Programmer_CLI live in ~/bin):
ssh ci.lan 'bash -lc "cd ~/code/tinyusb && python3 test/hil/hil_pool_check.py"'
```
## Notes
Missing firmware is **built on the spot** — never skipped (`--no-build` opts out; those boards
then report `flash-failed`). Builds need the family env, exported on the rig in
`~/.profile`/`~/.bashrc`: `PICO_SDK_PATH` for rp2040/rp2350 (`~/code/pico/pico-sdk`), the
ESP-IDF env (`get-idf`) for espressif — which also needs `esptool` on PATH (pip's
`~/.local/bin/esptool`; a non-login shell may lack it — run via `bash -lc`). An explicit `-B` is
searched exclusively for *existing* firmware; builds still land in `cmake-build/` and are noted
`built <example>`. Espressif boards park too when the IDF env is present. A first run on an
unbuilt tree builds for many minutes: use a command timeout ≥ 30 min and NEVER cancel early — a
killed run leaves detached cmake/ninja children still writing to `cmake-build/`.
Statuses: `ok` (flashed and verified; in `--scan-only` it only means the probe is present),
`flash-failed` (firmware delivery failed: probe missing, build failed, flasher error, silent
no-op, park unverified), `failed` (check ran but did not verify), `locked` (flock held;
untouched). Exit code = `flash-failed` + `failed` (clamped at 125); `locked` and scan-only rows
are *unverified*, not healthy — read the footer, not just `$?`. A `⚠ pid … source says …` note
means stale firmware or a silent flash no-op (J-Link lore); a device off the bus entirely needs
the usb-kernel-recover skill or a physical replug.
## Reporting
The user-facing answer to a pool check IS the tool's summary table: paste the complete per-board
table (and footer counts) verbatim — never truncate rows or reduce it to a prose digest like
"27/27 healthy"; at most one line of commentary below it.

46
.claude/skills/hil/SKILL.md
View File

@ -1,22 +1,22 @@
---
name: hil
description: Use when running TinyUSB Hardware-in-the-Loop (HIL) tests on physical boards, debugging HIL failures, or copying firmware to the ci.lan test rig. Covers per-host config selection (htpc uses local.json, ci uses tinyusb.json), local execution on either htpc or ci, remote execution over SSH from htpc, and debugging tips.
description: Use when running TinyUSB Hardware-in-the-Loop (HIL) tests on physical boards, debugging HIL failures, or copying firmware to the ci.lan test rig. Covers per-host config selection (infra rigs ci/tusb use tinyusb.json/hfp.json, any dev PC uses local.json), local and remote execution, the board-lock protocol, and debugging tips. For board/probe health scans ("pool check") use the hil-pool-check skill.
---
# Hardware-in-the-Loop (HIL) Testing
Run TinyUSB HIL tests on real boards. **Run `hostname` first** — it tells you which host you are on, which determines the default config and whether remote mode is possible.
Run TinyUSB HIL tests on real boards. **Run `hostname` first** — it tells you which host you are on, which determines the default config and whether remote mode is possible. Rule of thumb: only `ci` and `tusb` are infra rigs; **any other hostname is a dev PC** and uses `local.json`.
| Host | Local config | Remote (SSH → ci.lan)? |
|----------------------------|--------------------------------------|------------------------------------------------------|
| `htpc` (dev PC) | `test/hil/local.json` | yes (large pool, `test/hil/tinyusb.json`) |
| `ci` (the rig) | `test/hil/tinyusb.json` (large pool) | no — can't SSH to htpc, and boards are already local |
| `hifiphile` (external rig) | `test/hil/hfp.json` | no outbound SSH to htpc/ci; SSH-reachable FROM both |
| Host | Local config | Remote (SSH → ci.lan)? |
|-----------------------------------|--------------------------------------|--------------------------------------------------------|
| `ci` (the rig) | `test/hil/tinyusb.json` (large pool) | no — boards are already local |
| `tusb` (hifiphile's external rig) | `test/hil/hfp.json` | no outbound SSH to dev PCs/ci; SSH-reachable FROM both |
| anything else (a dev PC) | `test/hil/local.json` | yes (large pool, `test/hil/tinyusb.json`) |
Default to **local**. Use **remote** only when on `htpc` and the user says `remote`/`ci.lan`. Never attempt remote on `ci`.
Default to **local**. Use **remote** only when on a dev PC and the user says `remote`/`ci.lan`. Never attempt remote on `ci`.
`hifiphile` is an external rig (hosted by maintainer hifiphile), exercised by the GitHub CI
`hil-tinyusb (hfp.json)` matrix job — **never run HIL against it unless the user explicitly asks.**
`tusb` (ssh alias `hifiphile`) is an external rig (hosted by maintainer hifiphile), exercised by the
GitHub CI `hil-tinyusb (hfp.json)` matrix job — **never run HIL against it unless the user explicitly asks.**
## Board locks — the CI runner keeps running
@ -26,30 +26,35 @@ The `ci` rig also hosts a GitHub Actions runner that flashes boards and runs HIL
- For hardware work outside `hil_test.py` (JLink/GDB, manual flashing, `usbtest.py`, serial poking), hold the lock first:
```bash
python3 test/hil/board_lock.py hold BOARD [BOARD...] --reason "why"
python3 test/hil/hil_lock.py hold BOARD [BOARD...] --reason "why"
# ... hardware work ...
python3 test/hil/board_lock.py release BOARD [BOARD...]
python3 test/hil/hil_lock.py release BOARD [BOARD...]
```
- Never pre-hold boards you are about to run `hil_test.py` on — it self-locks and would treat your own hold as a conflict.
- Rig-wide operations (uhubctl power cycling, pci-rebind — bus renumbering) affect every board: `board_lock.py hold --all --reason "..."` first.
- `board_lock.py status` lists holders. Locks auto-release when the holder process dies (kernel flock); `/tmp` clears on reboot.
- Rig-wide operations (uhubctl power cycling, pci-rebind — bus renumbering) affect every board: `hil_lock.py hold --all --reason "..."` first.
- `hil_lock.py status` lists holders. Locks auto-release when the holder process dies (kernel flock); `/tmp` clears on reboot.
- Forcing past a lock: `HIL_NO_BOARD_LOCK=1 python3 test/hil/hil_test.py ...` bypasses the guard without killing the holder. Only with the user's explicit go-ahead — they accept the risk of colliding with whatever holds the board.
## Pool check (board/probe health)
Board/probe health scanning (`test/hil/hil_pool_check.py`) has its own skill: **hil-pool-check**.
Use it before a HIL campaign, after rig maintenance/reboot, or when boards fail to flash.
## Prerequisites
Examples must be built for the target board(s) — see CLAUDE.md "Build" → "All examples for a board" (produces `examples/cmake-build-<board>/`). `-B examples` points `hil_test.py` at that parent folder.
Examples must be built for the target board(s) — see CLAUDE.md "Build" → "All examples for a board" (produces `examples/cmake-build-<board>/`). `-B examples` points `hil_test.py` at that parent folder. (This applies to `hil_test.py`; `hil_pool_check.py` builds its own missing firmware.)
## Arguments
- **Board:** `-b BOARD_NAME` for one board; omit to run all boards in the config.
- **Pass-through:** `-v`, `-r N`, etc. forwarded unchanged.
If `local.json` is missing on `htpc`, ask the user to supply one (only fall back to `tinyusb.json` if told to).
If `local.json` is missing on a dev PC, ask the user to supply one (only fall back to `tinyusb.json` if told to).
## Local execution
Set `CONFIG` from `hostname` first (`test/hil/local.json` on htpc, `test/hil/tinyusb.json` on ci):
Set `CONFIG` from `hostname` first (`test/hil/local.json` on a dev PC, `test/hil/tinyusb.json` on ci, `test/hil/hfp.json` on tusb):
```bash
CONFIG=test/hil/local.json # on ci use: CONFIG=test/hil/tinyusb.json
@ -61,7 +66,7 @@ python3 test/hil/hil_test.py -B examples "$CONFIG"
python3 test/hil/hil_test.py -b stm32f723disco -B examples "$CONFIG"
```
## Remote execution (htpc → ci.lan only)
## Remote execution (dev PC → ci.lan only)
`test/hil/hil_ci.sh` handles dir setup, scp of test scripts, rsync of firmware (`.elf`/`.bin`/`.hex`), and runs `hil_test.py` on `ci.lan` with `tinyusb.json`:
@ -81,4 +86,7 @@ Runs take 2-5 min. Use a timeout ≥ 20 min (1200000 ms). NEVER cancel early.
## Reporting
Show the output, summarize pass/fail per board. On failure, retry with `-v`; if that's not enough, add temporary debug prints to `hil_test.py`.
The user-facing answer to a HIL run IS the tool's summary table: paste the complete per-board
table (and footer counts) verbatim — never truncate rows or reduce it to a prose digest; at most
one line of commentary below it. On failure, retry with `-v`; if that's not enough, add temporary
debug prints to `hil_test.py`.

2
.claude/skills/pre-pr/SKILL.md
View File

@ -24,7 +24,7 @@ Run the software + hardware gate for the current branch. The user invoking this
## 3. HIL boards
- `hilBoards` = chosen boards that are on the rig roster. This host must be able to reach the rig (per `.claude/skills/hil/SKILL.md`: host `ci` = local, `htpc` = remote). If none qualify, run software-only.
- `hilBoards` = chosen boards that are on the rig roster. This host must be able to reach the rig (per `.claude/skills/hil/SKILL.md`: host `ci`/`tusb` = local, any other host (dev PC) = remote). If none qualify, run software-only.
## 4. Launch

6
.claude/skills/target-debug/SKILL.md
View File

@ -30,9 +30,9 @@ Hold the board lock for the WHOLE manual session; never stop the
actions-runner (see the `hil` skill for the full lock protocol):
```bash
python3 test/hil/board_lock.py hold <board> --reason "target debug: <bug>"
python3 test/hil/hil_lock.py hold <board> --reason "target debug: <bug>"
# ... instrument / build / flash / capture / GDB ...
python3 test/hil/board_lock.py release <board>
python3 test/hil/hil_lock.py release <board>
```
Board → probe mapping: `test/hil/tinyusb.json` — `flasher.name` is the probe
@ -44,7 +44,7 @@ family, `flasher.uid` the **probe serial** (many identical probes on the rig):
`hw/bsp/<family>/boards/<board>/board.cmake` (or `board.mk`); family via
`ls -d hw/bsp/*/boards/<board>`.
- Run on the host that owns the probe — config `test/hil/tinyusb.json` on ci,
`local.json` on htpc (`hil` skill).
`test/hil/hfp.json` on tusb, `local.json` on any other host (dev PC) (`hil` skill).
- Espressif boards (S3/P4): different toolchain, probe model, and PHY
constraints entirely — read `esp-target-debug` first.

2
.claude/skills/usb-kernel-recover/SKILL.md
View File

@ -57,7 +57,7 @@ refuses a busport that now names a different device.
It bounces **every fixture under that root port** — on ci that is up to 25
devices. Hold the affected boards' locks first if you can, but note
`board_lock.py` uses `LOCK_EX | LOCK_NB` and so fails immediately when CI already
`hil_lock.py` uses `LOCK_EX | LOCK_NB` and so fails immediately when CI already
holds them; there is no wait-for-lock. When CI is mid-run you are choosing
between bouncing its fixtures and leaving the bus wedged for everything. The
automated path in `usbtest.py` takes no locks at all and accepts that collateral

10
.idea/codeStyles/Project.xml generated Normal file
View File

@ -0,0 +1,10 @@
<component name="ProjectCodeStyleConfiguration">
<code_scheme name="Project" version="173">
<RiderCodeStyleSettings>
<option name="/Default/CodeStyle/CodeFormatting/CppClangFormat/EnableClangFormatSupport/@EntryValue" value="true" type="bool" />
</RiderCodeStyleSettings>
<clangFormatSettings>
<option name="ENABLED" value="true" />
</clangFormatSettings>
</code_scheme>
</component>

5
.idea/codeStyles/codeStyleConfig.xml generated Normal file
View File

@ -0,0 +1,5 @@
<component name="ProjectCodeStyleConfiguration">
<state>
<option name="USE_PER_PROJECT_SETTINGS" value="true" />
</state>
</component>

2
.idea/hil-pool-check.iml generated Normal file
View File

@ -0,0 +1,2 @@
<?xml version="1.0" encoding="UTF-8"?>
<module classpath="CIDR" type="CPP_MODULE" version="4" />

8
.idea/modules.xml generated Normal file
View File

@ -0,0 +1,8 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="ProjectModuleManager">
<modules>
<module fileurl="file://$PROJECT_DIR$/.idea/hil-pool-check.iml" filepath="$PROJECT_DIR$/.idea/hil-pool-check.iml" />
</modules>
</component>
</project>

355
docs/superpowers/plans/2026-07-28-hil-test-split.md Normal file
View File

@ -0,0 +1,355 @@
# hil_test.py Split Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** Split `test/hil/hil_test.py` (2370 ln) into a test-focused core plus `hil_lock.py` (board locks + controller permits + operator CLI, superseding `board_lock.py`) and `hil_flash.py` (run_cmd + flash backends + firmware/serial lookup), with no behavior change.
**Architecture:** Pure code motion per `docs/superpowers/specs/2026-07-28-hil-test-refactor-design.md`. Import graph: `hil_test` → {`hil_lock`, `hil_flash`}; helpers import nothing local. Call sites use module-qualified names (`hil_lock.flash_permit(...)`), never wildcard mirroring.
**Tech Stack:** Python 3.11+ (existing `TypedDict`/`NotRequired` usage), stdlib only in the helpers (fcntl, json, glob, multiprocessing objects passed in).
## Global Constraints
- Work in worktree `.claude/worktrees/hil-test-split` (branch `claude/hil-test-split`); never touch the primary checkout.
- Behavior-preserving: `hil_test.py` CLI args, log lines, report format, lock/permit semantics, flash behavior all byte-identical. The ONLY user-visible change is the CLI filename `board_lock.py` → `hil_lock.py`.
- Moved functions are moved **verbatim** — no reformatting, no comment editing, no "improvements". A diff of a moved function's body against its old self must be empty.
- Commit messages: imperative, scoped, no Co-Authored-By/Claude-Session trailers.
- Every commit leaves the tree working: `python3 -m py_compile` clean on all touched modules, and `python3 .claude/skills/hil/pool_check.py --scan-only` exits 0 (safe on the rig: scan-only takes no locks, flashes nothing).
- Hardware steps (Task 4) run on the `ci` rig only, from this worktree, and rely on the tools' own board flocks — never pre-hold boards you are about to run `hil_test.py`/`pool_check.py` on.
---
### Task 1: Create hil_flash.py; repoint hil_test + pool_check flash call sites
**Files:**
- Create: `test/hil/hil_flash.py`
- Modify: `test/hil/hil_test.py` (delete moved code; add import; qualify call sites)
- Modify: `.claude/skills/hil/pool_check.py` (flash-related imports)
- Modify: `test/hil/hil_ci.sh` (scp list)
**Interfaces:**
- Produces (used by Tasks 2-4): module `hil_flash` with `CMD_TIMEOUT`, `run_cmd(cmd, cwd=None, timeout=CMD_TIMEOUT)`, `cmd_stdout_text(out)`, `OPENCOD_ADI_PATH`, `TINYUSB_ROOT`, `flash_jlink/reset_jlink`, `flash_stlink/reset_stlink`, `flash_stflash/reset_stflash`, `flash_openocd/reset_openocd`, `flash_openocd_wch/reset_openocd_wch`, `flash_openocd_adi/reset_openocd_adi`, `flash_wlink_rs/reset_wlink_rs`, `flash_esptool/reset_esptool`, `flash_uniflash/reset_uniflash`, `flash_lm4flash/reset_lm4flash`, `find_firmware(variant, example)`, `get_serial_dev(id, vendor_str, product_str, ifnum)`, module globals `build_dir = 'cmake-build'`, `verbose = False`.
- [ ] **Step 1: Create `test/hil/hil_flash.py`**
Header (new code), then the moved blocks verbatim:
```python
#!/usr/bin/env python3
# SPDX-License-Identifier: MIT
# Firmware flashing for the TinyUSB HIL rig: run_cmd, one flash_*/reset_* pair per
# flasher type (dispatched by config name via getattr), find_firmware, and the
# fixture serial-port resolver get_serial_dev (here, not hil_test: flash_esptool
# needs it and helpers must not import hil_test).
# Callers set module globals `build_dir` and `verbose` (hil_test.main from argparse,
# pool_check directly) exactly as they set hil_test's globals today.
import glob
import json
import os
import signal
import subprocess
import sys
from pathlib import Path
verbose = False
build_dir = 'cmake-build'
```
Then MOVE (cut from `hil_test.py`, paste unchanged, in this order):
1. `CMD_TIMEOUT = int(os.getenv('HIL_CMD_TIMEOUT', '180'))` (from the constants block; leave `POOL_TIMEOUT`/`SERIAL_*_TIMEOUT` in hil_test)
2. `def cmd_stdout_text(out)`
3. `OPENCOD_ADI_PATH = Path.home() / 'app' / 'openocd_adi'` and `TINYUSB_ROOT = Path(__file__).resolve().parents[2]`
4. `def get_serial_dev(id, vendor_str, product_str, ifnum)`
5. `def run_cmd(cmd, cwd=None, timeout=CMD_TIMEOUT)`
6. All ten `flash_*`/`reset_*` pairs listed in Interfaces, in current file order
7. `def find_firmware(variant, example)`
- [ ] **Step 2: Delete the moved code from `hil_test.py` and qualify call sites**
In `hil_test.py`: add `import hil_flash` under the existing imports; delete the moved definitions and the `build_dir = 'cmake-build'` global (line ~165) plus `global build_dir` in `main`. Repoint every use, all module-qualified:
- `globals()[f'flash_{...}']` → `getattr(hil_flash, f'flash_{...}')` (1 site, in `test_example`)
- `globals()[f'reset_{...}']` → `getattr(hil_flash, f'reset_{...}')` (3 sites: `test_host_device_info`, `test_host_cdc_msc_hid`, `test_host_msc_file_explorer`)
- bare `run_cmd(` → `hil_flash.run_cmd(` ; `cmd_stdout_text(` → `hil_flash.cmd_stdout_text(` ; `find_firmware(` → `hil_flash.find_firmware(` ; `get_serial_dev(` → `hil_flash.get_serial_dev(` ; `TINYUSB_ROOT` → `hil_flash.TINYUSB_ROOT` (in `build_board`, `CONTROLLER_CACHE` stays hil_test-local)
- In `main()`: `build_dir = args.build_dir` → `hil_flash.build_dir = args.build_dir`; where `verbose` is set, add `hil_flash.verbose = args.verbose` (hil_test keeps its own `verbose` for test-side prints)
- `run_cmd`'s `elif verbose:` branch now reads `hil_flash.verbose` (it moved with the function — verify it references the module-local name, not hil_test's)
Find every remaining call site mechanically:
Run: `grep -nE 'run_cmd|cmd_stdout_text|find_firmware|get_serial_dev|flash_[a-z]|reset_[a-z]|TINYUSB_ROOT|OPENCOD' test/hil/hil_test.py | grep -v hil_flash`
Expected: only hits inside comments/strings and the `reset_{flasher}` dispatch f-strings already qualified.
- [ ] **Step 3: Repoint pool_check's flash imports**
In `.claude/skills/hil/pool_check.py`: add `import hil_flash` next to `import hil_test`; replace `hil_test.find_firmware` → `hil_flash.find_firmware` (3 sites), `hil_test.cmd_stdout_text` → `hil_flash.cmd_stdout_text`, `hil_test.get_serial_dev` → `hil_flash.get_serial_dev`, `hil_test.TINYUSB_ROOT` → `hil_flash.TINYUSB_ROOT`, `hil_test.build_dir` → `hil_flash.build_dir` (2 sites incl. `main`'s assignment), `hil_test.verbose = args.verbose` → `hil_flash.verbose = args.verbose`, `getattr(hil_test, f'flash_...')`/`getattr(hil_test, f'reset_...')` → `getattr(hil_flash, ...)` (4 sites). Keep `import hil_test` and the pymtp shim for now (locks still live there; removed in Task 2).
- [ ] **Step 4: Add hil_flash.py to the hil_ci.sh scp list**
```bash
scp -q "$ROOT_DIR/test/hil/hil_test.py" \
"$ROOT_DIR/test/hil/hil_flash.py" \
"$ROOT_DIR/test/hil/pymtp.py" \
"$CONFIG" \
"$REMOTE:$REMOTE_DIR/test/hil/"
```
- [ ] **Step 5: Verify**
Run: `python3 -m py_compile test/hil/hil_flash.py test/hil/hil_test.py .claude/skills/hil/pool_check.py && python3 test/hil/hil_test.py --help >/dev/null && python3 .claude/skills/hil/pool_check.py --scan-only`
Expected: compiles; help prints nothing to stderr; scan-only prints the table and exits 0.
- [ ] **Step 6: Commit**
```bash
git add test/hil/hil_flash.py test/hil/hil_test.py test/hil/hil_ci.sh .claude/skills/hil/pool_check.py
git commit -m "hil: extract flashing into hil_flash.py"
```
---
### Task 2: Create hil_lock.py core (flock protocol + controller permits); repoint hil_test + pool_check
**Files:**
- Create: `test/hil/hil_lock.py`
- Modify: `test/hil/hil_test.py`
- Modify: `.claude/skills/hil/pool_check.py`
- Modify: `test/hil/hil_ci.sh`
**Interfaces:**
- Produces: module `hil_lock` with `BOARD_LOCK_DIR`, `CI_REASON = 'hil_test.py'`, `lock_path(board)`, `flock_nb(board)`, `write_record(fh, reason)`, `clear_record(fh)`, `read_record(board)`, `acquire_board_lock(board, reason=CI_REASON)`, `FLASH_PARALLEL`, `USBTEST_PARALLEL`, `CONTROLLER_SLOTS`, `controller_of(uid)`, `controller_slot(pci)`, `controller_permit`, `flash_permit(uid)`, `usbtest_permit(uid)`, `init_scheduling(b_sems, f_sems, cmap, cmeta, hints, log_fn=None)`.
- [ ] **Step 1: Create `test/hil/hil_lock.py` with the flock core**
New code (the protocol, factored from today's three copies — `board_lock.py` `cmd_hold`/`read_info`, `hil_test.acquire_board_lock`, pool_check `lock_board`; behavior identical to `hil_test.acquire_board_lock` for the acquire path):
```python
#!/usr/bin/env python3
# SPDX-License-Identifier: MIT
"""Board locks + controller permits for the TinyUSB HIL rig.
Board locks are kernel flocks in BOARD_LOCK_DIR arbitrating hardware access
between dev sessions and CI's hil_test.py (never stop the actions-runner).
Controller permits are in-process semaphores budgeting flashes and usbtest
batteries per host controller; they have no CLI meaning. The CLI below
(hold/release/status) manages board locks only; it supersedes board_lock.py.
"""
import argparse
import fcntl
import glob
import json
import os
import re
import select
import signal
import sys
import time
BOARD_LOCK_DIR = '/tmp/tinyusb-hil-locks'
CI_REASON = 'hil_test.py' # release-protected holder tag (release refuses to kill it)
PROFILE = os.environ.get('HIL_PROFILE') == '1'
def lock_path(board: str) -> str:
return os.path.join(BOARD_LOCK_DIR, f'{board}.lock')
def flock_nb(board: str):
"""Open-or-create the lock file WITHOUT truncating (a losing racer must not
wipe the winner's record) and take LOCK_EX|LOCK_NB. Returns the open handle;
raises OSError when the flock is held elsewhere (handle already closed)."""
fd = os.open(lock_path(board), os.O_RDWR | os.O_CREAT, 0o666)
fh = os.fdopen(fd, 'r+')
try:
fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
except OSError:
fh.close()
raise
return fh
def write_record(fh, reason: str) -> None:
"""Best-effort holder record; the flock itself is already held."""
try:
fh.truncate(0)
fh.seek(0)
json.dump({'pid': os.getpid(), 'reason': reason,
'since': time.strftime('%Y-%m-%dT%H:%M:%S%z')}, fh)
fh.flush()
except OSError:
pass
def clear_record(fh) -> None:
"""Clear our record before dropping the flock so records stay truthful."""
try:
fh.truncate(0)
except OSError:
pass
def read_record(board: str):
try:
with open(lock_path(board)) as f:
return json.load(f)
except (OSError, ValueError):
return None
```
Then MOVE `acquire_board_lock` from `hil_test.py` verbatim, with exactly two mechanical edits: signature becomes `def acquire_board_lock(board_name, reason=CI_REASON):` and the record-write dict's `'reason': 'hil_test.py'` becomes `'reason': reason`. Do NOT rewrite its body in terms of `flock_nb` — on conflict it reads holder info from the still-open handle before closing, which `flock_nb` (closes on conflict) cannot provide; the fail-open warning text and RuntimeError message must survive character-for-character.
- [ ] **Step 2: Move the controller-permit block into `hil_lock.py`**
MOVE verbatim from `hil_test.py`: the scheduling comment block + `FLASH_PARALLEL`, `USBTEST_PARALLEL`, `CONTROLLER_SLOTS`, the five module globals (`usbtest_sems`, `flash_sems`, `controller_map`, `controller_meta`, `controller_hints`), `controller_of`, `controller_slot`, `controller_permit`, `flash_permit`, `usbtest_permit`. Two mechanical adaptations:
- add at module scope `log = print` and a setter, replacing the two `log_line(...)` calls inside `controller_of`/`controller_permit` with `log(...)`:
```python
log = print # hil_test.init_worker points this at log_line via init_scheduling
def init_scheduling(b_sems, f_sems, cmap, cmeta, hints, log_fn=None):
"""Install per-worker scheduling state (called from hil_test.init_worker)."""
global usbtest_sems, flash_sems, controller_map, controller_meta, controller_hints, log
usbtest_sems, flash_sems = b_sems, f_sems
controller_map, controller_meta, controller_hints = cmap, cmeta, hints
if log_fn is not None:
log = log_fn
```
- `PROFILE` inside `controller_permit` now resolves to hil_lock's own module constant (defined in Step 1).
- [ ] **Step 3: Repoint `hil_test.py`**
Add `import hil_lock`. Delete the moved lock + permit code and the five globals. `init_worker` keeps its exact signature and initargs; its body sets the hil_test globals it still owns (`print_lock`, `shuffle_seed`) and forwards the rest:
```python
def init_worker(lock, seed, b_mutexes, f_sems, cmap, cmeta, hints_by_uid):
global print_lock, shuffle_seed
print_lock = lock
shuffle_seed = seed
hil_lock.init_scheduling(b_mutexes, f_sems, cmap, cmeta, hints_by_uid, log_fn=log_line)
```
Qualify remaining uses: `acquire_board_lock(name)` → `hil_lock.acquire_board_lock(name)` (in `test_board`), `flash_permit(` → `hil_lock.flash_permit(`, `usbtest_permit(` → `hil_lock.usbtest_permit(`, and `main()`'s startup log line + Semaphore construction read `hil_lock.FLASH_PARALLEL`/`hil_lock.USBTEST_PARALLEL`/`hil_lock.CONTROLLER_SLOTS`. `controller_map` reads in the hint-persistence block of `main` use the Manager dict it already holds locally (`cmap`) — no hil_lock global access there; verify.
- [ ] **Step 4: Repoint pool_check to hil_lock and drop its private copies + hil_test import**
In `pool_check.py`: replace `lock_board`/`unlock_board` bodies with the shared core —
```python
import hil_lock
def lock_board(name: str):
try:
fh = hil_lock.flock_nb(name)
except OSError:
info = hil_lock.read_record(name)
return json.dumps(info) if info else 'unknown holder'
hil_lock.write_record(fh, 'pool_check')
return fh
def unlock_board(fh) -> None:
hil_lock.clear_record(fh)
fh.close()
```
(Behavior note: `lock_board` currently returns the raw record text; JSON-dumping the parsed record is equivalent for display. `hil_lock.BOARD_LOCK_DIR` replaces `hil_test.BOARD_LOCK_DIR`; `os.makedirs(...)` call stays, now on `hil_lock.BOARD_LOCK_DIR`.) Then delete `import hil_test` and the pymtp stub block (`try: import pymtp ... sys.modules['pymtp'] = ...`) — pool_check now imports only `hil_lock` + `hil_flash`.
Run: `grep -n 'hil_test' .claude/skills/hil/pool_check.py`
Expected: only the docstring mention of the protocol/history, no code references (update the docstring's "imports test/hil/hil_test.py" line to name hil_lock/hil_flash).
- [ ] **Step 5: Add hil_lock.py to the hil_ci.sh scp list** (same block as Task 1 Step 4, one more line: `"$ROOT_DIR/test/hil/hil_lock.py" \`)
- [ ] **Step 6: Verify**
Run: `python3 -m py_compile test/hil/hil_lock.py test/hil/hil_test.py .claude/skills/hil/pool_check.py && python3 test/hil/hil_test.py --help >/dev/null && python3 .claude/skills/hil/pool_check.py --scan-only`
Expected: clean compile, working scan table, exit 0.
- [ ] **Step 7: Commit**
```bash
git add test/hil/hil_lock.py test/hil/hil_test.py test/hil/hil_ci.sh .claude/skills/hil/pool_check.py
git commit -m "hil: extract board locks and controller permits into hil_lock.py"
```
---
### Task 3: Absorb board_lock.py CLI into hil_lock.py; delete board_lock.py; rename in docs
**Files:**
- Modify: `test/hil/hil_lock.py` (append CLI)
- Delete: `test/hil/board_lock.py`
- Modify: `.claude/skills/hil/SKILL.md`, `.claude/agents/hil-operator.md`, `.claude/agents/target-debugger.md`, `.claude/skills/etm-trace/SKILL.md`, `.claude/skills/usb-kernel-recover/SKILL.md`, `.claude/skills/target-debug/SKILL.md`
**Interfaces:**
- Produces: `python3 test/hil/hil_lock.py hold|release|status` — identical subcommands, flags, output, and exit codes to today's `board_lock.py`.
- [ ] **Step 1: Move the CLI from `board_lock.py` into `hil_lock.py`**
MOVE verbatim to the end of `hil_lock.py`: `boards_from_config`, `is_locked`, `cmd_hold`, `cmd_release`, `cmd_status`, `main()`, and the `if __name__ == '__main__':` guard. Mechanical adaptations only:
- `LOCK_DIR` → `BOARD_LOCK_DIR` (all sites), `lock_path` already exists (delete the duplicate), `read_info` → `read_record` (all sites; delete the duplicate definition)
- `cmd_hold`'s holder loop body (the open/flock/json.dump block) becomes `fh = flock_nb(b)` + `write_record(fh, reason)` inside the existing try/except OSError
- `_bow_out`'s per-handle truncate loop becomes `clear_record(h)` per handle
- `cmd_release`'s probe uses `flock_nb(b)` in a try/except OSError (held → existing record/victim logic, with the literal `'hil_test.py'` comparison becoming `CI_REASON`); the free-path truncate becomes `clear_record(fh)`
- `main()`'s module docstring reference for `--help` text: keep the usage lines, updating the tool name to `hil_lock.py`
Then delete `test/hil/board_lock.py` (`git rm test/hil/board_lock.py`).
- [ ] **Step 2: Rename `board_lock.py` → `hil_lock.py` in the six live docs**
Run: `cd <worktree> && sed -i 's/board_lock\.py/hil_lock.py/g' .claude/skills/hil/SKILL.md .claude/agents/hil-operator.md .claude/agents/target-debugger.md .claude/skills/etm-trace/SKILL.md .claude/skills/usb-kernel-recover/SKILL.md .claude/skills/target-debug/SKILL.md`
Then: `grep -rn 'board_lock' .claude/ test/ --include='*.md' --include='*.py' --include='*.sh'`
Expected: zero hits outside `docs/superpowers/` history (which stays untouched).
- [ ] **Step 3: Verify CLI behavior end-to-end**
```bash
python3 test/hil/hil_lock.py status # expect: no locks (or current holders)
python3 test/hil/hil_lock.py hold stm32f072disco --reason "split test" &
sleep 1
python3 test/hil/hil_lock.py status # expect: stm32f072disco: {... 'reason': 'split test' ...}
python3 test/hil/hil_lock.py hold stm32f072disco --reason "rival" || echo "conflict OK" # expect: ERROR ... locked + conflict OK
python3 test/hil/hil_lock.py release stm32f072disco # expect: released holder pid NNN
python3 test/hil/hil_lock.py status # expect: no locks
```
Also verify CI-holder protection: create a fake record `echo '{"pid": 1, "reason": "hil_test.py"}' > /tmp/tinyusb-hil-locks/faketest.lock` — since pid 1 holds no flock, `release faketest` must clear the stale record without printing the mid-test error; then `rm -f /tmp/tinyusb-hil-locks/faketest.lock`.
- [ ] **Step 4: Commit**
```bash
git add -A test/hil .claude
git commit -m "hil: fold board_lock CLI into hil_lock.py, retire board_lock.py"
```
---
### Task 4: Rig verification + pre-commit
**Files:** none new (fixes only if verification fails)
- [ ] **Step 1: pool_check flash path on one board**
Run: `python3 .claude/skills/hil/pool_check.py -b stm32f407disco`
Expected: `✅ dfu_runtime ✅ cafe:...`, exit 0.
- [ ] **Step 2: Capture a pre-refactor baseline report**
Run: `cd /home/hathach/code/tinyusb && python3 test/hil/hil_test.py -b stm32f407disco -B examples test/hil/tinyusb.json && cp hil_report.md /tmp/claude-1000/-home-hathach-code-tinyusb/*/scratchpad/hil_report_master.md`
(Primary checkout = pre-refactor code but same rig/config; its working tree already carries the new probe uids.)
- [ ] **Step 3: Run the same board from the worktree and diff the report shape**
Run: `cd .claude/worktrees/hil-test-split && python3 test/hil/hil_test.py -b stm32f407disco -B /home/hathach/code/tinyusb/examples test/hil/tinyusb.json && diff <(sed 's/[0-9.]*s//g;s/[0-9.]* [kMG]B\/s//g' hil_report.md) <(sed 's/[0-9.]*s//g;s/[0-9.]* [kMG]B\/s//g' /tmp/claude-1000/-home-hathach-code-tinyusb/*/scratchpad/hil_report_master.md)`
Expected: empty diff after stripping timings/speeds. Note: `-B` accepts the absolute path so the worktree run reuses the primary checkout's built firmware; `find_firmware` resolves `TINYUSB_ROOT/<build_dir>` and an absolute `-B` overrides relative rooting — if it does not (Path join semantics), instead symlink `ln -s /home/hathach/code/tinyusb/examples/cmake-build-stm32f407disco examples/cmake-build-stm32f407disco` in the worktree and use `-B examples`.
- [ ] **Step 4: pre-commit + final grep hygiene**
Run: `pre-commit run --files test/hil/hil_test.py test/hil/hil_lock.py test/hil/hil_flash.py test/hil/hil_ci.sh .claude/skills/hil/pool_check.py $(git diff --name-only HEAD~3 -- '*.md')`
Expected: all hooks pass.
- [ ] **Step 5: Commit any verification fixes**
```bash
git add -A && git commit -m "hil: post-split verification fixes" # only if Steps 1-4 required changes
```

141
docs/superpowers/specs/2026-07-28-hil-test-refactor-design.md Normal file
View File

@ -0,0 +1,141 @@
# hil_test.py refactor: test core + infra helpers
**Date:** 2026-07-28
**Branch:** `claude/hil-test-split` (based on `claude/hil-pool-check`, which adds `pool_check.py`)
## Motivation
`test/hil/hil_test.py` is 2370 lines mixing five concerns: board-lock protocol, per-controller
scheduling permits, flash/reset backends, the actual per-example tests, and orchestration/report/CLI.
The lock protocol additionally exists in three copies (`hil_test.py`, `board_lock.py`,
`.claude/skills/hil/pool_check.py`), which has already produced drift (pool_check's copy lacks
hil_test's fail-open and error guards). Splitting the infrastructure out makes `hil_test.py`
test-focused and gives external tools (pool_check) one canonical import for locks, permits, and
flashing.
## Goal / non-goals
**Goal:** behavior-preserving code motion. `hil_test.py`'s CLI, arguments, output, report format,
and runtime behavior stay byte-identical. One deliberate user-visible change: the operator lock CLI
moves from `board_lock.py` to `hil_lock.py` (same subcommands, same behavior); `board_lock.py` is
deleted.
**Non-goals (explicit follow-ups, not this change):**
- The 15 pool_check findings from the 2026-07-28 code review (exception isolation, park-on-failure,
espressif coverage, probe-recovery criterion, etc.).
- pool_check adopting `flash_permit` controller budgeting (enabled by this split).
- Any change to lock semantics, permit widths, flash behavior, or test logic.
## Resulting layout (`test/hil/`)
| File | ~Lines | Role |
|---|---|---|
| `hil_test.py` | 1600 | tests + orchestration + report + CLI (unchanged interface) |
| `hil_lock.py` (new) | 420 | board-lock protocol + controller permits + operator CLI |
| `hil_flash.py` (new) | 250 | `run_cmd` + flash/reset backends + `find_firmware` |
| `board_lock.py` | deleted | superseded by `hil_lock.py` |
Import graph: `hil_test` → {`hil_lock`, `hil_flash`}; the helpers import nothing local (no cycles).
`pool_check.py` imports all three.
## hil_lock.py
Docstring states the scope: board locks + controller flash/battery permits; the CLI manages board
locks only (permits are in-process semaphores with no CLI meaning).
**Flock core** (protocol defined once; moved from `board_lock.py`/`hil_test.py`):
- `BOARD_LOCK_DIR = '/tmp/tinyusb-hil-locks'`, `lock_path(board)`
- `CI_REASON = 'hil_test.py'` — the release-protected holder tag (release refuses to kill it)
- `flock_nb(board) -> fh` — `os.open(O_RDWR|O_CREAT, 0o666)` **without O_TRUNC** (a losing racer
must not wipe the winner's record), `fdopen('r+')`, `LOCK_EX|LOCK_NB`; raises `OSError` when held
- `write_record(fh, reason)` — truncate+seek+`json.dump({pid, reason, since})`+flush
- `clear_record(fh)` — truncate(0), swallow OSError (records stay truthful on release)
- `read_record(board) -> dict | None` — today's `board_lock.read_info`
- `acquire_board_lock(board, reason=CI_REASON) -> fh | None` — today's `hil_test.acquire_board_lock`
with a `reason` parameter: `HIL_NO_BOARD_LOCK=1` bypass, fail-open with warning on lock-dir
OSError, `RuntimeError` carrying holder info on conflict
**Controller permits** (moved verbatim from `hil_test.py`):
- `FLASH_PARALLEL`, `USBTEST_PARALLEL`, `CONTROLLER_SLOTS` (env-overridable as today)
- `controller_of(uid)`, `controller_slot(pci)`, `controller_permit`, `flash_permit(uid)`,
`usbtest_permit(uid)`
- Per-worker globals (`usbtest_sems`, `flash_sems`, `controller_map`, `controller_meta`,
`controller_hints`) set by a new `init_scheduling(sems, fsems, cmap, cmeta, hints)` hook that
`hil_test.init_worker` calls from the Pool initializer. `controller_permit`'s PROFILE logging
calls back through a module-level `log = print`-style hook that `hil_test` points at `log_line`
during `init_scheduling` (keeps helpers free of hil_test imports). The `PROFILE` env flag
(`HIL_PROFILE=1`) is read independently in `hil_lock` at import, same derivation as today.
**Operator CLI** (moved verbatim from `board_lock.py`): `hold`/`release`/`status` subcommands with
the daemon-holder machinery (double-fork, setsid, stdio detach, success pipe, SIGTERM bow-out),
release policy (probe the flock; protect `CI_REASON` holders; SIGTERM other recorded pids),
`is_locked` pid-liveness, `--all`/`--config` roster handling. The hold/release/status internals
switch to the flock-core helpers above; observable behavior unchanged.
## hil_flash.py
Moved verbatim from `hil_test.py`:
- `CMD_TIMEOUT` (env-overridable), `run_cmd(cmd, cwd, timeout)`, `cmd_stdout_text(out)`
- `OPENCOD_ADI_PATH`, `TINYUSB_ROOT`
- All backends: `flash_jlink`/`reset_jlink`, `flash_stlink`/`reset_stlink`,
`flash_stflash`/`reset_stflash`, `flash_openocd`/`reset_openocd`,
`flash_openocd_wch`/`reset_openocd_wch`, `flash_openocd_adi`/`reset_openocd_adi`,
`flash_wlink_rs`/`reset_wlink_rs`, `flash_esptool`/`reset_esptool`,
`flash_uniflash`/`reset_uniflash`, `flash_lm4flash`/`reset_lm4flash`
- `find_firmware(variant, example)`
- `get_serial_dev(id, vendor_str, product_str, ifnum)` — moves here (not hil_test) because
`flash_esptool` calls it; keeping it test-side would create a helper→hil_test import cycle.
Tests call `hil_flash.get_serial_dev`.
- Module globals `build_dir = 'cmake-build'` and `verbose = False`, set by callers exactly as the
`hil_test` globals are today (`hil_test.main` sets them from argparse; pool_check sets them
directly). `run_cmd`'s verbose echo reads `hil_flash.verbose`.
Dispatch in callers stays string-based: `getattr(hil_flash, f'flash_{flasher["name"].lower()}')`.
## hil_test.py (what remains)
Config TypedDicts (`Board`, `FlasherCfg`, …), device-node lookup except `get_serial_dev`
(`get_disk_dev`, `get_hid_dev`, `get_alsa_capture_dev`, `open_serial_dev`, `serial_write_all`,
`read_disk_file`, `open_mtp_dev`, `get_printer_dev`/`open_printer_dev`), enum-timeout globals +
`wait_until`,
`log_line`/print-lock, `compact_output`, all `test_*` functions, test lists, `test_example`,
`build_board`, `test_board`, report rendering/accumulation, `main`. Call sites use explicit
module-qualified names (`hil_lock.flash_permit(...)`, `hil_flash.run_cmd(...)`) so provenance is
greppable; no `from … import *`-style mirroring.
`init_worker` keeps its signature (Pool initargs unchanged) and forwards the scheduling state to
`hil_lock.init_scheduling(...)`.
## Consumer updates (same commit)
- **`.claude/skills/hil/pool_check.py`** — drop its private `lock_board`/`unlock_board` in favor of
`hil_lock.flock_nb` + `write_record(fh, 'pool_check')` (+ `clear_record` on release; deliberately NOT `acquire_board_lock`, whose HIL_NO_BOARD_LOCK bypass and fail-open behavior pool_check must not inherit); import
flashers/`find_firmware`/`get_serial_dev`/`cmd_stdout_text`/`TINYUSB_ROOT`/`build_dir` from
`hil_flash`; `BOARD_LOCK_DIR` references move to `hil_lock`. pool_check then imports **only**
`hil_lock` + `hil_flash` (no `hil_test`), so its `pymtp` stub shim is deleted — that shim existed
solely because importing `hil_test` pulls in libmtp.
- **`test/hil/hil_ci.sh`** — the scp list is currently `hil_test.py`, `pymtp.py`, `$CONFIG`; add
`hil_lock.py` and `hil_flash.py` (hil_test cannot even import without them). `board_lock.py` was
never in the list.
- **Docs rename `board_lock.py` → `hil_lock.py`** (live docs only): `.claude/skills/hil/SKILL.md`,
`.claude/agents/hil-operator.md`, `.claude/agents/target-debugger.md`,
`.claude/skills/etm-trace/SKILL.md`, `.claude/skills/usb-kernel-recover/SKILL.md`,
`.claude/skills/target-debug/SKILL.md`. Historical `docs/superpowers/{plans,specs}` stay as
records.
- **CI workflow** — untouched (invokes `hil_test.py` CLI only).
## Verification
1. `python3 -m py_compile` on all three modules + pool_check.
2. `hil_lock.py hold/status/release` interplay: hold, conflicting hold, status listing, release,
protection of a `CI_REASON` record, stale-record cleanup.
3. `pool_check.py --scan-only`, then a single flash board (e.g. `-b stm32f407disco`).
4. Full `hil_test.py -b stm32f407disco -B examples tinyusb.json` on the rig; compare the report
row and log shape against a pre-refactor run.
5. `pre-commit run` on all touched files.
## Sequencing
Lands on top of `claude/hil-pool-check`. After merge, fix the pool_check review findings as a
separate change on the new module boundaries, and update agent-memory references to
`board_lock.py`.

254
test/hil/board_lock.py
View File

@ -1,254 +0,0 @@
#!/usr/bin/env python3
"""Per-board advisory locks for the HIL rig.
Arbitrates board access between dev sessions and CI's hil_test.py without
stopping the actions-runner. Locks are kernel flocks: the kernel releases
them automatically when the holder process dies, and holders clear their
lock-file record on release so records stay truthful (/tmp also clears on
reboot).
Usage:
board_lock.py hold BOARD [BOARD...] --reason TEXT
board_lock.py hold --all [--config CONFIG.json] --reason TEXT
board_lock.py release BOARD [BOARD...] | release --all
board_lock.py status
A holder process holds ALL boards given in one `hold` call; releasing any of
them kills that holder and releases all of its boards.
"""
import argparse
import fcntl
import json
import os
import select
import signal
import sys
import time
LOCK_DIR = '/tmp/tinyusb-hil-locks'
def lock_path(board: str) -> str:
return os.path.join(LOCK_DIR, f'{board}.lock')
def boards_from_config(config: str) -> list:
try:
with open(config) as f:
return [b['name'] for b in json.load(f)['boards']]
except (OSError, ValueError, KeyError) as e:
print(f'ERROR: cannot read board roster {config}: {e}', file=sys.stderr)
sys.exit(1)
def read_info(board: str):
try:
with open(lock_path(board)) as f:
return json.load(f)
except (OSError, ValueError):
return None
def is_locked(board: str) -> bool:
"""True if the recorded holder process is still alive.
Deliberately never touches the flock: even a momentary probe lock would
make a concurrent acquirer's LOCK_NB attempt fail spuriously. The flock
taken by acquirers themselves stays the only authority."""
info = read_info(board)
pid = info.get('pid') if isinstance(info, dict) else None
if not isinstance(pid, int) or pid <= 0:
return False
try:
os.kill(pid, 0)
except ProcessLookupError:
return False
except PermissionError:
return True # alive but owned by another user (e.g. the CI runner)
return True
def cmd_hold(boards, reason):
os.makedirs(LOCK_DIR, exist_ok=True)
# No pre-check: the holder's own LOCK_NB flock is the only authority — a
# recorded pid may be stale or recycled (e.g. a live hil_test.py worker
# that already released this board's flock but not its record).
# The holder signals success through this pipe. A generic is_locked()
# poll would be fooled by a RIVAL invocation's flock — only the holder
# itself knows whether it won every board.
r_fd, w_fd = os.pipe()
pid = os.fork()
if pid > 0:
os.close(w_fd)
os.waitpid(pid, 0) # reap intermediate child
ready, _, _ = select.select([r_fd], [], [], 10)
ok = bool(ready) and os.read(r_fd, 1) == b'1'
os.close(r_fd)
if ok:
print(f'held: {", ".join(boards)}')
return 0
for b in boards:
info = read_info(b)
if info:
print(f'ERROR: {b} locked: {info}', file=sys.stderr)
print('ERROR: holder failed to acquire locks', file=sys.stderr)
return 1
# intermediate child: detach, then spawn the actual holder
os.setsid()
if os.fork() > 0:
os._exit(0)
# holder (grandchild): acquire all flocks, signal the parent, sleep until killed
os.close(r_fd)
# Keep the success pipe clear of fds 0-2: invoked with stdio closed,
# os.pipe() can land there and the dup2 loop below would clobber it.
if w_fd <= 2:
w_fd = fcntl.fcntl(w_fd, fcntl.F_DUPFD, 3)
# Detach stdio: a `hold` whose output is captured must see EOF when the
# front-end exits — the immortal holder must not keep that pipe open.
devnull = os.open(os.devnull, os.O_RDWR)
for std_fd in (0, 1, 2):
os.dup2(devnull, std_fd)
if devnull > 2:
os.close(devnull)
try:
handles = []
for b in boards:
# O_RDWR without O_TRUNC: never truncate before the flock is
# held — a losing racer must not wipe the winner's holder info.
fd = os.open(lock_path(b), os.O_RDWR | os.O_CREAT, 0o666)
fh = os.fdopen(fd, 'r+')
fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
fh.truncate(0)
fh.seek(0)
json.dump({'pid': os.getpid(), 'reason': reason,
'since': time.strftime('%Y-%m-%dT%H:%M:%S%z')}, fh)
fh.flush()
handles.append(fh)
except OSError:
try:
os.write(w_fd, b'0')
except OSError:
pass
os._exit(1) # lost a race; parent reports the failure
os.write(w_fd, b'1')
os.close(w_fd)
def _bow_out(*_):
# clear the records before dying so read_info/status stay truthful
# (the kernel drops the flocks themselves on exit either way)
for h in handles:
try:
h.truncate(0)
except OSError:
pass
os._exit(0)
signal.signal(signal.SIGTERM, _bow_out)
while True:
signal.pause()
def cmd_release(boards):
rc = 0
victims = set()
for b in boards:
try:
fd = os.open(lock_path(b), os.O_RDWR)
except OSError:
continue # no lock file (or another user's): nothing we can release
fh = os.fdopen(fd, 'r+')
try:
fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
except OSError:
# flock genuinely held — never SIGTERM on a mere pid record: the
# pid may be recycled, or a live worker that already moved on.
fh.close()
info = read_info(b) or {}
pid = info.get('pid')
if info.get('reason') == 'hil_test.py':
print(f'ERROR: {b} is mid-test by hil_test.py (pid {pid}) — not killing a '
'CI run; wait for it to finish', file=sys.stderr)
rc = 1
elif isinstance(pid, int) and pid > 0:
victims.add(pid)
else:
print(f'ERROR: {b} is held but its record is unreadable', file=sys.stderr)
rc = 1
continue
# flock was free: only a stale record remained — clear it
try:
fh.truncate(0)
except OSError:
pass
fh.close()
for holder in sorted(victims):
try:
os.kill(holder, signal.SIGTERM)
print(f'released holder pid {holder}')
except ProcessLookupError:
pass
except PermissionError:
print(f'ERROR: holder pid {holder} belongs to another user — cannot signal it',
file=sys.stderr)
rc = 1
time.sleep(0.3)
still = [b for b in boards if is_locked(b)]
if still:
print(f'ERROR: still locked: {", ".join(still)}', file=sys.stderr)
return 1
return rc
def cmd_status():
if not os.path.isdir(LOCK_DIR):
print('no locks')
return 0
any_locked = False
for fn in sorted(os.listdir(LOCK_DIR)):
if not fn.endswith('.lock'):
continue
b = fn[:-5]
if is_locked(b):
any_locked = True
print(f'{b}: {read_info(b)}')
if not any_locked:
print('no locks')
return 0
def main():
ap = argparse.ArgumentParser(description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter)
sub = ap.add_subparsers(dest='cmd', required=True)
p_hold = sub.add_parser('hold')
p_hold.add_argument('boards', nargs='*')
p_hold.add_argument('--all', action='store_true')
p_hold.add_argument('--config',
default=os.path.join(os.path.dirname(os.path.abspath(__file__)),
'tinyusb.json'),
help='board roster JSON (default: tinyusb.json beside this script)')
p_hold.add_argument('--reason', required=True)
p_rel = sub.add_parser('release')
p_rel.add_argument('boards', nargs='*')
p_rel.add_argument('--all', action='store_true')
sub.add_parser('status')
a = ap.parse_args()
if a.cmd == 'hold':
boards = boards_from_config(a.config) if a.all else a.boards
if not boards:
ap.error('no boards given (name boards or use --all)')
sys.exit(cmd_hold(boards, a.reason))
if a.cmd == 'release':
if a.all:
boards = ([fn[:-5] for fn in os.listdir(LOCK_DIR) if fn.endswith('.lock')]
if os.path.isdir(LOCK_DIR) else [])
else:
boards = a.boards
if not boards:
ap.error('no boards given (name boards or use --all)')
sys.exit(cmd_release(boards))
sys.exit(cmd_status())
if __name__ == '__main__':
main()

9
test/hil/hil_ci.sh
View File

@ -44,15 +44,22 @@ echo "==> Setting up remote $REMOTE:$REMOTE_DIR"
ssh "$REMOTE" bash -s -- "$REMOTE_DIR" <<'REMOTE'
set -e
rm -rf -- "$1"
mkdir -p -- "$1/test/hil" "$1/examples"
# .claude path: usbtest.py's HUNG recovery resolves usb_recover.sh relative to the
# staged repo root — without it, recovery ENOENTs and the wedge is left in place
mkdir -p -- "$1/test/hil" "$1/examples" "$1/.claude/skills/usb-kernel-recover/scripts"
REMOTE
# Copy HIL test script and config
echo "==> Copying test scripts"
scp -q "$ROOT_DIR/test/hil/hil_test.py" \
"$ROOT_DIR/test/hil/hil_flash.py" \
"$ROOT_DIR/test/hil/hil_lock.py" \
"$ROOT_DIR/test/hil/usbtest.py" \
"$ROOT_DIR/test/hil/pymtp.py" \
"$CONFIG" \
"$REMOTE:$REMOTE_DIR/test/hil/"
scp -q "$ROOT_DIR/.claude/skills/usb-kernel-recover/scripts/usb_recover.sh" \
"$REMOTE:$REMOTE_DIR/.claude/skills/usb-kernel-recover/scripts/"
# Copy only firmware binaries (elf/bin/hex) plus esptool metadata
# (config.env + flash_args needed by the esptool flasher), preserving structure

293
test/hil/hil_flash.py Executable file
View File

@ -0,0 +1,293 @@
#!/usr/bin/env python3
# SPDX-License-Identifier: MIT
# Firmware flashing for the TinyUSB HIL rig: run_cmd, one flash_*/reset_* pair per
# flasher type (dispatched by config name via getattr), find_firmware, and the
# fixture serial-port resolver get_serial_dev (here, not hil_test: flash_esptool
# needs it and helpers must not import hil_test).
# Callers set module globals `build_dir` and `verbose` (hil_test.main from argparse,
# pool_check directly) exactly as they set hil_test's globals today.
#
# from __future__ import annotations (below): some moved function signatures use
# type hints (Any, Board) not defined in this module; postponed evaluation (PEP
# 563) keeps those as unevaluated strings so the verbatim-moved defs still load.
from __future__ import annotations
import glob
import json
import os
import signal
import subprocess
from pathlib import Path
verbose = False
build_dir = 'cmake-build'
CMD_TIMEOUT = int(os.getenv('HIL_CMD_TIMEOUT', '180'))
# flasher names (dispatch key, board['flasher']['name'].lower()) whose reset_* is a no-op
RESET_NOOP = {'esptool', 'lm4flash', 'stflash', 'uniflash'}
# extra parents find_firmware ALSO searches after build_dir. Empty by default so
# hil_test's -B stays authoritative (a board missing there must report "Skip (no
# binary)", never silently flash a stale binary from another tree); pool_check
# opts in to cover both standard layouts.
EXTRA_BUILD_DIRS: list = []
def cmd_stdout_text(out: Any) -> str:
if out is None:
return ''
if isinstance(out, bytes):
return out.decode('utf-8', errors='ignore')
return str(out)
# -------------------------------------------------------------
# Path
# -------------------------------------------------------------
OPENCOD_ADI_PATH = Path.home() / 'app' / 'openocd_adi'
TINYUSB_ROOT = Path(__file__).resolve().parents[2]
# get usb serial by id
def get_serial_dev(id, vendor_str, product_str, ifnum):
if vendor_str and product_str:
# known vendor and product
vendor_str = vendor_str.replace(' ', '_')
product_str = product_str.replace(' ', '_')
return f'/dev/serial/by-id/usb-{vendor_str}_{product_str}_{id}-if{ifnum:02d}'
else:
# just use id: mostly for cp210x/ftdi flasher
pattern = f'/dev/serial/by-id/usb-*_{id}-if*'
port_list = glob.glob(pattern)
if len(port_list) == 0:
raise RuntimeError(f'No serial device found for {pattern}')
return port_list[0]
# -------------------------------------------------------------
# Flashing firmware
# -------------------------------------------------------------
def run_cmd(cmd: str, cwd: str | None = None, timeout: int = CMD_TIMEOUT) -> subprocess.CompletedProcess:
popen_kwargs = {
'cwd': cwd,
'shell': True,
'stdout': subprocess.PIPE,
'stderr': subprocess.STDOUT,
'text': True,
'encoding': 'utf-8',
'errors': 'replace',
}
if os.name != 'nt':
# C-level setsid, same process-group semantics as preexec_fn=os.setsid but
# safe when called from threads (pool_check runs flashes from a thread pool)
popen_kwargs['start_new_session'] = True
p = subprocess.Popen(cmd, **popen_kwargs)
try:
out, _ = p.communicate(timeout=timeout)
r = subprocess.CompletedProcess(args=cmd, returncode=p.returncode, stdout=out)
except subprocess.TimeoutExpired as ex:
if os.name != 'nt':
try:
os.killpg(p.pid, signal.SIGKILL)
except ProcessLookupError:
pass
else:
p.kill()
try:
out, _ = p.communicate(timeout=10)
except subprocess.TimeoutExpired: # unkillable (e.g. D-state on wedged USB)
out = None
timeout_out = ex.stdout or out or b''
title = f'COMMAND TIMEOUT ({timeout}s): {cmd}'
print()
if os.getenv('CI'):
print(f"::group::{title}")
print(cmd_stdout_text(timeout_out))
print(f"::endgroup::")
else:
print(title)
print(cmd_stdout_text(timeout_out))
return subprocess.CompletedProcess(args=cmd, returncode=124, stdout=timeout_out)
if r.returncode != 0:
title = f'COMMAND FAILED: {cmd}'
print()
if os.getenv('CI'):
print(f"::group::{title}")
print(cmd_stdout_text(r.stdout))
print(f"::endgroup::")
else:
print(title)
print(cmd_stdout_text(r.stdout))
elif verbose:
print(cmd)
print(cmd_stdout_text(r.stdout))
return r
def flash_jlink(board: Board, firmware: str) -> subprocess.CompletedProcess:
flasher = board['flasher']
script = ['halt', 'r', f'loadfile {firmware}.elf', 'r', 'go', 'exit']
f_jlink = Path(f'{board["name"]}_{Path(firmware).name}.jlink')
with f_jlink.open('w') as f:
f.writelines(f'{s}\n' for s in script)
ret = run_cmd(f'JLinkExe -USB {flasher["uid"]} {flasher["args"]} -if swd -JTAGConf -1,-1 -speed auto -NoGui 1 -ExitOnError 1 -CommandFile {f_jlink}')
f_jlink.unlink(missing_ok=True)
return ret
def reset_jlink(board: Board) -> subprocess.CompletedProcess:
flasher = board['flasher']
script = ['halt', 'r', 'go', 'exit']
f_jlink = Path(f'{board["name"]}_reset.jlink')
if not f_jlink.exists():
with f_jlink.open('w') as f:
f.writelines(f'{s}\n' for s in script)
ret = run_cmd(f'JLinkExe -USB {flasher["uid"]} {flasher["args"]} -if swd -JTAGConf -1,-1 -speed auto -NoGui 1 -ExitOnError 1 -CommandFile {f_jlink}')
return ret
def flash_stlink(board, firmware):
flasher = board['flasher']
return run_cmd(f'STM32_Programmer_CLI --connect port=swd sn={flasher["uid"]} --write {firmware}.elf --go')
def reset_stlink(board):
flasher = board['flasher']
return run_cmd(f'STM32_Programmer_CLI --connect port=swd sn={flasher["uid"]} --rst --go')
def flash_stflash(board, firmware):
flasher = board['flasher']
ret = run_cmd(f'st-flash --serial {flasher["uid"]} write {firmware}.bin 0x8000000')
return ret
def reset_stflash(board):
flasher = board['flasher']
return subprocess.CompletedProcess(args=['dummy'], returncode=0)
def flash_openocd(board, firmware):
flasher = board['flasher']
ret = run_cmd(f'openocd -c "tcl_port disabled" -c "gdb_port disabled" -c "adapter serial {flasher["uid"]}" '
f'{flasher["args"]} -c "init; halt; program {firmware}.elf verify; reset; exit"')
return ret
def reset_openocd(board):
flasher = board['flasher']
ret = run_cmd(f'openocd -c "tcl_port disabled" -c "gdb_port disabled" -c "adapter serial {flasher["uid"]}" '
f'{flasher["args"]} -c "init; reset run; exit"')
return ret
def flash_openocd_wch(board, firmware):
flasher = board['flasher']
ret = run_cmd(f'openocd -c "tcl_port disabled" -c "gdb_port disabled" -c "telnet_port disabled" '
f'-c "adapter serial {flasher["uid"]}" {flasher.get("args", "")} -c "program {firmware}.elf reset exit"')
return ret
def reset_openocd_wch(board):
flasher = board['flasher']
ret = run_cmd(f'openocd -c "tcl_port disabled" -c "gdb_port disabled" -c "telnet_port disabled" '
f'-c "adapter serial {flasher["uid"]}" {flasher.get("args", "")} -c "init; reset run; exit"')
return ret
def flash_openocd_adi(board: Board, firmware: str) -> subprocess.CompletedProcess:
flasher = board['flasher']
openocd = OPENCOD_ADI_PATH / 'src' / 'openocd'
tcl_dir = OPENCOD_ADI_PATH / 'tcl'
ret = run_cmd(f'{openocd} -c "adapter serial {flasher["uid"]}" -s {tcl_dir} '
f'{flasher["args"]} -c "program {firmware}.elf reset exit"')
return ret
def reset_openocd_adi(board: Board) -> subprocess.CompletedProcess:
flasher = board['flasher']
openocd = OPENCOD_ADI_PATH / 'src' / 'openocd'
tcl_dir = OPENCOD_ADI_PATH / 'tcl'
ret = run_cmd(f'{openocd} -c "adapter serial {flasher["uid"]}" -s {tcl_dir} '
f'{flasher["args"]} -c "program reset exit"')
return ret
def flash_wlink_rs(board, firmware):
flasher = board['flasher']
# wlink use index for probe selection and lacking usb serial support
ret = run_cmd(f'wlink flash {firmware}.elf')
return ret
def reset_wlink_rs(board):
flasher = board['flasher']
# wlink use index for probe selection and lacking usb serial support
ret = run_cmd(f'wlink reset')
return ret
def flash_esptool(board: Board, firmware: str) -> subprocess.CompletedProcess:
flasher = board['flasher']
port = get_serial_dev(flasher["uid"], None, None, 0)
fw_dir = Path(f'{firmware}.bin').parent
with (fw_dir / 'config.env').open() as f:
idf_target = json.load(f)['IDF_TARGET']
with (fw_dir / 'flash_args').open() as f:
flash_args = f.read().strip().replace('\n', ' ')
command = (f'esptool --chip {idf_target} -p {port} {flasher["args"]} '
f'--before=default_reset --after=hard_reset write_flash {flash_args}')
ret = run_cmd(command, cwd=str(fw_dir))
return ret
def reset_esptool(board):
flasher = board['flasher']
return subprocess.CompletedProcess(args=['dummy'], returncode=0)
def flash_uniflash(board, firmware):
flasher = board['flasher']
ret = run_cmd(f'dslite.sh {flasher["args"]} -f {firmware}.hex')
return ret
def reset_uniflash(board):
flasher = board['flasher']
return subprocess.CompletedProcess(args=['dummy'], returncode=0)
def flash_lm4flash(board, firmware):
# TI Tiva-C / Stellaris ICDI: lightweight lm4flash, resets and runs after write
flasher = board['flasher']
ret = run_cmd(f'lm4flash -s {flasher["uid"]} {flasher["args"]} {firmware}.bin')
return ret
def reset_lm4flash(board):
# lm4flash has no reset-only mode; it resets+runs on flash, so reset is a no-op
flasher = board['flasher']
return subprocess.CompletedProcess(args=['dummy'], returncode=0)
def find_firmware(variant: str, example: str, roots: list | None = None):
"""Locate a built example's firmware base path (no extension) under
<build_dir>/cmake-build-<variant>/<example>/, then under EXTRA_BUILD_DIRS
(empty unless the caller opts in — see its comment). `roots` overrides that
search list entirely for one call (e.g. to find a build just produced by
tools/build.py in its fixed cmake-build/ layout without widening the global
policy). Accepts the single-config layout (firmware directly in the example
dir) or Ninja Multi-Config (a per-config subdir like RelWithDebInfo/).
Returns the base Path, or None if not built."""
base = Path(example).name
for bd in dict.fromkeys(roots if roots is not None else [build_dir, *EXTRA_BUILD_DIRS]):
fw_dir = TINYUSB_ROOT / bd / f'cmake-build-{variant}' / example
if not fw_dir.is_dir():
continue
for cand in [fw_dir / base, fw_dir / 'RelWithDebInfo' / base,
*(p.with_suffix('') for p in sorted(fw_dir.glob(f'*/{base}.elf')))]:
if cand.with_suffix('.elf').exists() or cand.with_suffix('.bin').exists():
return cand
return None

479
test/hil/hil_lock.py Executable file
View File

@ -0,0 +1,479 @@
#!/usr/bin/env python3
# SPDX-License-Identifier: MIT
"""Board locks + controller permits for the TinyUSB HIL rig.
Board locks are kernel flocks in BOARD_LOCK_DIR arbitrating hardware access
between dev sessions and CI's hil_test.py (never stop the actions-runner).
Controller permits are in-process semaphores budgeting flashes and usbtest
batteries per host controller; they have no CLI meaning. The CLI below
(hold/release/status) manages board locks only.
"""
import argparse
import fcntl
import glob
import json
import os
import re
import select
import signal
import sys
import time
BOARD_LOCK_DIR = '/tmp/tinyusb-hil-locks'
CI_REASON = 'hil_test.py' # release-protected holder tag (release refuses to kill it)
PROTECTED_REASONS = {CI_REASON, 'pool_check'} # cmd_release refuses to SIGTERM these holders
PROFILE = os.environ.get('HIL_PROFILE') == '1'
def lock_path(board: str) -> str:
return os.path.join(BOARD_LOCK_DIR, f'{board}.lock')
def flock_nb(board: str):
"""Open-or-create the lock file WITHOUT truncating (a losing racer must not
wipe the winner's record) and take LOCK_EX|LOCK_NB. Returns the open handle;
raises OSError when the flock is held elsewhere (handle already closed)."""
fd = os.open(lock_path(board), os.O_RDWR | os.O_CREAT, 0o666)
fh = os.fdopen(fd, 'r+')
try:
fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
except OSError:
fh.close()
raise
return fh
def write_record(fh, reason: str) -> bool:
"""Holder record; the flock itself is already held. Returns False on a write
failure — acquire_board_lock stays best-effort (the flock is the authority),
but cmd_hold aborts on it like board_lock.py did (a hold whose record is
missing is invisible to status/release)."""
try:
fh.truncate(0)
fh.seek(0)
json.dump({'pid': os.getpid(), 'reason': reason,
'since': time.strftime('%Y-%m-%dT%H:%M:%S%z')}, fh)
fh.flush()
return True
except OSError:
return False
def clear_record(fh) -> None:
"""Clear our record before dropping the flock so records stay truthful."""
try:
fh.truncate(0)
except OSError:
pass
def read_record(board: str):
try:
with open(lock_path(board)) as f:
return json.load(f)
except (OSError, ValueError):
return None
# --- per-board dev-session locks ------------------------------------------
def acquire_board_lock(board_name, reason=CI_REASON):
"""Take this board's flock for the duration of its flash+test.
Returns an open file handle (keep it referenced; closing releases it),
or None when HIL_NO_BOARD_LOCK=1 or the lock dir is unusable (fail-open:
locking must never break a test run by itself).
Raises RuntimeError only when another session holds the board."""
import fcntl
if os.environ.get('HIL_NO_BOARD_LOCK') == '1':
return None # user-authorized bypass — see hil skill
try:
os.makedirs(BOARD_LOCK_DIR, exist_ok=True)
fd = os.open(os.path.join(BOARD_LOCK_DIR, f'{board_name}.lock'),
os.O_RDWR | os.O_CREAT, 0o666)
fh = os.fdopen(fd, 'r+')
except OSError as e:
# odd lock dir (perms, path collision): proceed unlocked, but say so —
# a silent fail-open is indistinguishable from the intentional bypass
print(f'warning: board lock unavailable for {board_name} ({e}); proceeding unlocked',
flush=True)
return None
try:
fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
except OSError:
try:
info = fh.read(500).strip()
except (OSError, UnicodeDecodeError):
info = ''
fh.close()
raise RuntimeError(f'board locked: {info or "unknown holder"}')
# announce ourselves so the other side's conflict message is truthful;
# best-effort — the flock itself is already held
try:
fh.truncate(0)
fh.seek(0)
json.dump({'pid': os.getpid(), 'reason': reason,
'since': time.strftime('%Y-%m-%dT%H:%M:%S%z')}, fh)
fh.flush()
except OSError:
pass
return fh
# Per-host-controller concurrency (see controller_of/controller_slot below): a usbtest battery
# saturates its DUT's host controller, so batteries and flashes are budgeted per controller.
# - uPD720201 cards need their latest firmware (>= 2.0.2.6; RAM-uploaded, reloads every
# power cycle): ROM firmware dies under battery + re-enumeration churn, and usbtest.py
# refuses the unlink-stress cases on it.
# - widths (profiled 2026-07-13/14): wall time 22.2/14.3/12.5/10.8 min at usbtest width
# 1/2/3/4, plateau after; flash width beyond 8 only adds flasher-hub contention;
# battery case failures start at 12/8 (bandwidth stretch on shared leaf-hub uplinks).
# - a marginal DUT port bouncing during concurrent batteries can wedge/kill a uPD720201
# ("xHCI host not responding to stop endpoint command"): fix the port/cable or pull
# the board, don't lower the widths (2026-07-16: every death traced to one board's port).
FLASH_PARALLEL = int(os.getenv('HIL_FLASH_PARALLEL', '8'))
USBTEST_PARALLEL = int(os.getenv('HIL_USBTEST_PARALLEL', '4'))
CONTROLLER_SLOTS = 12 # lock slots; controllers are assigned to slots on first sight
usbtest_sems = None # CONTROLLER_SLOTS semaphores: per-slot usbtest-battery permits
flash_sems = None # CONTROLLER_SLOTS semaphores: per-slot flash permits
controller_map = None # shared dict: 'pci:<addr>' -> slot, 'uid:<uid>' -> pci addr cache
controller_meta = None # guards slot assignment in controller_map
controller_hints = {} # static uid -> pci from the last run's cache (read-only per worker)
log = print # hil_test.init_worker points this at log_line via init_scheduling
def init_scheduling(b_sems, f_sems, cmap, cmeta, hints, log_fn=None):
"""Install per-worker scheduling state (called from hil_test.init_worker)."""
global usbtest_sems, flash_sems, controller_map, controller_meta, controller_hints, log
usbtest_sems, flash_sems = b_sems, f_sems
controller_map, controller_meta, controller_hints = cmap, cmeta, hints
if log_fn is not None:
log = log_fn
# -------------------------------------------------------------
# Per-controller scheduling
# -------------------------------------------------------------
def controller_of(uid: str):
"""Resolve a DUT uid to its root host controller's PCI address, or None if the device
is not enumerated (e.g. parked in board_test firmware with USB off). Successful
resolutions are cached — cabling does not change mid-run. Dual-port parts (e.g.
CH32V307 usbhs/usbfs variants) share one uid and one cache entry: budgeting is only
exact when both ports sit on the same controller (true on this rig)."""
if controller_map is None:
return None
cached = controller_map.get(f'uid:{uid}')
if cached:
return cached
for f in glob.glob('/sys/bus/usb/devices/*/serial'):
d = os.path.dirname(f)
try:
if open(f).read().strip().lower() != uid.lower():
continue
bus = int(open(os.path.join(d, 'busnum')).read())
root = os.path.realpath(f'/sys/bus/usb/devices/usb{bus}')
m = re.findall(r'[0-9a-f]{4}:[0-9a-f]{2}:[0-9a-f]{2}\.[0-9a-f]', root)
if m:
controller_map[f'uid:{uid}'] = m[-1]
return m[-1]
except (OSError, ValueError):
continue
return None
def controller_slot(pci: str) -> int:
"""Map a controller PCI address to a lock slot (assigned on first sight)."""
key = f'pci:{pci}'
with controller_meta:
slot = controller_map.get(key)
if slot is None:
slot = controller_map.get('nslots', 0)
if slot >= CONTROLLER_SLOTS:
slot = 0 # more controllers than slots: overflow shares slot 0 (safe, over-serialized)
else:
controller_map['nslots'] = slot + 1
controller_map[key] = slot
return slot
class controller_permit:
"""Context manager: one permit from `sems` on the board's controller slot. If the
controller is unknown, fail closed: take one permit from EVERY slot, in order, so the
operation respects the budget wherever it might land. `warn_unknown` logs that fallback
(used by usbtest, where the device is expected to be enumerated by the caller)."""
def __init__(self, sems, uid: str, warn_unknown: bool = False):
self.sems = sems
self.slots = None
self.uid = uid
if sems is None:
return
pci = controller_of(uid)
if pci is None and not warn_unknown:
# last-run cabling hint, flash budgeting only: a mis-budgeted flash is harmless,
# but a battery must never trust a stale hint (it could stack two batteries on
# one controller). In practice only a board's first flash lands here - batteries
# assert enumeration before taking their permit.
pci = controller_hints.get(uid)
if pci is None and warn_unknown:
log(f'warning: cannot resolve {uid} to a host controller; '
'taking a permit on every slot (over-serialized)')
self.slots = [controller_slot(pci)] if pci else list(range(CONTROLLER_SLOTS))
def __enter__(self):
if self.slots:
t0 = time.monotonic()
taken = []
try:
for s in self.slots:
self.sems[s].acquire()
taken.append(s)
# stays inside the try: if this raises (e.g. broken stdout), the permits
# must be released - a failed __enter__ never gets its __exit__
if PROFILE and time.monotonic() - t0 > 1.0:
log(f'[prof] permit wait {time.monotonic() - t0:.1f}s '
f'(uid {self.uid}, slots {self.slots})')
except BaseException:
for s in reversed(taken):
self.sems[s].release()
raise
return self
def __exit__(self, *exc):
if self.slots:
for s in reversed(self.slots):
self.sems[s].release()
return False
def flash_permit(uid: str) -> controller_permit:
return controller_permit(flash_sems, uid)
def usbtest_permit(uid: str) -> controller_permit:
return controller_permit(usbtest_sems, uid, warn_unknown=True)
# --- operator CLI (hold/release/status) ------------------------------------
def boards_from_config(config: str) -> list:
"""All board names, INCLUDING boards-skip: `hold --all` guards rig-wide
operations, and parked boards can still be touched (pool_check -b names them
explicitly), so a rig-wide hold that skipped them would leave a gap."""
try:
with open(config) as f:
cfg = json.load(f)
return [b['name'] for b in cfg['boards'] + cfg.get('boards-skip', [])]
except (OSError, ValueError, KeyError) as e:
print(f'ERROR: cannot read board roster {config}: {e}', file=sys.stderr)
sys.exit(1)
def is_locked(board: str) -> bool:
"""True if the recorded holder process is still alive.
Deliberately never touches the flock: even a momentary probe lock would
make a concurrent acquirer's LOCK_NB attempt fail spuriously. The flock
taken by acquirers themselves stays the only authority."""
info = read_record(board)
pid = info.get('pid') if isinstance(info, dict) else None
if not isinstance(pid, int) or pid <= 0:
return False
try:
os.kill(pid, 0)
except ProcessLookupError:
return False
except PermissionError:
return True # alive but owned by another user (e.g. the CI runner)
return True
def cmd_hold(boards, reason):
os.makedirs(BOARD_LOCK_DIR, exist_ok=True)
# No pre-check: the holder's own LOCK_NB flock is the only authority — a
# recorded pid may be stale or recycled (e.g. a live hil_test.py worker
# that already released this board's flock but not its record).
# The holder signals success through this pipe. A generic is_locked()
# poll would be fooled by a RIVAL invocation's flock — only the holder
# itself knows whether it won every board.
r_fd, w_fd = os.pipe()
pid = os.fork()
if pid > 0:
os.close(w_fd)
os.waitpid(pid, 0) # reap intermediate child
ready, _, _ = select.select([r_fd], [], [], 10)
ok = bool(ready) and os.read(r_fd, 1) == b'1'
os.close(r_fd)
if ok:
print(f'held: {", ".join(boards)}')
return 0
for b in boards:
info = read_record(b)
if info:
print(f'ERROR: {b} locked: {info}', file=sys.stderr)
print('ERROR: holder failed to acquire locks', file=sys.stderr)
return 1
# intermediate child: detach, then spawn the actual holder
os.setsid()
if os.fork() > 0:
os._exit(0)
# holder (grandchild): acquire all flocks, signal the parent, sleep until killed
os.close(r_fd)
# Keep the success pipe clear of fds 0-2: invoked with stdio closed,
# os.pipe() can land there and the dup2 loop below would clobber it.
if w_fd <= 2:
w_fd = fcntl.fcntl(w_fd, fcntl.F_DUPFD, 3)
# Detach stdio: a `hold` whose output is captured must see EOF when the
# front-end exits — the immortal holder must not keep that pipe open.
devnull = os.open(os.devnull, os.O_RDWR)
for std_fd in (0, 1, 2):
os.dup2(devnull, std_fd)
if devnull > 2:
os.close(devnull)
try:
handles = []
for b in boards:
fh = flock_nb(b)
if not write_record(fh, reason):
raise OSError(f'cannot write holder record for {b}')
handles.append(fh)
except OSError:
try:
os.write(w_fd, b'0')
except OSError:
pass
os._exit(1) # lost a race; parent reports the failure
os.write(w_fd, b'1')
os.close(w_fd)
def _bow_out(*_):
# clear the records before dying so read_record/status stay truthful
# (the kernel drops the flocks themselves on exit either way)
for h in handles:
clear_record(h)
os._exit(0)
signal.signal(signal.SIGTERM, _bow_out)
while True:
signal.pause()
def cmd_release(boards):
rc = 0
victims = set()
for b in boards:
try:
fd = os.open(lock_path(b), os.O_RDWR)
except OSError:
continue # no lock file (or another user's): nothing we can release
fh = os.fdopen(fd, 'r+')
try:
fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
except OSError:
# flock genuinely held — never SIGTERM on a mere pid record: the
# pid may be recycled, or a live worker that already moved on.
fh.close()
info = read_record(b) or {}
pid = info.get('pid')
reason = info.get('reason')
if reason in PROTECTED_REASONS:
print(f'ERROR: {b} is mid-test by {reason} (pid {pid}) — not killing it; '
'wait for it to finish', file=sys.stderr)
rc = 1
elif isinstance(pid, int) and pid > 0:
victims.add(pid)
else:
print(f'ERROR: {b} is held but its record is unreadable', file=sys.stderr)
rc = 1
continue
# flock was free: only a stale record remained — clear it
clear_record(fh)
fh.close()
for holder in sorted(victims):
try:
os.kill(holder, signal.SIGTERM)
print(f'released holder pid {holder}')
except ProcessLookupError:
pass
except PermissionError:
print(f'ERROR: holder pid {holder} belongs to another user — cannot signal it',
file=sys.stderr)
rc = 1
time.sleep(0.3)
still = [b for b in boards if is_locked(b)]
if still:
print(f'ERROR: still locked: {", ".join(still)}', file=sys.stderr)
return 1
return rc
def cmd_status():
if not os.path.isdir(BOARD_LOCK_DIR):
print('no locks')
return 0
any_locked = False
for fn in sorted(os.listdir(BOARD_LOCK_DIR)):
if not fn.endswith('.lock'):
continue
b = fn[:-5]
if is_locked(b):
any_locked = True
print(f'{b}: {read_record(b)}')
if not any_locked:
print('no locks')
return 0
_CLI_USAGE = """Per-board advisory locks for the HIL rig.
Arbitrates board access between dev sessions and CI's hil_test.py without
stopping the actions-runner. Locks are kernel flocks: the kernel releases
them automatically when the holder process dies, and holders clear their
lock-file record on release so records stay truthful (/tmp also clears on
reboot).
Usage:
hil_lock.py hold BOARD [BOARD...] --reason TEXT
hil_lock.py hold --all [--config CONFIG.json] --reason TEXT
hil_lock.py release BOARD [BOARD...] | release --all
hil_lock.py status
A holder process holds ALL boards given in one `hold` call; releasing any of
them kills that holder and releases all of its boards.
"""
def main():
ap = argparse.ArgumentParser(description=_CLI_USAGE,
formatter_class=argparse.RawDescriptionHelpFormatter)
sub = ap.add_subparsers(dest='cmd', required=True)
p_hold = sub.add_parser('hold')
p_hold.add_argument('boards', nargs='*')
p_hold.add_argument('--all', action='store_true')
p_hold.add_argument('--config',
default=os.path.join(os.path.dirname(os.path.abspath(__file__)),
'tinyusb.json'),
help='board roster JSON (default: tinyusb.json beside this script)')
p_hold.add_argument('--reason', required=True)
p_rel = sub.add_parser('release')
p_rel.add_argument('boards', nargs='*')
p_rel.add_argument('--all', action='store_true')
sub.add_parser('status')
a = ap.parse_args()
if a.cmd == 'hold':
boards = boards_from_config(a.config) if a.all else a.boards
if not boards:
ap.error('no boards given (name boards or use --all)')
sys.exit(cmd_hold(boards, a.reason))
if a.cmd == 'release':
if a.all:
boards = ([fn[:-5] for fn in os.listdir(BOARD_LOCK_DIR) if fn.endswith('.lock')]
if os.path.isdir(BOARD_LOCK_DIR) else [])
else:
boards = a.boards
if not boards:
ap.error('no boards given (name boards or use --all)')
sys.exit(cmd_release(boards))
sys.exit(cmd_status())
if __name__ == '__main__':
main()

1013
test/hil/hil_pool_check.py Normal file
View File

File diff suppressed because it is too large Load Diff

530
test/hil/hil_test.py
View File

@ -46,10 +46,9 @@ import re
import select
import sys
import time
import signal
from contextlib import redirect_stdout
from pathlib import Path
from typing import Any, TypedDict, NotRequired, cast
from typing import TypedDict, NotRequired, cast
import serial
import subprocess
@ -58,6 +57,9 @@ import glob
import multiprocessing
from multiprocessing import TimeoutError as MpTimeoutError
import hil_flash
import hil_lock
# Raw Lock/Semaphore objects passed via Pool initargs are inheritable only under the fork
# start method (spawn/forkserver pickle them and fail at Pool creation) — pin it so a
# future interpreter default change cannot break the run at startup.
@ -68,51 +70,6 @@ import ctypes
from pymtp import MTP
import string
# --- per-board dev-session locks (see test/hil/board_lock.py) ------------
BOARD_LOCK_DIR = '/tmp/tinyusb-hil-locks'
def acquire_board_lock(board_name):
"""Take this board's flock for the duration of its flash+test.
Returns an open file handle (keep it referenced; closing releases it),
or None when HIL_NO_BOARD_LOCK=1 or the lock dir is unusable (fail-open:
locking must never break a test run by itself).
Raises RuntimeError only when another session holds the board."""
import fcntl
if os.environ.get('HIL_NO_BOARD_LOCK') == '1':
return None # user-authorized bypass — see board_lock.py / hil skill
try:
os.makedirs(BOARD_LOCK_DIR, exist_ok=True)
fd = os.open(os.path.join(BOARD_LOCK_DIR, f'{board_name}.lock'),
os.O_RDWR | os.O_CREAT, 0o666)
fh = os.fdopen(fd, 'r+')
except OSError as e:
# odd lock dir (perms, path collision): proceed unlocked, but say so —
# a silent fail-open is indistinguishable from the intentional bypass
print(f'warning: board lock unavailable for {board_name} ({e}); proceeding unlocked',
flush=True)
return None
try:
fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
except OSError:
try:
info = fh.read(500).strip()
except (OSError, UnicodeDecodeError):
info = ''
fh.close()
raise RuntimeError(f'board locked: {info or "unknown holder"}')
# announce ourselves so the other side's conflict message is truthful;
# best-effort — the flock itself is already held
try:
fh.truncate(0)
fh.seek(0)
json.dump({'pid': os.getpid(), 'reason': 'hil_test.py',
'since': time.strftime('%Y-%m-%dT%H:%M:%S%z')}, fh)
fh.flush()
except OSError:
pass
return fh
# Enumeration wait budget. The first attempt gets ENUM_TIMEOUT; retry attempts get the
# shorter ENUM_TIMEOUT_RETRY - the board was just re-flashed again, and a device that is
# going to enumerate shows up within a few seconds, so a failing test costs ~3-5x a
@ -162,41 +119,16 @@ verbose = False
PROFILE = os.environ.get('HIL_PROFILE') == '1' # timestamped logs + permit/flash timing + ctrl-map dump
test_only = []
board_test = {}
build_dir = 'cmake-build'
skip_flash = False
print_lock = None
shuffle_seed = None # per-run seed for the per-board test-order shuffle (HIL_SHUFFLE_SEED to replay)
# Per-host-controller concurrency (see controller_of/controller_slot below): a usbtest battery
# saturates its DUT's host controller, so batteries and flashes are budgeted per controller.
# - uPD720201 cards need their latest firmware (>= 2.0.2.6; RAM-uploaded, reloads every
# power cycle): ROM firmware dies under battery + re-enumeration churn, and usbtest.py
# refuses the unlink-stress cases on it.
# - widths (profiled 2026-07-13/14): wall time 22.2/14.3/12.5/10.8 min at usbtest width
# 1/2/3/4, plateau after; flash width beyond 8 only adds flasher-hub contention;
# battery case failures start at 12/8 (bandwidth stretch on shared leaf-hub uplinks).
# - a marginal DUT port bouncing during concurrent batteries can wedge/kill a uPD720201
# ("xHCI host not responding to stop endpoint command"): fix the port/cable or pull
# the board, don't lower the widths (2026-07-16: every death traced to one board's port).
FLASH_PARALLEL = int(os.getenv('HIL_FLASH_PARALLEL', '8'))
USBTEST_PARALLEL = int(os.getenv('HIL_USBTEST_PARALLEL', '4'))
CONTROLLER_SLOTS = 12 # lock slots; controllers are assigned to slots on first sight
usbtest_sems = None # CONTROLLER_SLOTS semaphores: per-slot usbtest-battery permits
flash_sems = None # CONTROLLER_SLOTS semaphores: per-slot flash permits
controller_map = None # shared dict: 'pci:<addr>' -> slot, 'uid:<uid>' -> pci addr cache
controller_meta = None # guards slot assignment in controller_map
controller_hints = {} # static uid -> pci from the last run's cache (read-only per worker)
def init_worker(lock, seed, b_mutexes, f_sems, cmap, cmeta, hints_by_uid):
global print_lock, shuffle_seed, usbtest_sems, flash_sems, controller_map, controller_meta, controller_hints
global print_lock, shuffle_seed
print_lock = lock
shuffle_seed = seed
usbtest_sems = b_mutexes
flash_sems = f_sems
controller_map = cmap
controller_meta = cmeta
controller_hints = hints_by_uid
hil_lock.init_scheduling(b_mutexes, f_sems, cmap, cmeta, hints_by_uid, log_fn=log_line)
def log_line(msg: str) -> None:
@ -210,108 +142,6 @@ def log_line(msg: str) -> None:
print(msg, file=out, flush=True)
# -------------------------------------------------------------
# Per-controller scheduling
# -------------------------------------------------------------
def controller_of(uid: str):
"""Resolve a DUT uid to its root host controller's PCI address, or None if the device
is not enumerated (e.g. parked in board_test firmware with USB off). Successful
resolutions are cached — cabling does not change mid-run. Dual-port parts (e.g.
CH32V307 usbhs/usbfs variants) share one uid and one cache entry: budgeting is only
exact when both ports sit on the same controller (true on this rig)."""
if controller_map is None:
return None
cached = controller_map.get(f'uid:{uid}')
if cached:
return cached
for f in glob.glob('/sys/bus/usb/devices/*/serial'):
d = os.path.dirname(f)
try:
if open(f).read().strip().lower() != uid.lower():
continue
bus = int(open(os.path.join(d, 'busnum')).read())
root = os.path.realpath(f'/sys/bus/usb/devices/usb{bus}')
m = re.findall(r'[0-9a-f]{4}:[0-9a-f]{2}:[0-9a-f]{2}\.[0-9a-f]', root)
if m:
controller_map[f'uid:{uid}'] = m[-1]
return m[-1]
except (OSError, ValueError):
continue
return None
def controller_slot(pci: str) -> int:
"""Map a controller PCI address to a lock slot (assigned on first sight)."""
key = f'pci:{pci}'
with controller_meta:
slot = controller_map.get(key)
if slot is None:
slot = controller_map.get('nslots', 0)
if slot >= CONTROLLER_SLOTS:
slot = 0 # more controllers than slots: overflow shares slot 0 (safe, over-serialized)
else:
controller_map['nslots'] = slot + 1
controller_map[key] = slot
return slot
class controller_permit:
"""Context manager: one permit from `sems` on the board's controller slot. If the
controller is unknown, fail closed: take one permit from EVERY slot, in order, so the
operation respects the budget wherever it might land. `warn_unknown` logs that fallback
(used by usbtest, where the device is expected to be enumerated by the caller)."""
def __init__(self, sems, uid: str, warn_unknown: bool = False):
self.sems = sems
self.slots = None
self.uid = uid
if sems is None:
return
pci = controller_of(uid)
if pci is None and not warn_unknown:
# last-run cabling hint, flash budgeting only: a mis-budgeted flash is harmless,
# but a battery must never trust a stale hint (it could stack two batteries on
# one controller). In practice only a board's first flash lands here - batteries
# assert enumeration before taking their permit.
pci = controller_hints.get(uid)
if pci is None and warn_unknown:
log_line(f'warning: cannot resolve {uid} to a host controller; '
'taking a permit on every slot (over-serialized)')
self.slots = [controller_slot(pci)] if pci else list(range(CONTROLLER_SLOTS))
def __enter__(self):
if self.slots:
t0 = time.monotonic()
taken = []
try:
for s in self.slots:
self.sems[s].acquire()
taken.append(s)
# stays inside the try: if this raises (e.g. broken stdout), the permits
# must be released - a failed __enter__ never gets its __exit__
if PROFILE and time.monotonic() - t0 > 1.0:
log_line(f'[prof] permit wait {time.monotonic() - t0:.1f}s '
f'(uid {self.uid}, slots {self.slots})')
except BaseException:
for s in reversed(taken):
self.sems[s].release()
raise
return self
def __exit__(self, *exc):
if self.slots:
for s in reversed(self.slots):
self.sems[s].release()
return False
def flash_permit(uid: str) -> controller_permit:
return controller_permit(flash_sems, uid)
def usbtest_permit(uid: str) -> controller_permit:
return controller_permit(usbtest_sems, uid, warn_unknown=True)
def compact_output(raw: str) -> str:
if not raw:
return ''
@ -365,47 +195,16 @@ class Board(TypedDict):
class HilConfig(TypedDict):
boards: list[Board]
CMD_TIMEOUT = int(os.getenv('HIL_CMD_TIMEOUT', '180'))
POOL_TIMEOUT = int(os.getenv('HIL_POOL_TIMEOUT', '4200')) # usbtest batteries are serialized fleet-wide, lengthening the tail
SERIAL_READ_TIMEOUT = float(os.getenv('HIL_SERIAL_READ_TIMEOUT', '5'))
SERIAL_WRITE_TIMEOUT = float(os.getenv('HIL_SERIAL_WRITE_TIMEOUT', '10'))
def cmd_stdout_text(out: Any) -> str:
if out is None:
return ''
if isinstance(out, bytes):
return out.decode('utf-8', errors='ignore')
return str(out)
MSC_README_TXT = \
b"This is tinyusb's MassStorage Class demo.\r\n\r\n\
If you find any bugs or get any questions, feel free to file an\r\n\
issue at github.com/hathach/tinyusb"
# -------------------------------------------------------------
# Path
# -------------------------------------------------------------
OPENCOD_ADI_PATH = Path.home() / 'app' / 'openocd_adi'
TINYUSB_ROOT = Path(__file__).resolve().parents[2]
# get usb serial by id
def get_serial_dev(id, vendor_str, product_str, ifnum):
if vendor_str and product_str:
# known vendor and product
vendor_str = vendor_str.replace(' ', '_')
product_str = product_str.replace(' ', '_')
return f'/dev/serial/by-id/usb-{vendor_str}_{product_str}_{id}-if{ifnum:02d}'
else:
# just use id: mostly for cp210x/ftdi flasher
pattern = f'/dev/serial/by-id/usb-*_{id}-if*'
port_list = glob.glob(pattern)
if len(port_list) == 0:
raise RuntimeError(f'No serial device found for {pattern}')
return port_list[0]
# get usb disk by id
def get_disk_dev(id, vendor_str, lun):
return f'/dev/disk/by-id/usb-{vendor_str}_Mass_Storage_{id}-0:{lun}'
@ -529,215 +328,13 @@ def open_printer_dev(id: str, vendor_str, product_str, ifnum: int) -> str:
return lp_dev
# -------------------------------------------------------------
# Flashing firmware
# -------------------------------------------------------------
def run_cmd(cmd: str, cwd: str | None = None, timeout: int = CMD_TIMEOUT) -> subprocess.CompletedProcess:
popen_kwargs = {
'cwd': cwd,
'shell': True,
'stdout': subprocess.PIPE,
'stderr': subprocess.STDOUT,
'text': True,
'encoding': 'utf-8',
'errors': 'replace',
}
if os.name != 'nt':
popen_kwargs['preexec_fn'] = os.setsid
p = subprocess.Popen(cmd, **popen_kwargs)
try:
out, _ = p.communicate(timeout=timeout)
r = subprocess.CompletedProcess(args=cmd, returncode=p.returncode, stdout=out)
except subprocess.TimeoutExpired as ex:
if os.name != 'nt':
try:
os.killpg(p.pid, signal.SIGKILL)
except ProcessLookupError:
pass
else:
p.kill()
out, _ = p.communicate()
timeout_out = ex.stdout or out or b''
title = f'COMMAND TIMEOUT ({timeout}s): {cmd}'
print()
if os.getenv('CI'):
print(f"::group::{title}")
print(cmd_stdout_text(timeout_out))
print(f"::endgroup::")
else:
print(title)
print(cmd_stdout_text(timeout_out))
return subprocess.CompletedProcess(args=cmd, returncode=124, stdout=timeout_out)
if r.returncode != 0:
title = f'COMMAND FAILED: {cmd}'
print()
if os.getenv('CI'):
print(f"::group::{title}")
print(cmd_stdout_text(r.stdout))
print(f"::endgroup::")
else:
print(title)
print(cmd_stdout_text(r.stdout))
elif verbose:
print(cmd)
print(cmd_stdout_text(r.stdout))
return r
def flash_jlink(board: Board, firmware: str) -> subprocess.CompletedProcess:
flasher = board['flasher']
script = ['halt', 'r', f'loadfile {firmware}.elf', 'r', 'go', 'exit']
f_jlink = Path(f'{board["name"]}_{Path(firmware).name}.jlink')
with f_jlink.open('w') as f:
f.writelines(f'{s}\n' for s in script)
ret = run_cmd(f'JLinkExe -USB {flasher["uid"]} {flasher["args"]} -if swd -JTAGConf -1,-1 -speed auto -NoGui 1 -ExitOnError 1 -CommandFile {f_jlink}')
f_jlink.unlink(missing_ok=True)
return ret
def reset_jlink(board: Board) -> subprocess.CompletedProcess:
flasher = board['flasher']
script = ['halt', 'r', 'go', 'exit']
f_jlink = Path(f'{board["name"]}_reset.jlink')
if not f_jlink.exists():
with f_jlink.open('w') as f:
f.writelines(f'{s}\n' for s in script)
ret = run_cmd(f'JLinkExe -USB {flasher["uid"]} {flasher["args"]} -if swd -JTAGConf -1,-1 -speed auto -NoGui 1 -ExitOnError 1 -CommandFile {f_jlink}')
return ret
def flash_stlink(board, firmware):
flasher = board['flasher']
return run_cmd(f'STM32_Programmer_CLI --connect port=swd sn={flasher["uid"]} --write {firmware}.elf --go')
def reset_stlink(board):
flasher = board['flasher']
return run_cmd(f'STM32_Programmer_CLI --connect port=swd sn={flasher["uid"]} --rst --go')
def flash_stflash(board, firmware):
flasher = board['flasher']
ret = run_cmd(f'st-flash --serial {flasher["uid"]} write {firmware}.bin 0x8000000')
return ret
def reset_stflash(board):
flasher = board['flasher']
return subprocess.CompletedProcess(args=['dummy'], returncode=0)
def flash_openocd(board, firmware):
flasher = board['flasher']
ret = run_cmd(f'openocd -c "tcl_port disabled" -c "gdb_port disabled" -c "adapter serial {flasher["uid"]}" '
f'{flasher["args"]} -c "init; halt; program {firmware}.elf verify; reset; exit"')
return ret
def reset_openocd(board):
flasher = board['flasher']
ret = run_cmd(f'openocd -c "tcl_port disabled" -c "gdb_port disabled" -c "adapter serial {flasher["uid"]}" '
f'{flasher["args"]} -c "init; reset run; exit"')
return ret
def flash_openocd_wch(board, firmware):
flasher = board['flasher']
ret = run_cmd(f'openocd -c "tcl_port disabled" -c "gdb_port disabled" -c "telnet_port disabled" '
f'-c "adapter serial {flasher["uid"]}" {flasher.get("args", "")} -c "program {firmware}.elf reset exit"')
return ret
def reset_openocd_wch(board):
flasher = board['flasher']
ret = run_cmd(f'openocd -c "tcl_port disabled" -c "gdb_port disabled" -c "telnet_port disabled" '
f'-c "adapter serial {flasher["uid"]}" {flasher.get("args", "")} -c "init; reset run; exit"')
return ret
def flash_openocd_adi(board: Board, firmware: str) -> subprocess.CompletedProcess:
flasher = board['flasher']
openocd = OPENCOD_ADI_PATH / 'src' / 'openocd'
tcl_dir = OPENCOD_ADI_PATH / 'tcl'
ret = run_cmd(f'{openocd} -c "adapter serial {flasher["uid"]}" -s {tcl_dir} '
f'{flasher["args"]} -c "program {firmware}.elf reset exit"')
return ret
def reset_openocd_adi(board: Board) -> subprocess.CompletedProcess:
flasher = board['flasher']
openocd = OPENCOD_ADI_PATH / 'src' / 'openocd'
tcl_dir = OPENCOD_ADI_PATH / 'tcl'
ret = run_cmd(f'{openocd} -c "adapter serial {flasher["uid"]}" -s {tcl_dir} '
f'{flasher["args"]} -c "program reset exit"')
return ret
def flash_wlink_rs(board, firmware):
flasher = board['flasher']
# wlink use index for probe selection and lacking usb serial support
ret = run_cmd(f'wlink flash {firmware}.elf')
return ret
def reset_wlink_rs(board):
flasher = board['flasher']
# wlink use index for probe selection and lacking usb serial support
ret = run_cmd(f'wlink reset')
return ret
def flash_esptool(board: Board, firmware: str) -> subprocess.CompletedProcess:
flasher = board['flasher']
port = get_serial_dev(flasher["uid"], None, None, 0)
fw_dir = Path(f'{firmware}.bin').parent
with (fw_dir / 'config.env').open() as f:
idf_target = json.load(f)['IDF_TARGET']
with (fw_dir / 'flash_args').open() as f:
flash_args = f.read().strip().replace('\n', ' ')
command = (f'esptool --chip {idf_target} -p {port} {flasher["args"]} '
f'--before=default_reset --after=hard_reset write_flash {flash_args}')
ret = run_cmd(command, cwd=str(fw_dir))
return ret
def reset_esptool(board):
flasher = board['flasher']
return subprocess.CompletedProcess(args=['dummy'], returncode=0)
def flash_uniflash(board, firmware):
flasher = board['flasher']
ret = run_cmd(f'dslite.sh {flasher["args"]} -f {firmware}.hex')
return ret
def reset_uniflash(board):
flasher = board['flasher']
return subprocess.CompletedProcess(args=['dummy'], returncode=0)
def flash_lm4flash(board, firmware):
# TI Tiva-C / Stellaris ICDI: lightweight lm4flash, resets and runs after write
flasher = board['flasher']
ret = run_cmd(f'lm4flash -s {flasher["uid"]} {flasher["args"]} {firmware}.bin')
return ret
def reset_lm4flash(board):
# lm4flash has no reset-only mode; it resets+runs on flash, so reset is a no-op
flasher = board['flasher']
return subprocess.CompletedProcess(args=['dummy'], returncode=0)
# -------------------------------------------------------------
# Tests: dual
# -------------------------------------------------------------
def test_dual_host_info_to_device_cdc(board):
uid = board['uid']
declared_devs = [f'{d["vid_pid"]}_{d["serial"]}' for d in board['tests']['dev_attached']]
port = get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 0)
port = hil_flash.get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 0)
ser = open_serial_dev(port)
ser.timeout = 0.1
@ -785,12 +382,12 @@ def test_host_device_info(board):
flasher = board['flasher']
declared_devs = [f'{d["vid_pid"]}_{d["serial"]}' for d in board['tests']['dev_attached']]
port = get_serial_dev(flasher["uid"], None, None, 0)
port = hil_flash.get_serial_dev(flasher["uid"], None, None, 0)
ser = open_serial_dev(port)
ser.timeout = 0.1
# reset device since we can miss the first line
ret = globals()[f'reset_{flasher["name"].lower()}'](board)
ret = getattr(hil_flash, f'reset_{flasher["name"].lower()}')(board)
assert ret.returncode == 0, 'Failed to reset device'
# read until all expected devices are enumerated
@ -864,12 +461,12 @@ def test_host_cdc_msc_hid(board):
if not cdc_devs and not msc_devs:
return 'skipped'
port = get_serial_dev(flasher["uid"], None, None, 0)
port = hil_flash.get_serial_dev(flasher["uid"], None, None, 0)
ser = open_serial_dev(port)
ser.timeout = 0.1
# reset device to catch mount messages
ret = globals()[f'reset_{flasher["name"].lower()}'](board)
ret = getattr(hil_flash, f'reset_{flasher["name"].lower()}')(board)
assert ret.returncode == 0, 'Failed to reset device'
# Wait for all expected mount messages
@ -957,12 +554,12 @@ def test_host_msc_file_explorer(board):
if not msc_devs:
return 'skipped'
port = get_serial_dev(flasher["uid"], None, None, 0)
port = hil_flash.get_serial_dev(flasher["uid"], None, None, 0)
ser = open_serial_dev(port)
ser.timeout = 0.1
# reset device to catch mount messages
ret = globals()[f'reset_{flasher["name"].lower()}'](board)
ret = getattr(hil_flash, f'reset_{flasher["name"].lower()}')(board)
assert ret.returncode == 0, 'Failed to reset device'
# Wait for MSC mount (Disk Size message)
@ -1051,8 +648,8 @@ def test_device_board_test(board):
def test_device_cdc_dual_ports(board):
uid = board['uid']
port = [
get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 0),
get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 2)
hil_flash.get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 0),
hil_flash.get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 2)
]
ser = [open_serial_dev(p) for p in port]
@ -1091,7 +688,7 @@ def test_device_cdc_dual_ports(board):
def test_device_cdc_msc(board):
uid = board['uid']
# CDC Echo test
port = get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 0)
port = hil_flash.get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 0)
ser = open_serial_dev(port)
def rand_ascii(length):
@ -1140,7 +737,7 @@ def test_device_cdc_msc_throughput(board):
assert timeout > 0, f'Disk {dev} not found'
# Wait for CDC tty enumeration
tty = get_serial_dev(uid, 'TinyUSB', 'Throughput', 0)
tty = hil_flash.get_serial_dev(uid, 'TinyUSB', 'Throughput', 0)
timeout = enum_timeout()
while timeout > 0:
if os.path.exists(tty):
@ -1159,8 +756,8 @@ def test_device_cdc_msc_throughput(board):
pass
# Put tty in raw mode so dd sees pure binary throughput.
rs = run_cmd(f'timeout 30 stty -F {tty} raw -echo')
assert rs.returncode == 0, f'stty failed: {cmd_stdout_text(rs.stdout)}'
rs = hil_flash.run_cmd(f'timeout 30 stty -F {tty} raw -echo')
assert rs.returncode == 0, f'stty failed: {hil_flash.cmd_stdout_text(rs.stdout)}'
# Payload aim: ~5 s per direction at FS (~830 kB/s), much less at HS.
msc_count = 2 if is_fs else 16 # bs=1M
@ -1168,21 +765,21 @@ def test_device_cdc_msc_throughput(board):
tmp_file = f'/tmp/cdc_msc_tp_{uid}.bin'
rw = run_cmd(f'timeout 30 dd if=/dev/zero of={tty} bs=64K count={cdc_count} 2>&1')
assert rw.returncode == 0, f'CDC dd write failed: {cmd_stdout_text(rw.stdout)}'
cdc_w = parse_speed(cmd_stdout_text(rw.stdout))
rw = hil_flash.run_cmd(f'timeout 30 dd if=/dev/zero of={tty} bs=64K count={cdc_count} 2>&1')
assert rw.returncode == 0, f'CDC dd write failed: {hil_flash.cmd_stdout_text(rw.stdout)}'
cdc_w = parse_speed(hil_flash.cmd_stdout_text(rw.stdout))
rr = run_cmd(f'timeout 30 dd if={tty} of=/dev/null bs=64K count={cdc_count} iflag=fullblock 2>&1')
assert rr.returncode == 0, f'CDC dd read failed: {cmd_stdout_text(rr.stdout)}'
cdc_r = parse_speed(cmd_stdout_text(rr.stdout))
rr = hil_flash.run_cmd(f'timeout 30 dd if={tty} of=/dev/null bs=64K count={cdc_count} iflag=fullblock 2>&1')
assert rr.returncode == 0, f'CDC dd read failed: {hil_flash.cmd_stdout_text(rr.stdout)}'
cdc_r = parse_speed(hil_flash.cmd_stdout_text(rr.stdout))
rmr = run_cmd(f'dd if={dev} of={tmp_file} bs=1M count={msc_count} iflag=direct 2>&1')
assert rmr.returncode == 0, f'MSC dd read failed: {cmd_stdout_text(rmr.stdout)}'
msc_r = parse_speed(cmd_stdout_text(rmr.stdout))
rmr = hil_flash.run_cmd(f'dd if={dev} of={tmp_file} bs=1M count={msc_count} iflag=direct 2>&1')
assert rmr.returncode == 0, f'MSC dd read failed: {hil_flash.cmd_stdout_text(rmr.stdout)}'
msc_r = parse_speed(hil_flash.cmd_stdout_text(rmr.stdout))
rmw = run_cmd(f'dd if={tmp_file} of={dev} bs=1M count={msc_count} oflag=direct 2>&1')
assert rmw.returncode == 0, f'MSC dd write failed: {cmd_stdout_text(rmw.stdout)}'
msc_w = parse_speed(cmd_stdout_text(rmw.stdout))
rmw = hil_flash.run_cmd(f'dd if={tmp_file} of={dev} bs=1M count={msc_count} oflag=direct 2>&1')
assert rmw.returncode == 0, f'MSC dd write failed: {hil_flash.cmd_stdout_text(rmw.stdout)}'
msc_w = parse_speed(hil_flash.cmd_stdout_text(rmw.stdout))
try:
os.remove(tmp_file)
@ -1213,8 +810,8 @@ def test_device_dfu(board):
deadline = time.monotonic() + enum_timeout()
found = False
while time.monotonic() < deadline:
ret = run_cmd(f'dfu-util -l')
stdout = cmd_stdout_text(ret.stdout)
ret = hil_flash.run_cmd(f'dfu-util -l')
stdout = hil_flash.cmd_stdout_text(ret.stdout)
if f'serial="{uid}"' in stdout and 'Found DFU: [cafe:400b]' in stdout:
found = True
break
@ -1232,10 +829,10 @@ def test_device_dfu(board):
except OSError:
pass
ret = run_cmd(f'dfu-util -S {uid} -a 0 -U {f_dfu0}')
ret = hil_flash.run_cmd(f'dfu-util -S {uid} -a 0 -U {f_dfu0}')
assert ret.returncode == 0, 'Upload failed'
ret = run_cmd(f'dfu-util -S {uid} -a 1 -U {f_dfu1}')
ret = hil_flash.run_cmd(f'dfu-util -S {uid} -a 1 -U {f_dfu1}')
assert ret.returncode == 0, 'Upload failed'
with open(f_dfu0) as f:
@ -1254,8 +851,8 @@ def test_device_dfu_runtime(board):
deadline = time.monotonic() + enum_timeout()
found = False
while time.monotonic() < deadline:
ret = run_cmd(f'dfu-util -l')
stdout = cmd_stdout_text(ret.stdout)
ret = hil_flash.run_cmd(f'dfu-util -l')
stdout = hil_flash.cmd_stdout_text(ret.stdout)
if f'serial="{uid}"' in stdout and 'Found Runtime: [cafe:400c]' in stdout:
found = True
break
@ -1291,7 +888,7 @@ def test_device_printer_to_cdc(board):
uid = board['uid']
# Wait for CDC port and printer device
cdc_port = get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 0)
cdc_port = hil_flash.get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 0)
ser = open_serial_dev(cdc_port)
lp_dev = open_printer_dev(uid, 'TinyUSB', 'TinyUSB_Device', 2)
@ -1731,15 +1328,15 @@ def test_device_usbtest(board):
# its normal driver. usbtest_permit budgets USBTEST_PARALLEL batteries per controller.
script = Path(__file__).resolve().parent / 'usbtest.py'
cmd = f'python3 "{script}" --serial "{uid}" --json --keep-binding --timeout 60'
with usbtest_permit(uid):
r = run_cmd(cmd, timeout=200)
out = cmd_stdout_text(r.stdout)
with hil_lock.usbtest_permit(uid):
r = hil_flash.run_cmd(cmd, timeout=200)
out = hil_flash.cmd_stdout_text(r.stdout)
brace = out.find('{')
try:
data = json.loads(out[brace:])
passed, failed = int(data['passed']), int(data['failed'])
except (ValueError, KeyError, json.JSONDecodeError):
raise TestFail(f'usbtest did not run: {compact_output(out) or cmd_stdout_text(r.stderr)}',
raise TestFail(f'usbtest did not run: {compact_output(out) or hil_flash.cmd_stdout_text(r.stderr)}',
metric=f'{REPORT_CELL["fail"]} 0/30')
total = passed + failed
@ -1788,21 +1385,6 @@ host_test = [
]
def find_firmware(variant: str, example: str):
"""Locate a built example's firmware base path (no extension) under
cmake-build-<variant>/<example>/. Accepts the single-config layout (firmware
directly in the example dir) or Ninja Multi-Config (a per-config subdir like
RelWithDebInfo/). Returns the base Path, or None if not built."""
fw_dir = TINYUSB_ROOT / build_dir / f'cmake-build-{variant}' / example
base = Path(example).name
if fw_dir.is_dir():
for cand in [fw_dir / base, fw_dir / 'RelWithDebInfo' / base,
*(p.with_suffix('') for p in sorted(fw_dir.glob(f'*/{base}.elf')))]:
if cand.with_suffix('.elf').exists() or cand.with_suffix('.bin').exists():
return cand
return None
def test_example(board: Board, variant: str, example: str) -> tuple[int, str, str | None]:
"""
Test example firmware
@ -1820,7 +1402,7 @@ def test_example(board: Board, variant: str, example: str) -> tuple[int, str, st
test_name = f'{variant:40} {example:30} ...'
fw_name = find_firmware(variant, example)
fw_name = hil_flash.find_firmware(variant, example)
if fw_name is None:
log_line(f'{test_name} Skip (no binary)')
return 0, 'skip', None
@ -1840,9 +1422,9 @@ def test_example(board: Board, variant: str, example: str) -> tuple[int, str, st
attempt_out = io.StringIO()
with redirect_stdout(attempt_out):
if not skip_flash:
with flash_permit(board['uid']):
with hil_lock.flash_permit(board['uid']):
t_flash = time.monotonic()
ret = globals()[f'flash_{board["flasher"]["name"].lower()}'](board, str(fw_name))
ret = getattr(hil_flash, f'flash_{board["flasher"]["name"].lower()}')(board, str(fw_name))
if PROFILE:
log_line(f'[prof] {variant} {example} flash attempt {i + 1}: '
f'{time.monotonic() - t_flash:.1f}s rc={ret.returncode}')
@ -1917,7 +1499,7 @@ def build_board(board: Board) -> tuple[str, int]:
failed = 0
for v in variants:
cmd = [sys.executable, str(TINYUSB_ROOT / 'tools' / 'build.py'), '-b', name]
cmd = [sys.executable, str(hil_flash.TINYUSB_ROOT / 'tools' / 'build.py'), '-b', name]
for d in extra_defs:
cmd += ['-D', d]
if v['name'] != name:
@ -1929,7 +1511,7 @@ def build_board(board: Board) -> tuple[str, int]:
if verbose:
cmd.append('-v')
print(f' + {" ".join(cmd)}')
r = subprocess.run(cmd, cwd=TINYUSB_ROOT)
r = subprocess.run(cmd, cwd=hil_flash.TINYUSB_ROOT)
if r.returncode != 0:
failed += 1
return name, failed
@ -1940,7 +1522,7 @@ def test_board(board: Board) -> tuple[str, int, list[str], list, float]:
flasher = board['flasher']
try:
_lock_fh = acquire_board_lock(name)
_lock_fh = hil_lock.acquire_board_lock(name)
except RuntimeError as e:
log_line(f'{name:25} {STATUS_FAILED}: {e}')
# visible report row so the ❌ matches the exit code; failed-tests stays
@ -2033,7 +1615,7 @@ def test_board(board: Board) -> tuple[str, int, list[str], list, float]:
try:
# clear our pid record before dropping the flock: this worker
# process lives on (pool reuse), so a stale record would make
# board_lock.py's pid-liveness checks report a freed board as
# hil_lock.py's pid-liveness checks report a freed board as
# still locked for the rest of the run
_lock_fh.truncate(0)
except OSError:
@ -2174,7 +1756,6 @@ def main() -> None:
global verbose
global test_only
global board_test
global build_dir
global max_retry
global skip_flash
@ -2204,13 +1785,14 @@ def main() -> None:
config_file = Path(args.config_file)
boards = args.board
verbose = args.verbose
hil_flash.verbose = args.verbose
test_only = args.test_only
for entry in args.board_test:
bname, _, tnames = entry.partition(':')
if not bname or not tnames:
parser.error(f'invalid --board-test value: {entry!r} (expected BOARD:test1,test2)')
board_test[bname] = [t for t in tnames.split(',') if t]
build_dir = args.build_dir
hil_flash.build_dir = args.build_dir
max_retry = args.retry
skip_flash = args.skip_flash
@ -2234,8 +1816,8 @@ def main() -> None:
build_err = 0
if args.build:
if build_dir != 'cmake-build':
print(f'warning: --build writes into cmake-build/, but -B is {build_dir!r}; '
if hil_flash.build_dir != 'cmake-build':
print(f'warning: --build writes into cmake-build/, but -B is {hil_flash.build_dir!r}; '
f'tests will not find the freshly built firmware')
print('-' * 30)
print(f'Build phase: {len(config_boards)} board(s)')
@ -2264,7 +1846,7 @@ def main() -> None:
seed = os.getenv('HIL_SHUFFLE_SEED') or str(int(time.time()))
log_line(f'test-order shuffle seed: {seed} (HIL_SHUFFLE_SEED={seed} to replay); '
f'flash/usbtest parallel per controller: {FLASH_PARALLEL}/{USBTEST_PARALLEL}; '
f'flash/usbtest parallel per controller: {hil_lock.FLASH_PARALLEL}/{hil_lock.USBTEST_PARALLEL}; '
f'enum timeout first/retry: {ENUM_TIMEOUT}/{ENUM_TIMEOUT_RETRY}s')
hints = {}
@ -2283,8 +1865,8 @@ def main() -> None:
mgr = Manager()
cmap = mgr.dict()
initargs = (Lock(), seed,
[Semaphore(USBTEST_PARALLEL) for _ in range(CONTROLLER_SLOTS)],
[Semaphore(FLASH_PARALLEL) for _ in range(CONTROLLER_SLOTS)],
[Semaphore(hil_lock.USBTEST_PARALLEL) for _ in range(hil_lock.CONTROLLER_SLOTS)],
[Semaphore(hil_lock.FLASH_PARALLEL) for _ in range(hil_lock.CONTROLLER_SLOTS)],
cmap, Lock(), hints_by_uid)
with Pool(processes=os.cpu_count() or 1, initializer=init_worker, initargs=initargs) as pool:
async_ret = pool.map_async(test_board, config_boards)

28
test/hil/tinyusb.json
View File

@ -464,7 +464,7 @@
},
"flasher": {
"name": "openocd_wch",
"uid": "EBCA8F0670AF",
"uid": "A76D8F062C2A",
"args": "-f target/wch-riscv.cfg"
}
},
@ -513,32 +513,13 @@
},
"flasher": {
"name": "openocd_wch",
"uid": "7FD88F0604B5",
"uid": "57468F06DC03",
"args": "-f target/wch-riscv.cfg"
}
},
{
"name": "nrf5340dk",
"uid": "78E60E166B5F88BE",
"tests": {
"device": true,
"host": false,
"dual": false,
"skip": ["device/cdc_msc_freertos", "device/audio_test_freertos"],
"comment": "board new to HIL: FreeRTOS examples hardfault (UFSR=INVPC) at first task launch on the CM33_NTZ port - pre-existing upstream issue, non-FreeRTOS examples and usbtest pass; fix separately"
},
"flasher": {
"name": "jlink",
"uid": "001050076405",
"args": "-device NRF5340_XXAA_APP"
}
}
],
"boards-skip": [
{
"name": "mimxrt1064_evk",
"uid": "BAE96FB95AFA6DBB8F00005002001200",
"comment-skip": "device-port cable degraded from enum drops to killing the uPD720201 mid-battery (2026-07-17); replace the cable, verify enum, then move back",
"tests": {
"device": true,
"host": true,
@ -568,7 +549,6 @@
{
"name": "nrf54lm20dk",
"uid": "899C3DE5B0F4D5CA",
"comment-skip": "J-Link probe fails most flashes (2026-07-16); replug/repair the probe, then move back",
"tests": {
"device": true,
"host": false,
@ -597,7 +577,9 @@
"uid": "000831915224",
"args": "-device R7FA6M5BH"
}
},
}
],
"boards-skip": [
{
"name": "ra8m1_ek",
"uid": "797D142D36345030364E1737922E4B4E",