usbtest: drill a board whose recovery flasher is not openocd (#4160)
STM32_Programmer_CLI -halt cannot stand in: on stm32g0b1nucleo DHCSR read the core running again after the CLI exited.
diff --git a/.claude/skills/usbtest/SKILL.md b/.claude/skills/usbtest/SKILL.md
index 851e068..9a92fd2 100644
--- a/.claude/skills/usbtest/SKILL.md
+++ b/.claude/skills/usbtest/SKILL.md
@@ -59,8 +59,14 @@
core, so the drill SIGKILLs a halted openocd instead. On nanoch32v203-fsdev the WCH-LinkE attach
(openocd 0ce743125) itself rewrote RCC, taking the USB clock off 48 MHz, so case 27 failed -EPIPE
instead of hanging and the drill stayed inconclusive (2026-10-06).
+The halt runs through openocd. A board whose recovery flasher is not openocd (the hfp stlink
+boards) names the openocd halt with `--halt-openocd`, whose interface and target cfgs must match
+its probe and MCU, e.g. `--halt-openocd "-f interface/stlink.cfg -f target/stm32l4x.cfg"` for
+stm32l412nucleo; it injects the wedge only, and the recovery under test stays the board's own.
+`STM32_Programmer_CLI -halt` cannot stand in: the core ran again after the CLI exited (DHCSR,
+stm32g0b1nucleo, CubeProgrammer v2.23.0).
Its `cleanup` is the drill's own reset and never stands in for the harness's recovery. Boards
-whose recovery flasher is not openocd are refused.
+whose recovery flasher is not convoy-safe or has no probe reset are refused.
## Rig hazards
diff --git a/.claude/skills/usbtest/scripts/wedge_drill.py b/.claude/skills/usbtest/scripts/wedge_drill.py
index f01d38a..1aeef5b 100755
--- a/.claude/skills/usbtest/scripts/wedge_drill.py
+++ b/.claude/skills/usbtest/scripts/wedge_drill.py
@@ -3,20 +3,24 @@
"""Try to wedge one roster board on purpose and check that usbtest.py's in-run recovery clears it.
wedge_drill.py --config CONFIG --board BOARD [--variant NAME] [--delay S] [--timeout S]
- [--allow-concurrent]
+ [--halt-openocd ARGS] [--allow-concurrent]
Lock, flash device/usbtest, start case 27 (bulk write perf, whose kernel wait in test_queue is
unbounded) with post-hang recovery, and --delay seconds after its testusb process appears halt
-the DUT core through the board's recovery flasher. The intent: a halted core arms no new buffers,
-the transfer in flight never completes, testusb sits in D state on the device lock, and usbtest.py
-must call it HUNG, reset the DUT through the probe and reap the original testusb. Whether a halt
-does that is variant-dependent (USB engine, DMA, debug freeze): read the variant's reference
-manual (read-doc) before drilling it. With no flash between, case 1 must then pass on the same
-serial. The board is parked on device/board_test only when it is known to be usable.
+the DUT core through the board's openocd recovery flasher, or through --halt-openocd's openocd
+args on the board's probe when its recovery flasher is not openocd. The intent: a halted core
+arms no new buffers, the transfer in flight never completes, testusb sits in D state on the
+device lock, and usbtest.py must call it HUNG, reset the DUT through the probe and reap the
+original testusb. Whether a halt does that is variant-dependent (USB engine, DMA, debug freeze):
+read the variant's reference manual (read-doc) before drilling it. With no flash between, case 1
+must then pass on the same serial. The board is parked on device/board_test only when it is
+known to be usable.
-Refuses before touching hardware (exit 2) as run_case.py does, and when the board's recovery
-flasher is not openocd (the halt runs through it) or is not convoy-safe, or --delay plus the halt
-bound leaves the halt no room to finish before the case times out.
+Refuses before touching hardware (exit 2) as run_case.py does; when the board's recovery flasher
+is not convoy-safe or has no probe reset; when there is no openocd halt (no openocd recovery
+flasher and no --halt-openocd), or --halt-openocd is not convoy-safe, lacks a probe uid or
+overrides an openocd recovery flasher; or when --delay plus the halt bound leaves the halt no
+room to finish before the case times out.
stdout ends with one JSON line: {"drill", "board", "variant", "reason", "halt", "cases", "wedged",
"recovery", "device", "smoke", "cleanup", "boardState", "error"}. drill is "pass" (HUNG, reset
@@ -135,15 +139,36 @@
def recovery_board(board):
"""The board with its recovery flasher as `flasher`, or Refused when the drill cannot use it."""
rec = hil_flash.recover_flasher(board)
- if rec.get('name') != 'openocd':
- raise Refused(f'{board["name"]}: the drill halts through an openocd recovery flasher; '
- f'this board recovers through {rec.get("name")}')
if not hil_flash.convoy_safe(rec):
raise Refused(f'{board["name"]}: its recovery flasher is not convoy-safe, so usbtest.py '
f'would not attempt a recovery')
+ if hil_flash.reset_primitive(rec['name']) is None:
+ raise Refused(f'{board["name"]}: {rec["name"]} has no probe reset, the recovery the drill '
+ f'checks for')
return {'name': board['name'], 'flasher': rec}
+def halt_board(board, rec_board, halt_args):
+ """The board with the openocd entry that halts it as `flasher`: its openocd recovery flasher,
+ else `halt_args` on its probe."""
+ name = board['name']
+ if rec_board['flasher']['name'] == 'openocd':
+ if halt_args is not None:
+ raise Refused(f'{name}: --halt-openocd is for a recovery flasher that is not openocd; '
+ f'this board halts through its own')
+ return rec_board
+ if halt_args is None:
+ raise Refused(f'{name}: the drill halts through openocd and this board recovers through '
+ f'{rec_board["flasher"]["name"]}; name the openocd halt with --halt-openocd')
+ uid = board['flasher'].get('uid')
+ if not uid:
+ raise Refused(f'{name}: no probe uid for openocd\'s adapter serial')
+ halt = {'name': 'openocd', 'uid': uid, 'args': halt_args}
+ if not hil_flash.convoy_safe(halt):
+ raise Refused(f'{name}: --halt-openocd {halt_args!r} is not convoy-safe')
+ return {'name': name, 'flasher': halt}
+
+
def identity(uid):
"""(node, devnum) of the one enumerated device with this serial, else None."""
found = hil_util.usb_scan(vid_pid=(usbtest.VID, usbtest.PID), serial=uid)
@@ -197,7 +222,7 @@
return rec
-def inject(rec_board, node, delay, timeout, battery_done, report):
+def inject(halt_with, node, delay, timeout, battery_done, report):
"""Halt the core while case 27's testusb is in flight, finishing before the case can time
out. True only when the halt succeeded and the case was still running after it."""
deadline = time.monotonic() + APPEAR_TIMEOUT
@@ -219,8 +244,8 @@
return False
report['halt']['issued'] = True # before the call: a halt that times out may still land
t = time.monotonic()
- halt = halt_held if wch(rec_board) else halt_openocd
- ret = halt(rec_board, timeout=HALT_TIMEOUT)
+ halt = halt_held if wch(halt_with) else halt_openocd
+ ret = halt(halt_with, timeout=HALT_TIMEOUT)
report['halt'].update(rc=ret.returncode, took=round(time.monotonic() - t, 2),
caseRunningAfter=alive(pid))
if ret.returncode != 0:
@@ -266,7 +291,7 @@
return now is not None and testusb_pid(now[0]) is None
-def drill(board, rec_board, fw, delay, timeout, report):
+def drill(board, halt_with, fw, delay, timeout, report):
err = run_case.flash(board, fw)
report['boardState'] = 'flash failed' if err else 'usbtest firmware'
if err:
@@ -295,7 +320,7 @@
worker = threading.Thread(target=run_battery)
worker.start()
try:
- halted = inject(rec_board, before[0], delay, timeout, done, report)
+ halted = inject(halt_with, before[0], delay, timeout, done, report)
except ProbeHeld as e:
report['halt']['probeHeld'] = e.pid
run_case.kill_children(INHERITED) # before the battery's own recovery opens the probe
@@ -367,7 +392,7 @@
if stuck:
report['halt']['probeHeld'] = stuck[0]
if report['halt'].get('probeHeld'):
- report['boardState'] = (f'unknown, not parked: openocd {report["halt"]["probeHeld"]} may '
+ report['boardState'] = (f'unknown, not parked: probe process {report["halt"]["probeHeld"]} may '
f'still hold the probe (usb-kernel-recover)')
return
if report['drill'] == 'pass':
@@ -400,6 +425,9 @@
p.add_argument('--variant', help='roster variant, required when the board has several')
p.add_argument('--delay', type=float, default=0.3, help='seconds into case 27 before the halt (default 0.3)')
p.add_argument('--timeout', type=int, default=60, help='per-case timeout in seconds (default 60, as HIL)')
+ p.add_argument('--halt-openocd', metavar='ARGS',
+ help='openocd args that halt the core on the board\'s probe, required when its recovery '
+ 'flasher is not openocd, e.g. "-f interface/stlink.cfg -f target/stm32l4x.cfg"')
p.add_argument('--allow-concurrent', action='store_true',
help="skip the live-battery check; see run_case.py's description first")
args = p.parse_args()
@@ -429,6 +457,7 @@
board, variant = run_case.resolve(args.config, args.board, args.variant)
report['variant'] = variant
rec_board = recovery_board(board)
+ halt_with = halt_board(board, rec_board, args.halt_openocd)
flasher = board['flasher']['name']
fw = run_case.firmware(variant, 'device/usbtest', flasher)
park_fw = run_case.firmware(variant, 'device/board_test', flasher)
@@ -452,7 +481,7 @@
try:
if not args.allow_concurrent:
run_case.check_peers() # again: a peer may have started before the lock
- report['drill'] = drill(board, rec_board, fw, args.delay, args.timeout, report)
+ report['drill'] = drill(board, halt_with, fw, args.delay, args.timeout, report)
except Refused as e:
report['error'] = str(e)
return finish(2)
diff --git a/.claude/test/test_usbtest_wedge_drill.py b/.claude/test/test_usbtest_wedge_drill.py
index 997ddd0..eaeba9a 100644
--- a/.claude/test/test_usbtest_wedge_drill.py
+++ b/.claude/test/test_usbtest_wedge_drill.py
@@ -23,7 +23,9 @@
wedge_drill = importlib.util.module_from_spec(spec)
spec.loader.exec_module(wedge_drill)
run_case = wedge_drill.run_case
+REAL_RESET_PRIMITIVE = run_case.hil_flash.reset_primitive
+ST_HALT = '-f interface/stlink.cfg -f target/stm32l4x.cfg'
JLINK_REC = {'name': 'openocd', 'args': '-f interface/jlink.cfg -f target/stm32f4x.cfg'}
ROSTER = {'boards': [
{'name': 'ok', 'uid': 'UID1', 'flasher': {'name': 'jlink', 'uid': 'P1', 'args': '-device X'},
@@ -33,6 +35,8 @@
{'name': 'esp', 'uid': 'UID3', 'flasher': {'name': 'esptool', 'uid': 'P3'}, 'tests': {'device': True}},
{'name': 'wch', 'uid': 'UID4', 'flasher': {'name': 'openocd', 'uid': 'P4', 'args': '-f target/wch-riscv.cfg',
'vid_pid': '0x1a86 0x8010'}, 'tests': {'device': True}},
+ {'name': 'st', 'uid': 'UID5', 'flasher': {'name': 'stlink', 'uid': 'P5'}, 'tests': {'device': True}},
+ {'name': 'st-nouid', 'uid': 'UID7', 'flasher': {'name': 'stlink', 'uid': ''}, 'tests': {'device': True}},
]}
RECOVERED = ('auto-recovering: resetting ok via openocd probe\nreset rc 0\n'
'recovery freed the device: testusb reaped\n')
@@ -71,6 +75,8 @@
self.halted = threading.Event()
self.injected = threading.Event() # inject returned, halted or not
self.live_children = []
+ self.battery_boards = []
+ self.halt_boards = []
tmp = tempfile.TemporaryDirectory()
test.addCleanup(tmp.cleanup)
self.config = Path(tmp.name) / 'rig.json'
@@ -78,6 +84,7 @@
def battery(board, fw, tests, timeout):
self.calls.append(('battery', tuple(tests)))
+ self.battery_boards.append(board)
if tests == [SMOKE_CASE] and self.smoke_raises:
raise self.smoke_raises
if tests == [27]:
@@ -88,6 +95,7 @@
def halt(board, timeout, kind='halt'):
self.calls.append((kind, board['flasher']['name'], timeout))
+ self.halt_boards.append(board)
self.halted.set()
if self.halt_raises:
raise self.halt_raises
@@ -394,9 +402,25 @@
self.assertEqual(r['boardState'], 'untouched')
self.assertEqual(rig.calls, [])
- def test_no_openocd_recovery_flasher(self):
- self.refused('--board', 'norec', says='halts through an openocd recovery flasher')
- self.refused('--board', 'esp', says='halts through an openocd recovery flasher')
+ def test_a_recovery_flasher_the_drill_cannot_check(self):
+ self.refused('--board', 'norec', says='not convoy-safe')
+ rig = Rig(self) # its reset stub would hide esptool's missing reset
+ with mock.patch.object(run_case.hil_flash, 'reset_primitive', REAL_RESET_PRIMITIVE):
+ rc, r = rig.run('--board', 'esp')
+ self.assertEqual(rc, 2)
+ self.assertIn('has no probe reset', r['error'])
+ self.assertEqual(rig.calls, [])
+
+ def test_a_non_openocd_recovery_needs_a_halt_override(self):
+ self.refused('--board', 'st', says='--halt-openocd')
+
+ def test_a_bad_halt_override(self):
+ self.refused('--board', 'st', '--halt-openocd', ' ', says='not convoy-safe')
+ self.refused('--board', 'st', '--halt-openocd', '-f interface/cmsis-dap.cfg', says='not convoy-safe')
+ self.refused('--board', 'st-nouid', '--halt-openocd', ST_HALT, says='no probe uid')
+
+ def test_an_override_on_an_openocd_recovery_board(self):
+ self.refused('--board', 'ok', '--halt-openocd', ST_HALT, says='halts through its own')
def test_no_room_for_the_halt(self):
self.refused('--board', 'ok', '--delay', '50', '--timeout', '60', says='must end before')
@@ -406,6 +430,27 @@
HALTS = f'print({wedge_drill.HALTED_MARK!r}, flush=True)\ntime.sleep(30)\n' # a fake openocd that halts
+class HaltOverride(unittest.TestCase):
+ """--halt-openocd injects the wedge on the board's probe; the battery and the cleanup keep
+ the board's own recovery flasher."""
+
+ def test_the_override_halts_and_stlink_recovers(self):
+ rig = Rig(self)
+ rc, r = rig.run('--board', 'st', '--halt-openocd', ST_HALT, '--delay', '0')
+ self.assertEqual((rc, r['drill']), (0, 'pass'), r)
+ self.assertEqual(rig.halt_boards[0]['flasher'], {'name': 'openocd', 'uid': 'P5', 'args': ST_HALT})
+ self.assertTrue(rig.battery_boards)
+ for b in rig.battery_boards: # the halt entry must not become the battery's recovery
+ self.assertEqual(run_case.hil_flash.recover_flasher(b), {'name': 'stlink', 'uid': 'P5'})
+
+ def test_a_failed_override_halt_cleans_up_through_stlink(self):
+ rig = Rig(self)
+ rig.halt_rc = 1
+ _, r = rig.run('--board', 'st', '--halt-openocd', ST_HALT, '--delay', '0')
+ self.assertEqual(r['drill'], 'inconclusive')
+ self.assertIn(('cleanup-reset', 'stlink'), rig.calls)
+
+
class WchHeldHalt(unittest.TestCase):
"""A WCH-Link's openocd shutdown resumes the core, so its halt is held by SIGKILLing a
halted openocd; the probe opens once, so that openocd must be gone before the recovery."""
@@ -529,7 +574,7 @@
self.assertEqual(rc, 1)
self.assertEqual(killed, [wedge_drill.INHERITED]) # an inherited logger is spared
self.assertEqual(r['halt']['probeHeld'], 777)
- self.assertIn('openocd 777', r['boardState'])
+ self.assertIn('probe process 777', r['boardState'])
self.assertNotIn('cleanup-reset', [c[0] for c in rig.calls])
self.assertFalse(rig.parked())
self.assertTrue(rig.lock.closed)
