Run the agent skill scripts on Windows (#4247)
diff --git a/.claude/skills/build-doc/scripts/gen_doc.py b/.claude/skills/build-doc/scripts/gen_doc.py
index 60cb7ee..75d8c94 100755
--- a/.claude/skills/build-doc/scripts/gen_doc.py
+++ b/.claude/skills/build-doc/scripts/gen_doc.py
@@ -21,7 +21,7 @@
"""key: value pairs of the /* metadata: ... */ block, {} when the file or block is missing."""
if not path.exists():
return {}
- m = re.search(r'/\*\s*metadata:(.*?)\*/', path.read_text(), re.DOTALL)
+ m = re.search(r'/\*\s*metadata:(.*?)\*/', path.read_text(encoding='utf-8'), re.DOTALL)
if not m:
return {}
return {k: v.strip() for k, v in re.findall(r'^[ \t]*(\w+):[ \t]*(.+)$', m.group(1), re.MULTILINE)}
@@ -30,7 +30,8 @@
def boards(bsp):
"""{manufacturer: {board: [name, family, url, note]}} plus the paths skipped for lack of metadata."""
vendors, skipped = {}, []
- for family_dir in sorted(p for p in bsp.iterdir() if p.is_dir()):
+ # by name: Windows paths compare case-insensitively, which would reorder boards.rst
+ for family_dir in sorted((p for p in bsp.iterdir() if p.is_dir()), key=lambda p: p.name):
family_c = family_dir / 'family.c'
if not family_c.exists():
family_c = family_dir / 'boards' / 'family.c'
@@ -39,7 +40,7 @@
skipped.append(family_dir)
continue
manufacturer = f_meta.get('manufacturer', '')
- for board_dir in sorted(p for p in (family_dir / 'boards').iterdir() if p.is_dir()):
+ for board_dir in sorted((p for p in (family_dir / 'boards').iterdir() if p.is_dir()), key=lambda p: p.name):
b_meta = metadata(board_dir / 'board.h')
if not b_meta:
skipped.append(board_dir)
@@ -107,7 +108,7 @@
def hil_md(root):
out = [HEADER, '']
for title, src in (('ci rig', 'test/hil/tinyusb.json'), ('hfp rig', 'test/hil/hfp.json')):
- entries = json.loads((root / src).read_text()).get('boards', [])
+ entries = json.loads((root / src).read_text(encoding='utf-8')).get('boards', [])
if not entries:
continue
out += [f'### {title}', '', f'{len(entries)} boards, from `{src}`.', '',
@@ -124,14 +125,14 @@
gen_deps_doc(root / 'docs/reference/dependencies.rst')
print(f'wrote {root / "docs/reference/dependencies.rst"}')
vendors, skipped = boards(root / 'hw' / 'bsp')
- (root / 'docs/reference/boards.rst').write_text(boards_rst(vendors))
+ (root / 'docs/reference/boards.rst').write_text(boards_rst(vendors), encoding='utf-8')
print(f'wrote {root / "docs/reference/boards.rst"}')
- (root / 'docs/reference/hil_boards.md').write_text(hil_md(root))
+ (root / 'docs/reference/hil_boards.md').write_text(hil_md(root), encoding='utf-8')
print(f'wrote {root / "docs/reference/hil_boards.md"}')
if skipped:
print('no metadata block, left out of boards.rst:', file=sys.stderr)
for path in skipped:
- print(f' {path.relative_to(root)}', file=sys.stderr)
+ print(f' {path.relative_to(root).as_posix()}', file=sys.stderr)
return 0
diff --git a/.claude/skills/build/scripts/check_build.py b/.claude/skills/build/scripts/check_build.py
index 3e7f6a3..2120f70 100755
--- a/.claude/skills/build/scripts/check_build.py
+++ b/.claude/skills/build/scripts/check_build.py
@@ -432,7 +432,7 @@
configuration does not satisfy is an empty translation unit, whatever built."""
argv = family_json.preprocess_argv(entry) + ['-E', entry['file']]
try:
- r = subprocess.run(argv, capture_output=True, text=True, cwd=entry['directory'])
+ r = subprocess.run(argv, capture_output=True, encoding='utf-8', errors='replace', cwd=entry['directory'])
except OSError:
return None
if r.returncode != 0:
@@ -659,7 +659,7 @@
def roster_variants(boards, config):
"""(board, build name, defines, cflags) for each variant hil_test.py runs of each board."""
try:
- roster = {b['name']: b for b in json.loads(Path(config).read_text())['boards']}
+ roster = {b['name']: b for b in json.loads(Path(config).read_text(encoding='utf-8'))['boards']}
except (OSError, ValueError, KeyError, TypeError) as e:
fail(f'could not read the HIL roster {config}: {e}')
unknown = [b for b in boards if b not in roster]
@@ -741,10 +741,11 @@
def run(cmd, verbose):
# streamed as it arrives: a variant build runs for minutes, and a silent buffer reads as a
- # stall. PYTHONUNBUFFERED, or tools/build.py block-buffers its piped stdout until it exits
+ # stall. PYTHONUNBUFFERED, or tools/build.py block-buffers its piped stdout until it exits;
+ # PYTHONIOENCODING, or on Windows it writes the pipe in the ANSI code page, not as decoded here
out = []
- with subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, cwd=ROOT,
- env={**os.environ, 'PYTHONUNBUFFERED': '1'}) as p:
+ with subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, encoding='utf-8', errors='replace', cwd=ROOT,
+ env={**os.environ, 'PYTHONUNBUFFERED': '1', 'PYTHONIOENCODING': 'utf-8'}) as p:
for line in p.stdout:
out.append(line)
if verbose:
diff --git a/.claude/skills/build/scripts/coverage_audit.py b/.claude/skills/build/scripts/coverage_audit.py
index 8aac5b4..ff6c094 100755
--- a/.claude/skills/build/scripts/coverage_audit.py
+++ b/.claude/skills/build/scripts/coverage_audit.py
@@ -27,6 +27,7 @@
historical tree, since the current graph cannot know what built the deleted file.
"""
import argparse
+import importlib.util
import json
import os
import re
@@ -37,6 +38,11 @@
from pathlib import Path
ROOT = Path(__file__).resolve().parents[4]
+# split_command, from this script's checkout under a private name: graph's `import build` must
+# still get the audited checkout's own family_json
+_spec = importlib.util.spec_from_file_location('_coverage_audit_family_json', ROOT / 'tools' / 'family_json.py')
+family_json = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(family_json)
DEPS_HEADER = re.compile(r'^(\S.*): #deps \d+')
ROLES = ('device', 'host', 'dual', 'typec')
FORMAT = 3
@@ -46,11 +52,20 @@
pass
+# Windows: the paths rel() found on another drive than the checkout, outside it as a '..' is on
+# POSIX; graph counts them per board on stderr, so a checkout split across drives shows up
+other_drive = set()
+
+
def rel(path, root, base=None):
"""Repo-relative path of `path` (relative ones resolved against `base`, default `root`),
or None for a file outside the checkout (toolchain headers) or a generated one."""
p = os.path.normpath(path if os.path.isabs(path) else os.path.join(base or root, path))
- r = os.path.relpath(p, root)
+ try:
+ r = Path(os.path.relpath(p, root)).as_posix()
+ except ValueError: # Windows: another drive (a toolchain header) is outside
+ other_drive.add(p)
+ return None
return None if r.startswith('..') or r.startswith('cmake-build') else r
@@ -112,7 +127,7 @@
and the -L dirs - plus the including script's own directory."""
seen = set() if seen is None else seen
try:
- with open(script, errors='replace') as fh:
+ with open(script, encoding='utf-8', errors='replace') as fh:
text = fh.read()
except OSError:
return seen
@@ -199,7 +214,7 @@
def run(cmd, cwd, **kw):
- return subprocess.run(cmd, cwd=cwd, capture_output=True, text=True, **kw)
+ return subprocess.run(cmd, cwd=cwd, capture_output=True, encoding='utf-8', errors='replace', **kw)
def run_ok(cmd, cwd):
@@ -219,14 +234,15 @@
if '__gnu_lto_slim' not in r.stdout:
return defines_symbols(r.stdout)
try:
- n = run([compiler + '-nm', obj_path], cwd)
+ nm = compiler[:-4] + '-nm.exe' if compiler.lower().endswith('.exe') else compiler + '-nm'
+ n = run([nm, obj_path], cwd)
except OSError:
return True
return n.returncode != 0 or nm_defines_symbols(n.stdout)
def compiler_of(entry):
- return entry['arguments'][0] if 'arguments' in entry else entry['command'].split()[0]
+ return entry['arguments'][0] if 'arguments' in entry else family_json.split_command(entry['command'])[0]
def extract(build_dir, root, example=None):
@@ -235,11 +251,11 @@
its CMake inputs are its own too. `nodeps` lists live objects ninja recorded no deps
for: only their source is known."""
src_of = {}
- with open(os.path.join(build_dir, 'compile_commands.json')) as fh:
+ with open(os.path.join(build_dir, 'compile_commands.json'), encoding='utf-8') as fh:
cc = json.load(fh)
for e in cc:
out = e['output'] if os.path.isabs(e['output']) else os.path.join(e['directory'], e['output'])
- src_of[os.path.relpath(out, build_dir)] = e
+ src_of[Path(os.path.relpath(out, build_dir)).as_posix()] = e # as ninja names it
targets = run_ok(['ninja', '-C', build_dir, '-t', 'targets', 'all'], root)
known = {line.split(':', 1)[0] for line in targets.splitlines()}
deps = parse_ninja_deps(run_ok(['ninja', '-C', build_dir, '-t', 'deps'], root))
@@ -272,7 +288,11 @@
s.add(f)
continue
a = os.path.normpath(os.path.join(build_dir, d))
- for t in (d, os.path.relpath(a, build_dir), a):
+ try:
+ r = Path(os.path.relpath(a, build_dir)).as_posix()
+ except ValueError: # Windows: on another drive than the build, so no relative ninja name
+ r = d
+ for t in (d, r, a):
if t in known:
s |= generated_inputs(t)
break
@@ -365,6 +385,7 @@
name = f'audit-{os.getpid()}-{board}'
build_dir = os.path.join(root, 'cmake-build', f'cmake-build-{name}')
shutil.rmtree(build_dir, ignore_errors=True)
+ other_drive.clear()
try:
r = run([sys.executable, 'tools/build.py', '-b', board, '--build-name', name], root)
rec = {'format': FORMAT, 'head': head, 'board': board, 'config': 'default',
@@ -378,6 +399,9 @@
finally:
shutil.rmtree(build_dir, ignore_errors=True)
rec['secs'] = round(time.monotonic() - t0, 1)
+ if other_drive:
+ print(f'{board}: {len(other_drive)} path(s) on another drive than the checkout counted as outside it, '
+ f'e.g. {min(other_drive)}', file=sys.stderr)
with open(path, 'w') as fh:
json.dump(rec, fh)
print(f'{board}: {rec["status"]} {rec["secs"]}s '
diff --git a/.claude/skills/hil-contract/SKILL.md b/.claude/skills/hil-contract/SKILL.md
index 50967b1..d9a7c61 100644
--- a/.claude/skills/hil-contract/SKILL.md
+++ b/.claude/skills/hil-contract/SKILL.md
@@ -121,6 +121,13 @@
An agent launches it with `--run-id` and waits on it as Timing says.
+It refuses, before touching the rig, when no `rsync` is on PATH. On a Windows dev PC that must be
+an MSYS2 or Cygwin `rsync.exe` in an absolute PATH dir, and it must find its own distribution's
+`ssh` first on PATH, since with Windows OpenSSH (`System32\OpenSSH`) the argument conversion and
+the pipes rsync relies on break. There it also refuses a config or firmware dir on another drive
+than the checkout, which rsync cannot name, and snapshots receipt-pinned firmware under
+`.hil-remote/`. The Windows path is not yet proved end to end on a rig.
+
One invocation per board is wrong here, not merely slow: each run `rm -rf`s `REMOTE_DIR`
and rewrites the report, so only the last board's rows survive. A second run sharing
`REMOTE_DIR` is refused, before the wipe, while the first holds `<REMOTE_DIR>.lock`. The rig
diff --git a/.claude/skills/hil-contract/scripts/hil_remote.py b/.claude/skills/hil-contract/scripts/hil_remote.py
index 989619e..325b32e 100644
--- a/.claude/skills/hil-contract/scripts/hil_remote.py
+++ b/.claude/skills/hil-contract/scripts/hil_remote.py
@@ -27,7 +27,7 @@
"""
import argparse
import contextlib
-import fcntl
+import errno
import hashlib
import importlib
import json
@@ -41,6 +41,12 @@
import time
from pathlib import Path
+try:
+ import fcntl
+except ImportError: # Windows
+ fcntl = None
+ import msvcrt
+
ROOT = Path(os.environ.get('ROOT_DIR') or Path(__file__).resolve().parents[4]).resolve()
# Everything the rig executes, by repo-relative path. test_hil_bounded's RemoteStaging walks
@@ -168,7 +174,7 @@
config_path = config_path.relative_to(ROOT)
board_args = ' '.join(f'--board {shlex.quote(b)}' for b in boards) or '--board <board>'
line = (f'python3 .claude/skills/build/scripts/check_build.py {board_args} --shared '
- f'--variants {shlex.quote(str(config_path))}')
+ f'--variants {shlex.quote(config_path.as_posix())}')
if build_dir != 'cmake-build':
line += f'\n (it writes under cmake-build/, not {build_dir}/)'
return line
@@ -208,14 +214,14 @@
found = variant_dirs(config, build_dir, name)
present = [d for d in found if d.is_dir()]
if boards and not present:
- missing.append(f' {name}: none of {", ".join(str(d.relative_to(ROOT)) for d in found)}')
+ missing.append(f' {name}: none of {", ".join(d.relative_to(ROOT).as_posix() for d in found)}')
unbuilt.append(name)
for i, d in enumerate(found):
if d not in present and boards:
# hil_test.py logs `Skip (no binary)` and exits 0 for these cells, except that a
# one-test run that flashes fails a later variant's same-PID boundary (HIL contract, Build)
boundary = '' if args.skip_flash or i == 0 else ' or fail the same-PID boundary on a one-test run'
- print(f'warning: {name}: no {d.relative_to(ROOT)} -- its cells will be skipped{boundary}',
+ print(f'warning: {name}: no {d.relative_to(ROOT).as_posix()} -- its cells will be skipped{boundary}',
file=sys.stderr)
dirs += present
if missing:
@@ -231,7 +237,7 @@
"""hil_report.json is not namespaced by rig, so a sidecar from another rig or config
would merge boards that never ran here. Require one row from this roster."""
try:
- rows = json.loads(sidecar.read_text()).get('rows') or []
+ rows = json.loads(sidecar.read_text(encoding='utf-8')).get('rows') or []
except (OSError, ValueError, AttributeError):
return False
variants_of = helper('hil_report').variants_of
@@ -271,16 +277,44 @@
f'exec env {shlex.join([*forwarded_env(), *run])}')
+def local_operand(path):
+ """A local file as rsync takes it. Windows rsync reads a drive letter as a remote host, so
+ there the file is named relative to the checkout, where rsync runs."""
+ if os.name != 'nt':
+ return str(path)
+ try:
+ return Path(os.path.relpath(path, ROOT)).as_posix()
+ except ValueError:
+ fail(f'{path} is not on the drive of the checkout {ROOT}, so rsync cannot name it')
+
+
+def rsync_program():
+ """The rsync to run, None when there is none. Windows: an rsync.exe in an absolute PATH dir,
+ which is all subprocess can start by name, and not one in the cwd, which CreateProcess
+ searches first; a relative dir would name it from the cwd, not from ROOT where it runs."""
+ if os.name != 'nt':
+ return 'rsync' if shutil.which('rsync') else None
+ for d in os.environ.get('PATH', '').split(os.pathsep):
+ if os.path.isabs(d) and os.path.isfile(os.path.join(d, 'rsync.exe')):
+ return os.path.join(d, 'rsync.exe')
+ return None
+
+
def rsync(guard, *args, optional=False):
"""rsync with a remote endpoint under `guard` = (remote_dir, token): the remote rsync
server starts only under a shared hold on the tree's lock with this run's token still
- in place, so a transfer never lands in a tree another run rebuilt after our lease died."""
+ in place, so a transfer never lands in a tree another run rebuilt after our lease died.
+ Local files are Paths; options and host:path specs are str, passed as they are."""
remote_dir, token = guard
+ args = [local_operand(a) if isinstance(a, Path) else a for a in args]
+ program = rsync_program() if os.name == 'nt' else 'rsync'
+ if program is None:
+ fail('rsync.exe is no longer on PATH')
# rsync appends its --server arguments after --rsync-path, so `sh -c '...' rsync` makes
# them "$@" of the guard shell, which execs the real rsync on them
server = f'sh -c {shlex.quote(tree_guard(remote_dir, token) + " && exec rsync \"$@\"")} rsync'
# an optional fetch (a green run writes no .failed) must not print rsync's code-23 error
- return subprocess.run(['rsync', '-e', shlex.join(['ssh', *SSH_OPTS]), f'--rsync-path={server}', *args],
+ return subprocess.run([program, '-e', shlex.join(['ssh', *SSH_OPTS]), f'--rsync-path={server}', *args],
cwd=ROOT, stderr=subprocess.DEVNULL if optional else None).returncode
@@ -293,7 +327,7 @@
md, js = ROOT / report.REPORT_MD, ROOT / report.REPORT_JSON
tmp = [p.with_name(p.name + '.tmp') for p in (md, js)]
guard = (remote_dir, token)
- ok = all(rsync(guard, '-q', f'{remote}:{remote_dir}/{p.name}', str(t), optional=True) == 0 and t.is_file()
+ ok = all(rsync(guard, '-q', f'{remote}:{remote_dir}/{p.name}', t, optional=True) == 0 and t.is_file()
for p, t in zip((md, js), tmp))
if ok:
for p, t in zip((md, js), tmp):
@@ -308,7 +342,7 @@
# A green run writes no .failed, so a stale local spec would re-flash boards that passed.
spec = ROOT / f'{config_name}.failed'
spec.unlink(missing_ok=True)
- if rsync(guard, '-q', f'{remote}:{remote_dir}/{spec.name}', str(spec), optional=True) == 0 and spec.is_file():
+ if rsync(guard, '-q', f'{remote}:{remote_dir}/{spec.name}', spec, optional=True) == 0 and spec.is_file():
copied.append(spec.name)
print(f'==> {spec.name} copied to {spec}')
return copied
@@ -319,14 +353,16 @@
"""Wipe and recreate remote_dir, yielding (its absolute path, this run's token) while this
ssh session holds its lock: another runner sharing REMOTE_DIR would otherwise rm -rf this
run's tree."""
+ if not rsync_program(): # before the wipe: every transfer of the run needs it
+ fail('rsync is not on PATH; the staging and the copy-back need it (on Windows, MSYS2\'s or Cygwin\'s rsync.exe)')
token = f'{os.getpid()}-{time.time_ns()}'
# stderr to a file, not a pipe: nothing drains a pipe while the lease is held, and a
# remote that writes more than the pipe buffer (rm -rf reporting hundreds of entries)
# would block with the lock taken
- err_file = tempfile.TemporaryFile('w+')
+ err_file = tempfile.TemporaryFile('w+', encoding='utf-8', errors='replace')
lease = subprocess.Popen(
['ssh', *SSH_OPTS, remote, shlex.join(['bash', '-c', SETUP_SCRIPT, 'hil-setup', remote_dir, build_dir, token])],
- stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=err_file, text=True)
+ stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=err_file, encoding='utf-8', errors='replace')
marked = next((ln.rstrip('\n').partition('=')[2] for ln in lease.stdout
if ln.startswith('HIL_REMOTE_DIR=')), '')
if not marked.startswith('/'):
@@ -362,7 +398,7 @@
def staged_files(dirs, base=None):
"""{path relative to base (default the checkout): sha256} of what FIRMWARE_FILTER stages from dirs."""
- return {str(f.relative_to(base or ROOT)): digest(f) for f in staged_paths(dirs)}
+ return {f.relative_to(base or ROOT).as_posix(): digest(f) for f in staged_paths(dirs)}
def snapshot(dirs, into):
@@ -382,7 +418,7 @@
def example_firmware(d, files):
"""The <role>/<name> example dirs of variant dir d holding staged <name>.<suffix> firmware,
directly or in one config subdir, where hil_flash.find_firmware() looks."""
- prefix = f'{d.relative_to(ROOT)}/'
+ prefix = f'{d.relative_to(ROOT).as_posix()}/'
return {'/'.join(parts[:2]) for parts in (p[len(prefix):].split('/') for p in files if p.startswith(prefix))
if len(parts) in (3, 4) and parts[0] in EXAMPLE_ROLES
and Path(parts[-1]).stem == parts[1] and Path(parts[-1]).suffix in STAGED_SUFFIXES}
@@ -405,7 +441,7 @@
else:
registered = tools_build.cmake_registered_targets(str(d))
if registered is None:
- fail(f'cmake cannot list the targets of {d.relative_to(ROOT)}, so the firmware its build made cannot be told '
+ fail(f'cmake cannot list the targets of {d.relative_to(ROOT).as_posix()}, so the firmware its build made cannot be told '
f'from a stale one')
built = {e for e in examples if e.split('/')[1] in registered}
return sorted(examples - built)
@@ -419,7 +455,7 @@
args = helper('hil_args').build_parser().parse_args([*argv, str(config_path)])
check_build_dir(args.build_dir)
try:
- config = json.loads(config_path.read_text())
+ config = json.loads(config_path.read_text(encoding='utf-8'))
except (OSError, ValueError) as e:
fail(f'could not read the config {config_path}: {e}')
return args, config, config_path
@@ -438,11 +474,11 @@
files = staged_files([d for _, d in dirs])
# only example firmware: CMake's compiler-ABI probes leave a .bin in a dir never built
examples = {d: example_firmware(d, files) for _, d in dirs}
- unbuilt = [str(d.relative_to(ROOT)) for _, d in dirs if not examples[d]]
+ unbuilt = [d.relative_to(ROOT).as_posix() for _, d in dirs if not examples[d]]
if unbuilt:
fail(f'not built: {", ".join(unbuilt)}; a receipt covers every variant the run selects, build with\n'
f' {build_command(config_path, boards, args.build_dir)}')
- stale = [f'{d.relative_to(ROOT)}/{e}' for b, d in dirs for e in unverified_examples(b, d, examples[d])]
+ stale = [f'{d.relative_to(ROOT).as_posix()}/{e}' for b, d in dirs for e in unverified_examples(b, d, examples[d])]
if stale:
fail(f'firmware of an example the build does not build: {", ".join(stale)}; a receipt pins only firmware '
f'the build verified and a run stages every file there, so remove those dirs and write the receipt again')
@@ -466,13 +502,13 @@
dirs are every variant dir of the run's boards, staged the checkout-relative digests of
the firmware to stage."""
try:
- receipt = json.loads(Path(path).read_text())
+ receipt = json.loads(Path(path).read_text(encoding='utf-8'))
head, pinned = receipt['head'], receipt['files']
config_digest, built = receipt['configDigest'], receipt['boards']
except (OSError, ValueError, KeyError, TypeError) as e:
fail(f'unusable build receipt {path}: {e!r}')
unbuilt = [b for b in boards if b not in built]
- prefixes = tuple(f'{d.relative_to(ROOT)}/' for d in dirs)
+ prefixes = tuple(f'{d.relative_to(ROOT).as_posix()}/' for d in dirs)
differ = sorted({*(p for p in staged if pinned.get(p) != staged[p]),
*(p for p in pinned if p.startswith(prefixes) and p not in staged)})
why = [*checkout_drift(head, config_digest, config_path),
@@ -502,12 +538,17 @@
try:
helper('hil_lock') # the rig refuses an off-pin lib/agentrc too, but only after the wipe
except ImportError as e:
- fail(e)
+ if e.name != 'fcntl': # Windows: agentrc.load() passed the pin check before the POSIX-only lock ran
+ fail(e)
if args.build:
fail(f'--build would build on the rig, which gets binaries only; build locally with\n'
f' {build_command(config_path, select_boards(config, args), args.build_dir)}')
firmware = resolve_firmware(config, config_path, args)
- with tempfile.TemporaryDirectory(prefix='hil-remote-') as snap:
+ snap_parent = None
+ if os.name == 'nt': # in the checkout, where local_operand() can name it to rsync
+ snap_parent = ROOT / RUNS_DIR
+ snap_parent.mkdir(exist_ok=True)
+ with tempfile.TemporaryDirectory(prefix='hil-remote-', dir=snap_parent) as snap:
pinned = None
if receipt is not None:
firmware = snapshot(firmware, Path(snap))
@@ -515,6 +556,8 @@
pinned = check_receipt(receipt, config_path, boards,
[d for b in boards for d in variant_dirs(config, args.build_dir, b)],
staged_files(firmware, Path(snap)))
+ for p in (config_path, *firmware):
+ local_operand(p) # Windows: refuse one rsync cannot name now, not after the lease wiped the rig's tree
print(f'==> Setting up remote {remote}:{remote_dir}')
with remote_lease(remote, remote_dir, args.build_dir) as (remote_dir, token):
@@ -535,16 +578,16 @@
"uploading it; this run's table will REPLACE the previous one", file=sys.stderr)
else:
print('==> Uploading hil_report.json as the --accumulate merge base')
- if rsync(guard, '-q', str(sidecar), f'{remote}:{remote_dir}/') != 0:
+ if rsync(guard, '-q', sidecar, f'{remote}:{remote_dir}/') != 0:
fail('could not upload hil_report.json')
print('==> Copying the harness and config')
if rsync(guard, '-aqR', *HARNESS_FILES, f'{remote}:{remote_dir}/') != 0 or \
- rsync(guard, '-q', str(config_path), f'{remote}:{remote_dir}/test/hil/') != 0:
+ rsync(guard, '-q', config_path, f'{remote}:{remote_dir}/test/hil/') != 0:
fail('could not stage the harness')
print(f'==> Copying firmware from {len(firmware)} build dir(s) under {args.build_dir}/')
# one transfer: every dir lands under the same <build_dir>/, by its own name
- if rsync(guard, '-a', *FIRMWARE_FILTER, *map(str, firmware), f'{remote}:{remote_dir}/{args.build_dir}/') != 0:
+ if rsync(guard, '-a', *FIRMWARE_FILTER, *firmware, f'{remote}:{remote_dir}/{args.build_dir}/') != 0:
fail('could not stage the firmware')
# the harness and roster went from the live checkout, which may have moved since the receipt check
drift = pinned and list(filter(None, checkout_drift(*pinned, config_path)))
@@ -578,19 +621,83 @@
"""Publish a record under an exclusive flock taken before it appears, refusing an existing
one; returns the fd that holds the lock, which the kernel drops when this process dies."""
tmp = path.with_name(f'.{path.name}.{os.getpid()}.tmp')
- tmp.write_text(json.dumps(data) + '\n')
+ try:
+ f = open(tmp, 'x')
+ except FileExistsError:
+ if tmp.stat().st_nlink > 1: # a killed run's hidden name under a reused pid: writing it rewrites that record
+ raise
+ tmp.unlink()
+ f = open(tmp, 'x')
+ with f:
+ f.write(json.dumps(data) + '\n')
fd = os.open(tmp, os.O_RDONLY)
try:
- fcntl.flock(fd, fcntl.LOCK_EX)
+ if fcntl:
+ fcntl.flock(fd, fcntl.LOCK_EX)
+ else:
+ win_lock(fd)
os.link(tmp, path)
except BaseException:
os.close(fd)
+ tmp.unlink(missing_ok=True)
raise
- finally:
+ if fcntl: # Windows cannot remove a name of a file still open: run_recorded does, after the close
tmp.unlink(missing_ok=True)
return fd
+# msvcrt locks are mandatory and per byte: lock one far past the record, so readers still read it
+WIN_LOCK_OFFSET = 0x7FFFFFFE
+
+
+def win_lock(fd, block=True):
+ """The Windows stand-in for flock(LOCK_EX), held until win_unlock() or the handle closes;
+ BlockingIOError when not `block` and another handle holds it."""
+ pos = os.lseek(fd, 0, os.SEEK_CUR)
+ os.lseek(fd, WIN_LOCK_OFFSET, os.SEEK_SET)
+ try:
+ while True:
+ try:
+ msvcrt.locking(fd, msvcrt.LK_NBLCK, 1)
+ return
+ except OSError as e:
+ if e.errno not in (errno.EACCES, errno.EDEADLOCK):
+ raise
+ if not block:
+ raise BlockingIOError
+ time.sleep(0.05) # LK_LOCK gives up after about 10 s
+ finally:
+ os.lseek(fd, pos, os.SEEK_SET)
+
+
+def win_unlock(fd):
+ """Release win_lock() now: Windows drops a closed handle's lock only "after a delay"."""
+ pos = os.lseek(fd, 0, os.SEEK_CUR)
+ os.lseek(fd, WIN_LOCK_OFFSET, os.SEEK_SET)
+ try:
+ msvcrt.locking(fd, msvcrt.LK_UNLCK, 1)
+ finally:
+ os.lseek(fd, pos, os.SEEK_SET)
+
+
+# a waiter's probe holds the lock for a moment; a run holds it for its whole life
+PROBE_TRIES, PROBE_SECS = 10, 0.02
+
+
+def win_probe(fd):
+ """flock(LOCK_SH | LOCK_NB) for run_alive(): msvcrt has no shared lock, so two waiters' probes
+ contend; BlockingIOError only when the lock stays held across the retries."""
+ for _ in range(PROBE_TRIES):
+ try:
+ win_lock(fd, block=False)
+ except BlockingIOError:
+ time.sleep(PROBE_SECS)
+ continue
+ win_unlock(fd)
+ return
+ raise BlockingIOError
+
+
def pop_option(argv, name):
"""(the option's value or None, argv without it): the rest goes to hil_test.py."""
for i, a in enumerate(argv):
@@ -612,10 +719,12 @@
background run."""
started, done = run_paths(run_id)
started.parent.mkdir(exist_ok=True)
+ if not fcntl:
+ sweep_stale_tmps(started.parent)
try:
lock = publish_locked(started, {'runId': run_id, 'pid': os.getpid(), 'startedAt': time.time()})
except FileExistsError:
- fail(f'run id {run_id} was used before ({started.relative_to(ROOT)}); pick a new one')
+ fail(f'run id {run_id} was used before ({started.relative_to(ROOT).as_posix()}); pick a new one')
rc, reports = 1, []
try:
rc, reports = run(argv, receipt)
@@ -625,14 +734,36 @@
raise
finally:
write_json(done, {'runId': run_id, 'exit': rc, 'reports': reports})
- os.close(lock)
+ try:
+ if not fcntl:
+ win_unlock(lock)
+ finally:
+ os.close(lock)
+ if not fcntl:
+ with contextlib.suppress(OSError): # a waiter's probe holding the file open makes Windows refuse; the next sweep takes it
+ started.with_name(f'.{started.name}.{os.getpid()}.tmp').unlink(missing_ok=True)
+
+
+def sweep_stale_tmps(runs):
+ """Windows: remove the hidden second name of each started record a killed run left behind
+ (run_recorded removes its own after the close, which a kill skips). A live run holds its
+ file open, which Windows refuses to unlink, and a launch not yet linked has one name."""
+ for tmp in runs.glob('.*.started.json.*.tmp'):
+ try:
+ if tmp.stat().st_nlink > 1:
+ tmp.unlink()
+ except OSError:
+ pass
def run_alive(started):
"""Its run still holds the started record's lock."""
fd = os.open(started, os.O_RDONLY)
try:
- fcntl.flock(fd, fcntl.LOCK_SH | fcntl.LOCK_NB)
+ if fcntl:
+ fcntl.flock(fd, fcntl.LOCK_SH | fcntl.LOCK_NB)
+ else:
+ win_probe(fd)
return False
except BlockingIOError:
return True
diff --git a/.claude/skills/make-release/scripts/release.py b/.claude/skills/make-release/scripts/release.py
index 8177ecc..ca43dee 100755
--- a/.claude/skills/make-release/scripts/release.py
+++ b/.claude/skills/make-release/scripts/release.py
@@ -13,6 +13,7 @@
ancestor of HEAD, a first-parent commit that is not a merged PR."""
import argparse
import json
+import os
import re
import subprocess
import sys
@@ -58,7 +59,7 @@
raise Refused(f'version must be X.Y.Z, got {version!r}')
major, minor, rev = version.split('.')
option_h = root / 'src/tusb_option.h'
- current = current_version(option_h.read_text())
+ current = current_version(option_h.read_text(encoding='utf-8'))
if current == version:
raise Refused(f'{version} is already the version in src/tusb_option.h')
@@ -68,26 +69,26 @@
(r'(#define TUSB_VERSION_MINOR *) \d+', rf'\g<1> {minor}'),
(r'(#define TUSB_VERSION_REVISION *) \d+', rf'\g<1> {rev}'),
],
- root / 'repository.yml': repository_edits((root / 'repository.yml').read_text(), version, major),
+ root / 'repository.yml': repository_edits((root / 'repository.yml').read_text(encoding='utf-8'), version, major),
root / 'library.json': [(r'( {4}"version":) "\d+\.\d+\.\d+"', rf'\1 "{version}"')],
root / 'sonar-project.properties': [(r'(sonar\.projectVersion=)\d+\.\d+\.\d+', rf'\g<1>{version}')],
}
pending = []
for path, subs in edits.items(): # validate everything first: a refusal writes nothing
- before = path.read_text()
+ before = path.read_text(encoding='utf-8')
after = before
for pattern, repl in subs:
after = re.sub(pattern, repl, after, count=1)
if after == before:
- raise Refused(f'{path.relative_to(root)}: no line matched the version pattern; nothing written')
+ raise Refused(f'{path.relative_to(root).as_posix()}: no line matched the version pattern; nothing written')
pending.append((path, after))
for path, after in pending:
- path.write_text(after)
+ path.write_bytes(after.encode('utf-8')) # as on POSIX, no CRLF; write_text's newline= needs 3.10
return [path for path, _ in pending]
def git(root, *args):
- r = subprocess.run(['git', '-C', str(root), *args], capture_output=True, text=True)
+ r = subprocess.run(['git', '-C', str(root), *args], capture_output=True, encoding='utf-8')
if r.returncode != 0:
raise Refused(f'git {" ".join(args)} failed: {r.stderr.strip()}')
return r.stdout
@@ -120,7 +121,7 @@
"""One PR's number, title, labels and author; a few hundred calls in a row hit transient API errors."""
for attempt in range(attempts):
r = subprocess.run(['gh', 'pr', 'view', str(number), '--json', 'number,title,labels,author'],
- cwd=root, capture_output=True, text=True)
+ cwd=root, capture_output=True, encoding='utf-8')
if r.returncode == 0:
return json.loads(r.stdout)
raise Refused(f'gh pr view {number} failed {attempts} times: {r.stderr.strip()}')
@@ -142,11 +143,14 @@
help='the previous release tag; PRs are those reachable from --head but not from it')
q.add_argument('--head', default='HEAD', help='the release branch (default HEAD)')
a = p.parse_args()
+ if os.name == 'nt': # PR titles are UTF-8 whatever the console's code page
+ sys.stdout.reconfigure(encoding='utf-8')
+ sys.stderr.reconfigure(encoding='utf-8', errors='backslashreplace')
root = a.root.resolve()
try:
if a.cmd == 'bump':
for path in bump(root, a.version):
- print(f'{a.version} -> {path.relative_to(root)}')
+ print(f'{a.version} -> {path.relative_to(root).as_posix()}')
print(f'now add docs/changelog/{a.version}.md and list it first in docs/changelog/index.rst')
return 0
prs = fetch(root, merged_prs(root, a.prev, a.head))
diff --git a/.claude/skills/update-sponsor/update_sponsor.py b/.claude/skills/update-sponsor/update_sponsor.py
index d09c8b2..085713f 100644
--- a/.claude/skills/update-sponsor/update_sponsor.py
+++ b/.claude/skills/update-sponsor/update_sponsor.py
@@ -17,6 +17,7 @@
import difflib
import hashlib
import json
+import os
import re
import subprocess
import tempfile
@@ -61,7 +62,7 @@
def gh(*args, **kw):
- out = subprocess.run(["gh", *args], capture_output=True, text=True, **kw)
+ out = subprocess.run(["gh", *args], capture_output=True, encoding="utf-8", **kw)
if out.returncode:
sys.exit(f"gh {' '.join(args[:2])} failed:\n{out.stderr.strip()}")
return out.stdout
@@ -187,25 +188,29 @@
new_text.splitlines(), n=0) if l.startswith("+")]
if not added:
return []
- with tempfile.NamedTemporaryFile("w", suffix=".txt", delete=True) as probe:
- # outside the repo and not dot-prefixed: codespell skips hidden files unless
- # .codespellrc is picked up from cwd, which would make this guard fail open
- probe.write("\n".join(added) + "\n")
- probe.flush()
- try:
- run = subprocess.run(["codespell", "--ignore-words", str(IGNORE_WORDS), probe.name],
- capture_output=True, text=True)
- except FileNotFoundError:
- print(" note: codespell not on PATH - generated block is UNCHECKED")
- return []
+ # closed before codespell reads it: Windows cannot open a file a second time while it is open
+ probe = tempfile.NamedTemporaryFile("w", suffix=".txt", delete=False, encoding="utf-8")
+ try:
+ with probe:
+ # outside the repo and not dot-prefixed: codespell skips hidden files unless
+ # .codespellrc is picked up from cwd, which would make this guard fail open
+ probe.write("\n".join(added) + "\n")
+ run = subprocess.run(["codespell", "--ignore-words", str(IGNORE_WORDS), probe.name],
+ capture_output=True, encoding="utf-8", errors="replace")
+ except FileNotFoundError:
+ print(" note: codespell not on PATH - generated block is UNCHECKED")
+ return []
+ finally:
+ Path(probe.name).unlink(missing_ok=True)
if run.returncode not in (0, 65): # 65 = typos found; anything else is a tool error
print(f" note: codespell failed (rc={run.returncode}) - generated block is UNCHECKED")
return []
out = run.stdout
+ skip = 2 + probe.name.count(":") # <file>:<line>: <word> ==> <fix>, and a Windows path has a drive colon
# v2.2.4 (the pinned hook) lowercases dictionary keys before testing ignore-words,
# so a cased entry would never match and -w would rewrite the login anyway.
- return sorted({l.split(":", 2)[2].split("==>")[0].strip().lower()
- for l in out.splitlines() if l.count(":") >= 2})
+ return sorted({l.split(":", skip)[skip].split("==>")[0].strip().lower()
+ for l in out.splitlines() if l.count(":") >= skip})
def readme_diff(original, new_text):
@@ -404,6 +409,9 @@
only.add_argument("--readme-only", action="store_true", help="skip the label pass")
only.add_argument("--labels-only", action="store_true", help="skip the README pass")
args = p.parse_args()
+ if os.name == "nt": # labels carry emoji, names and titles any script, whatever the console's code page
+ sys.stdout.reconfigure(encoding="utf-8")
+ sys.stderr.reconfigure(encoding="utf-8", errors="backslashreplace")
if args.rules:
return print_rules()
diff --git a/.claude/skills/usbtest/SKILL.md b/.claude/skills/usbtest/SKILL.md
index 5f160bc..55743f1 100644
--- a/.claude/skills/usbtest/SKILL.md
+++ b/.claude/skills/usbtest/SKILL.md
@@ -40,9 +40,10 @@
```
`--after leave` keeps usbtest running for a debug session. `--variant` is required when the board
-has several. It refuses with exit 2 and the reason before touching hardware, including while any
-battery or another `run_case.py` runs on the host; pass `--allow-concurrent` only when another
-battery's load on the board's host controller does not matter for what you are measuring. A wedged device is never parked. The last stdout
+has several. It refuses with exit 2 and the reason before touching hardware, including on a host
+that is not Linux and while any battery or another `run_case.py` runs on the host; pass
+`--allow-concurrent` only when another battery's load on the board's host controller does not
+matter for what you are measuring. A wedged device is never parked. The last stdout
line is its JSON verdict; each of its `cases` carries `usbtest.py`'s `detail`, and `stderr` and
`dmesg` when it captured them.
diff --git a/.claude/skills/usbtest/scripts/kernel_src.py b/.claude/skills/usbtest/scripts/kernel_src.py
index b10db2f..48afa18 100644
--- a/.claude/skills/usbtest/scripts/kernel_src.py
+++ b/.claude/skills/usbtest/scripts/kernel_src.py
@@ -164,6 +164,8 @@
default=Path(os.environ.get('XDG_CACHE_HOME') or Path.home() / '.cache') / 'tinyusb-usbtest',
help='where fetched files are kept (default: %(default)s)')
args = p.parse_args()
+ if os.name == 'nt': # the kernel's source is UTF-8 whatever the console's code page
+ sys.stdout.reconfigure(encoding='utf-8')
if args.tag:
if not RE_TAG.fullmatch(args.tag):
@@ -186,7 +188,7 @@
if args.case is None:
return 0
- lines = usbtest_c.read_text(errors='replace').splitlines()
+ lines = usbtest_c.read_text(encoding='utf-8', errors='replace').splitlines()
print(f'\n== case {args.case} in usbtest_do_ioctl()')
for n, text in case_block(lines, args.case):
print(f'{n:5d} {text}')
diff --git a/.claude/skills/usbtest/scripts/run_case.py b/.claude/skills/usbtest/scripts/run_case.py
index 19ba4bb..0a659df 100644
--- a/.claude/skills/usbtest/scripts/run_case.py
+++ b/.claude/skills/usbtest/scripts/run_case.py
@@ -15,6 +15,7 @@
process is alive on this host, checked before and again after taking the lock. That check is
host-wide and does not see a hil_test.py started after it: --allow-concurrent skips it only when
another battery's load on this board's host controller does not matter for what you are measuring.
+A host that is not Linux is refused first, once argparse has accepted the command line.
Flashes with the roster's flasher from cmake-build/cmake-build-<variant> (the build skill's
--shared --variants layout), waits for cafe:4010 with the board's serial, and runs usbtest.py
@@ -24,7 +25,8 @@
stdout ends with one JSON line: {"pass", "board", "variant", "tests", "cases", "wedged",
"boardState", "error"}, each case with usbtest.py's num, name, status, detail and, when it captured
them, stderr and dmesg. Exit 0 every case passed, 1 a case failed or the run failed after the
-lock was taken, 2 refused before any hardware action. A SIGTERM (`hil_lock.py release` sends one
+lock was taken, 2 refused before any hardware action; a command line argparse rejects exits 2
+with its usage on stderr and no JSON line. A SIGTERM (`hil_lock.py release` sends one
to the lock holder) kills the running flasher or usbtest.py, reports boardState unknown with a
verdict line, then releases the lock.
"""
@@ -37,6 +39,33 @@
import time
from pathlib import Path
+
+def build_parser():
+ p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+ p.add_argument('--config', required=True, help="this host's HIL config json (the HIL contract)")
+ p.add_argument('--board', required=True)
+ p.add_argument('--tests', required=True, help='comma-separated usbtest case numbers')
+ p.add_argument('--after', required=True, choices=('park', 'leave'),
+ help='park: flash device/board_test afterwards; leave: keep usbtest running')
+ p.add_argument('--variant', help='roster variant, required when the board has several')
+ p.add_argument('--timeout', type=int, default=60, help='per-case timeout in seconds (default 60, as HIL)')
+ p.add_argument('--allow-concurrent', action='store_true',
+ help='skip the live-battery check; see the description first')
+ return p
+
+
+def new_report(args):
+ return {'pass': False, 'board': args.board, 'variant': args.variant, 'tests': args.tests,
+ 'cases': [], 'wedged': False, 'boardState': 'untouched', 'error': ''}
+
+
+if __name__ == '__main__' and sys.platform != 'linux': # after argparse, so --help works; before the harness imports
+ _report = new_report(build_parser().parse_args())
+ _report['error'] = 'run_case.py needs a Linux host (usbfs, /proc, the usbtest kernel module)'
+ print(f'error: {_report["error"]}', file=sys.stderr)
+ print(json.dumps(_report))
+ sys.exit(2)
+
ROOT = Path(__file__).resolve().parents[4]
HIL_DIR = ROOT / 'test' / 'hil'
sys.path.insert(0, str(HIL_DIR))
@@ -268,20 +297,8 @@
def main():
- p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
- p.add_argument('--config', required=True, help="this host's HIL config json (the HIL contract)")
- p.add_argument('--board', required=True)
- p.add_argument('--tests', required=True, help='comma-separated usbtest case numbers')
- p.add_argument('--after', required=True, choices=('park', 'leave'),
- help='park: flash device/board_test afterwards; leave: keep usbtest running')
- p.add_argument('--variant', help='roster variant, required when the board has several')
- p.add_argument('--timeout', type=int, default=60, help='per-case timeout in seconds (default 60, as HIL)')
- p.add_argument('--allow-concurrent', action='store_true',
- help='skip the live-battery check; see the description first')
- args = p.parse_args()
-
- report = {'pass': False, 'board': args.board, 'variant': args.variant, 'tests': args.tests,
- 'cases': [], 'wedged': False, 'boardState': 'untouched', 'error': ''}
+ args = build_parser().parse_args()
+ report = new_report(args)
def finish(code, error=''):
signal.signal(signal.SIGTERM, signal.SIG_IGN) # one verdict line; the release still follows
diff --git a/.claude/skills/usbtest/scripts/wedge_drill.py b/.claude/skills/usbtest/scripts/wedge_drill.py
index 21644fe..6787355 100755
--- a/.claude/skills/usbtest/scripts/wedge_drill.py
+++ b/.claude/skills/usbtest/scripts/wedge_drill.py
@@ -28,7 +28,8 @@
(the harness did not recover a wedge it was given), or "inconclusive" (no wedge was produced:
the case ended before the halt landed, the halt failed, or no case hung). `cleanup` is the drill's
own reset after a non-pass and never counts as harness recovery. Exit 0 only for a pass with the
-board parked; 1 otherwise; 2 refused.
+board parked; 1 otherwise; 2 refused (a command line argparse rejects: its usage on stderr, no JSON
+line).
"""
import argparse
import json
@@ -43,6 +44,35 @@
import time
from pathlib import Path
+
+def build_parser():
+ p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+ p.add_argument('--config', required=True, help="this host's HIL config json (the HIL contract)")
+ p.add_argument('--board', required=True)
+ 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")
+ return p
+
+
+def new_report(args):
+ return {'drill': 'inconclusive', 'board': args.board, 'variant': args.variant, 'reason': '',
+ 'halt': {}, 'cases': [], 'wedged': False, 'recovery': {}, 'device': {},
+ 'smoke': None, 'cleanup': None, 'boardState': 'untouched', 'error': ''}
+
+
+if __name__ == '__main__' and sys.platform != 'linux': # after argparse, so --help works; before run_case's harness imports
+ _report = new_report(build_parser().parse_args())
+ _report['error'] = 'wedge_drill.py needs a Linux host (usbfs, /proc, the usbtest kernel module)'
+ print(f'error: {_report["error"]}', file=sys.stderr)
+ print(json.dumps(_report))
+ sys.exit(2)
+
sys.path.insert(0, str(Path(__file__).resolve().parent))
import run_case # noqa: E402
from run_case import PROC, Refused, Terminated, hil_flash, hil_lock, hil_util, usb_observe, usbtest # noqa: E402
@@ -419,22 +449,8 @@
def main():
- p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
- p.add_argument('--config', required=True, help="this host's HIL config json (the HIL contract)")
- p.add_argument('--board', required=True)
- 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()
-
- report = {'drill': 'inconclusive', 'board': args.board, 'variant': args.variant, 'reason': '',
- 'halt': {}, 'cases': [], 'wedged': False, 'recovery': {}, 'device': {},
- 'smoke': None, 'cleanup': None, 'boardState': 'untouched', 'error': ''}
+ args = build_parser().parse_args()
+ report = new_report(args)
def finish(code):
signal.signal(signal.SIGTERM, signal.SIG_IGN)
diff --git a/.claude/test/test_build.py b/.claude/test/test_build.py
index 541ca05..552d367 100644
--- a/.claude/test/test_build.py
+++ b/.claude/test/test_build.py
@@ -795,6 +795,12 @@
self.assertEqual((rc, err.getvalue()), (3, ''))
self.assertEqual(sorted(out.split()), ['x', 'y'])
+ def test_the_child_writes_utf8_whatever_its_inherited_encoding(self):
+ # a Windows child writes a pipe in the ANSI code page; latin-1 stands in for it on any host
+ with mock.patch.dict(os.environ, {'PYTHONIOENCODING': 'latin-1'}):
+ rc, out = build.run([sys.executable, '-c', 'print("\u2192 \u00f8")'], False)
+ self.assertEqual((rc, out), (0, '\u2192 \u00f8\n'))
+
UTILS = Path(__file__).resolve().parents[2] / 'tools' / 'build_utils.py'
_uspec = importlib.util.spec_from_file_location('build_utils_under_test', UTILS)
diff --git a/.claude/test/test_build_doc.py b/.claude/test/test_build_doc.py
index 4e21604..121955d 100644
--- a/.claude/test/test_build_doc.py
+++ b/.claude/test/test_build_doc.py
@@ -1,54 +1,110 @@
"""Tests for the build-doc skill's build_doc.py: it runs sphinx-build strictly
by default, forwards the flags it documents, and propagates the build's exit
-code without inventing success."""
+code without inventing success. The docs and build dirs are temp dirs, never
+the checkout's docs/_build, which --clean removes."""
+import importlib.util
+import io
import os
+import shutil
import subprocess
import sys
import tempfile
+import types
import unittest
+from contextlib import redirect_stderr, redirect_stdout
from pathlib import Path
+from unittest import mock
SCRIPT = Path(__file__).resolve().parents[1] / 'skills' / 'build-doc' / 'scripts' / 'build_doc.py'
-ROOT = Path(__file__).resolve().parents[2]
-
-
-def run(args, rc=0):
- """Run build_doc.py with a stub sphinx-build on PATH that logs its argv and exits rc."""
- with tempfile.TemporaryDirectory() as d:
- stub = Path(d) / 'sphinx-build'
- log = Path(d) / 'argv'
- stub.write_text(f'#!/bin/sh\nprintf "%s\\n" "$@" > {log}\nexit {rc}\n')
- stub.chmod(0o755)
- env = dict(os.environ, PATH=f'{d}:{os.environ["PATH"]}')
- r = subprocess.run([sys.executable, str(SCRIPT), *args], capture_output=True, text=True, env=env)
- argv = log.read_text().split('\n')[:-1] if log.exists() else None
- return r, argv
+spec = importlib.util.spec_from_file_location('build_doc', SCRIPT)
+build_doc = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(build_doc)
class BuildDocTest(unittest.TestCase):
+ def setUp(self):
+ tmp = tempfile.TemporaryDirectory()
+ self.addCleanup(tmp.cleanup)
+ self.docs = Path(tmp.name) / 'docs'
+ self.build = self.docs / '_build'
+ for name, value in (('DOCS', self.docs), ('BUILD', self.build)):
+ patch = mock.patch.object(build_doc, name, value)
+ patch.start()
+ self.addCleanup(patch.stop)
+
+ def run_main(self, args, rc=0):
+ """build_doc.py's main() with its sphinx-build call recorded and answered with rc, in-process:
+ a stub on PATH that a host cannot execute would let the real sphinx-build run."""
+ calls = []
+ out, err = io.StringIO(), io.StringIO()
+
+ def call(cmd):
+ calls.append(cmd)
+ self.build_existed = self.build.exists()
+ return rc
+ with mock.patch.object(build_doc.subprocess, 'call', call), \
+ mock.patch.object(sys, 'argv', [str(SCRIPT), *args]), redirect_stdout(out), redirect_stderr(err):
+ try:
+ code = build_doc.main()
+ except SystemExit as e:
+ code = e.code
+ if calls:
+ assert calls[0][0] == 'sphinx-build', calls
+ return types.SimpleNamespace(returncode=code, stdout=out.getvalue(), stderr=err.getvalue()), \
+ calls[0][1:] if calls else None
+
def test_default_is_html_strict_from_docs_into_docs_build(self):
- r, argv = run([])
+ r, argv = self.run_main([])
self.assertEqual(r.returncode, 0, r.stderr)
- self.assertEqual(argv, ['-b', 'html', '-W', str(ROOT / 'docs'), str(ROOT / 'docs' / '_build')])
+ self.assertEqual(argv, ['-b', 'html', '-W', str(self.docs), str(self.build)])
self.assertIn('Docs built:', r.stdout)
def test_no_strict_drops_W_and_nothing_else(self):
- r, argv = run(['--no-strict'])
+ r, argv = self.run_main(['--no-strict'])
self.assertEqual(r.returncode, 0, r.stderr)
self.assertNotIn('-W', argv)
self.assertEqual(argv[:2], ['-b', 'html'])
+ def test_clean_removes_the_build_dir_first(self):
+ (self.build / 'stale').mkdir(parents=True)
+ r, argv = self.run_main(['-c'])
+ self.assertEqual(r.returncode, 0, r.stderr)
+ self.assertFalse(self.build_existed, 'removed before sphinx-build ran')
+
def test_sphinx_failure_is_the_exit_code_and_no_success_line(self):
- r, argv = run([], rc=2)
+ r, argv = self.run_main([], rc=2)
self.assertEqual(r.returncode, 2)
self.assertIsNotNone(argv)
self.assertNotIn('Docs built:', r.stdout)
def test_unknown_flag_is_refused_before_sphinx_runs(self):
- r, argv = run(['-W'])
+ r, argv = self.run_main(['-W'])
self.assertEqual(r.returncode, 2)
self.assertIsNone(argv)
+@unittest.skipIf(os.name == 'nt', 'the stub is a #! script Windows cannot exec: a miss would run the real sphinx-build')
+class ProcessTest(unittest.TestCase):
+ """The script as a process, copied into a temp checkout so its docs/_build is a temp dir:
+ main()'s status is the exit status."""
+
+ def test_the_exit_status_is_mains(self):
+ for rc in (0, 3):
+ with self.subTest(rc=rc), tempfile.TemporaryDirectory() as d:
+ script = Path(d) / '.claude' / 'skills' / 'build-doc' / 'scripts' / 'build_doc.py'
+ script.parent.mkdir(parents=True)
+ shutil.copy(SCRIPT, script)
+ stub = Path(d) / 'bin' / 'sphinx-build'
+ stub.parent.mkdir()
+ stub.write_text(f'#!/bin/sh\nprintf "%s\\n" "$@" > {d}/argv\nexit {rc}\n')
+ stub.chmod(0o755)
+ env = dict(os.environ, PATH=f'{stub.parent}{os.pathsep}{os.environ["PATH"]}')
+ r = subprocess.run([sys.executable, str(script)], capture_output=True, text=True, env=env, timeout=60)
+ self.assertEqual(r.returncode, rc, r.stderr)
+ self.assertEqual((Path(d) / 'argv').read_text().split('\n')[:-1],
+ ['-b', 'html', '-W', str(Path(d).resolve() / 'docs'), str(Path(d).resolve() / 'docs' / '_build')])
+ self.assertEqual('Docs built:' in r.stdout, rc == 0)
+
+
if __name__ == '__main__':
unittest.main()
diff --git a/.claude/test/test_coverage_audit.py b/.claude/test/test_coverage_audit.py
index 651f978..09b4c61 100644
--- a/.claude/test/test_coverage_audit.py
+++ b/.claude/test/test_coverage_audit.py
@@ -4,10 +4,13 @@
import importlib.util
import json
import os
+import shlex
import subprocess
+import sys
import tempfile
import unittest
from pathlib import Path
+from unittest import mock
SCRIPT = Path(__file__).resolve().parents[1] / 'skills' / 'build' / 'scripts' / 'coverage_audit.py'
spec = importlib.util.spec_from_file_location('coverage_audit', SCRIPT)
@@ -51,6 +54,16 @@
self.assertEqual(ca.compiler_of({'command': '/x/riscv-none-elf-gcc -c a.c'}), '/x/riscv-none-elf-gcc')
self.assertEqual(ca.compiler_of({'arguments': ['arm-none-eabi-gcc', '-c']}), 'arm-none-eabi-gcc')
+ def test_compiler_of_reads_a_quoted_compiler_path(self):
+ gcc = 'C:/Program Files/Arm GNU Toolchain/bin/arm-none-eabi-gcc.exe' if os.name == 'nt' else '/opt/arm gnu/bin/arm-none-eabi-gcc'
+ join = subprocess.list2cmdline if os.name == 'nt' else shlex.join # as CMake writes it on each host
+ self.assertEqual(ca.compiler_of({'command': join([gcc, '-c', 'a.c'])}), gcc)
+
+ def test_its_family_json_leaves_the_audited_checkouts_to_import_build(self):
+ # graph imports the audited checkout's tools/build.py, whose `import family_json` must find that checkout's
+ self.assertEqual(Path(ca.family_json.__file__), SCRIPT.parents[4] / 'tools' / 'family_json.py')
+ self.assertIsNot(sys.modules.get('family_json'), ca.family_json)
+
def test_global_assembly_label_is_live_local_one_is_not(self):
self.assertTrue(ca.defines_symbols(' 25: 00000000 2 NOTYPE GLOBAL DEFAULT 7 Default_Handler\n'))
self.assertFalse(ca.defines_symbols(' 4: 00000000 0 NOTYPE LOCAL DEFAULT 1 $t\n'))
@@ -100,6 +113,25 @@
self.assertIsNone(ca.rel('/repo/cmake-build/cmake-build-x/gen.h', '/repo'))
self.assertEqual(ca.rel('../../src/tusb.c', '/repo', '/repo/cmake-build/b'), 'src/tusb.c')
+ @unittest.skipUnless(os.name == 'nt', 'drive letters are Windows paths')
+ def test_rel_is_posix_and_another_drive_is_outside(self):
+ self.assertEqual(ca.rel('E:\\repo\\src\\tusb.h', 'E:/repo'), 'src/tusb.h')
+ ca.other_drive.clear()
+ self.assertIsNone(ca.rel('C:/Tool/arm/include/stdint.h', 'E:/repo'))
+ self.assertEqual(ca.other_drive, {os.path.normpath('C:/Tool/arm/include/stdint.h')}, 'counted for the graph note')
+
+ def test_an_lto_object_is_asked_through_its_compilers_own_nm(self):
+ for compiler, nm in (('/x/arm-none-eabi-gcc', '/x/arm-none-eabi-gcc-nm'),
+ ('C:\\x\\arm-none-eabi-gcc.exe', 'C:\\x\\arm-none-eabi-gcc-nm.exe')):
+ calls = []
+
+ def run(cmd, cwd):
+ calls.append(cmd[0])
+ return subprocess.CompletedProcess(cmd, 0, '__gnu_lto_slim\n' if cmd[0] == 'readelf' else '00000000 T f\n', '')
+ with mock.patch.object(ca, 'run', run):
+ self.assertTrue(ca.is_live('x.obj', compiler, '.'))
+ self.assertEqual(calls, ['readelf', nm])
+
class ScopeTest(unittest.TestCase):
def test_example_cmake_reaches_its_example_only(self):
@@ -166,8 +198,9 @@
class FakeBuild:
"""A built tree on disk (sources, compile_commands.json) whose ninja and readelf answers
are canned: two examples, each linking a different library archive."""
- def __init__(self, root, fail=None, built=('device/a', 'device/b')):
+ def __init__(self, root, fail=None, built=('device/a', 'device/b'), extra_deps=()):
self.root, self.bd, self.fail = root, os.path.join(root, 'cmake-build', 'b'), fail
+ self.extra_deps = ''.join(f' {d}\n' for d in extra_deps) # in every object's deps
for f in ('examples/device/a/main.c', 'examples/device/b/main.c', 'lib/os/os.c', 'lib/os/os.h',
'lib/usb/usb.c', 'hw/bsp/f/start.S', 'hw/bsp/f/mem.ld', 'src/dead.c', 'src/prog.pio'):
os.makedirs(os.path.dirname(os.path.join(root, f)), exist_ok=True)
@@ -207,7 +240,7 @@
out = ''.join(f'{t}: x\n' for t in self.query)
elif cmd[4] == 'deps':
out = ''.join(f'{o}: #deps 2, deps mtime 1 (VALID)\n {os.path.join(self.root, s)}\n'
- f' {os.path.join(self.root, "lib/os/os.h")}\n'
+ f' {os.path.join(self.root, "lib/os/os.h")}\n' + self.extra_deps
+ (f' {self.bd}/gen/prog.h\n' if o.endswith('usb.c.obj') else '') + '\n'
for o, s in self.objs.items() if not o.endswith('start.S.obj'))
elif cmd[4] == 'query':
@@ -219,10 +252,10 @@
class ExtractTest(unittest.TestCase):
- def extract(self, fail=None, built=('device/a', 'device/b')):
+ def extract(self, fail=None, built=('device/a', 'device/b'), extra_deps=lambda root: ()):
orig = ca.run
with tempfile.TemporaryDirectory() as root:
- fb = FakeBuild(root, fail, built)
+ fb = FakeBuild(root, fail, built, extra_deps(root))
ca.run = fb
try:
return ca.extract(fb.bd, root)
@@ -243,6 +276,18 @@
with self.assertRaises(ca.ExtractError):
self.extract(built=())
+ @unittest.skipUnless(os.name == 'nt', 'drive letters are Windows paths')
+ def test_a_dependency_on_another_drive_is_outside(self):
+ def header(root):
+ drive = 'D:' if os.path.splitdrive(root)[0].upper() == 'C:' else 'C:'
+ return [f'{drive}/Tool/arm/include/stdint.h']
+ ca.other_drive.clear()
+ g = self.extract(extra_deps=header)
+ self.assertEqual(g['examples']['device/a'], ['examples/device/a/main.c', 'hw/bsp/f/link.ld', 'hw/bsp/f/mem.ld',
+ 'hw/bsp/f/start.S', 'lib/os/os.c', 'lib/os/os.h'])
+ self.assertEqual({os.path.splitdrive(p)[1] for p in ca.other_drive},
+ {os.path.normpath('/Tool/arm/include/stdint.h')}, 'counted for the graph note')
+
def test_a_failing_ninja_tool_fails_the_extraction(self):
for tool in ('targets', 'deps', 'query', 'commands'):
with self.assertRaises(ca.ExtractError, msg=tool):
diff --git a/.claude/test/test_gen_doc.py b/.claude/test/test_gen_doc.py
index 9002fcb..4f9a115 100644
--- a/.claude/test/test_gen_doc.py
+++ b/.claude/test/test_gen_doc.py
@@ -39,7 +39,8 @@
bsp = root / 'hw' / 'bsp'
(bsp / 'acme' / 'boards' / 'acme_one').mkdir(parents=True)
(bsp / 'acme' / 'boards' / 'acme_bare').mkdir()
- (bsp / 'acme' / 'family.c').write_text(METADATA_FAMILY)
+ # UTF-8 beyond the metadata block, as several real family.c have: read as UTF-8 on every host
+ (bsp / 'acme' / 'family.c').write_text('// Acme \u2192 \u2714\n' + METADATA_FAMILY, encoding='utf-8')
(bsp / 'acme' / 'boards' / 'acme_one' / 'board.h').write_text(METADATA_BOARD)
(bsp / 'acme' / 'boards' / 'acme_bare' / 'board.h').write_text('#define BOARD_H_\n')
(bsp / 'nometa' / 'boards' / 'nometa_x').mkdir(parents=True)
@@ -50,9 +51,9 @@
(root / 'test' / 'hil' / 'tinyusb.json').write_text(json.dumps({'boards': [
{'name': 'acme_one', 'tests': {'device': True, 'host': True, 'dual': False},
'flasher': {'name': 'jlink'}, 'variant': [{'name': 'acme_one_hs'}],
- 'note': 'needs a | jumper'},
+ 'note': 'needs a | jumper \u00f8'},
{'name': 'acme_two', 'tests': {'only': ['host/cdc_msc', 'device/hid']}},
- ]}))
+ ]}, ensure_ascii=False), encoding='utf-8')
(root / 'test' / 'hil' / 'hfp.json').write_text(json.dumps({'boards': []}))
@@ -69,6 +70,24 @@
self.assertEqual(gen_doc.metadata(Path(d) / 'absent.h'), {})
+class BoardsTest(unittest.TestCase):
+ def test_families_and_boards_sort_by_name_case_sensitively_on_every_host(self):
+ # the committed boards.rst is Linux's order; paths here compare case-insensitively, as on Windows
+ with tempfile.TemporaryDirectory() as d:
+ bsp = Path(d)
+ for family in ('acme', 'Zeta'):
+ (bsp / family / 'family.c').parent.mkdir()
+ (bsp / family / 'family.c').write_text(METADATA_FAMILY)
+ for board, has_header in (('board_x', True), ('Board_Y', True), ('bare', False), ('Bare_', False)):
+ (bsp / family / 'boards' / f'{family}_{board}').mkdir(parents=True)
+ if has_header:
+ (bsp / family / 'boards' / f'{family}_{board}' / 'board.h').write_text(METADATA_BOARD)
+ with mock.patch.object(type(bsp), '__lt__', lambda a, b: str(a).casefold() < str(b).casefold()):
+ vendors, skipped = gen_doc.boards(bsp)
+ self.assertEqual(list(vendors['Acme']), ['Zeta_Board_Y', 'Zeta_board_x', 'acme_Board_Y', 'acme_board_x'])
+ self.assertEqual([p.name for p in skipped], ['Zeta_Bare_', 'Zeta_bare', 'acme_Bare_', 'acme_bare'])
+
+
class MdTableTest(unittest.TestCase):
def test_github_table_pads_headers_by_two_and_strips_cells(self):
self.assertEqual(gen_doc.md_table(['H', 'Name'], [['a ', 'long value'], ['bb', '']]).split('\n'), [
@@ -86,8 +105,8 @@
fixture(root)
r = subprocess.run([sys.executable, str(GEN_DOC), '--root', str(root)], capture_output=True, text=True)
self.assertEqual(r.returncode, 0, r.stderr)
- boards = (root / 'docs/reference/boards.rst').read_text()
- hil = (root / 'docs/reference/hil_boards.md').read_text()
+ boards = (root / 'docs/reference/boards.rst').read_text(encoding='utf-8')
+ hil = (root / 'docs/reference/hil_boards.md').read_text(encoding='utf-8')
deps = (root / 'docs/reference/dependencies.rst').read_text()
self.assertIn('Acme\n----\n\n', boards)
self.assertIn('acme_one Acme One acme https://acme.example/one', boards)
@@ -101,8 +120,8 @@
'',
])
self.assertTrue(hil.startswith(gen_doc.HEADER + '\n\n### ci rig\n\n2 boards, from `test/hil/tinyusb.json`.'))
- self.assertIn('| acme_one | device, host | jlink | acme_one_hs | needs a \\| jumper |', hil)
- self.assertIn('| acme_two | device, host | | | |', hil)
+ self.assertIn('| acme_one | device, host | jlink | acme_one_hs | needs a \\| jumper \u00f8 |', hil)
+ self.assertIn('| acme_two | device, host | | | |', hil)
self.assertNotIn('hfp rig', hil)
self.assertIn('lib/fatfs', deps)
diff --git a/.claude/test/test_hil_remote.py b/.claude/test/test_hil_remote.py
index c3d28d5..6fe07e0 100644
--- a/.claude/test/test_hil_remote.py
+++ b/.claude/test/test_hil_remote.py
@@ -1,10 +1,12 @@
"""hil_remote.py against a local "rig": a fake ssh runs each remote command under sh with a
clean environment and a temp HOME, a fake rsync strips the host prefix, and a fake python3
in that HOME's ~/.local/bin stands in for hil_test.py. So the quoting, the tilde expansion,
-the PATH shim and every refusal-before-wipe execute for real. Plumbing only: the real
-transport is proved by a run on ci.lan."""
-import fcntl
+the PATH shim and every refusal-before-wipe execute for real. Windows cannot exec the #!
+fakes, so there only the Refusals and Units run. Plumbing only: the real transport is proved
+by a run on ci.lan."""
+import contextlib
import importlib.util
+import io
import json
import os
import shlex
@@ -12,8 +14,10 @@
import subprocess
import sys
import tempfile
+import types
import unittest
from pathlib import Path
+from unittest import mock
REPO = Path(__file__).resolve().parents[2]
SCRIPT = REPO / '.claude' / 'skills' / 'hil-contract' / 'scripts' / 'hil_remote.py'
@@ -24,6 +28,8 @@
REAL_RSYNC = shutil.which('rsync')
needs_rsync = unittest.skipUnless(REAL_RSYNC, 'rsync not installed; the fake rsync execs the real one')
needs_flock = unittest.skipUnless(shutil.which('flock', path='/usr/bin:/bin'), 'flock not on the fake rig PATH (/usr/bin:/bin)')
+# the fake ssh, rsync and python3 are #! scripts Windows cannot exec: a run there would reach the real ones
+needs_fake_rig = unittest.skipIf(os.name == 'nt', 'the fake rig is #! scripts and a POSIX shell')
FAKE_SSH = f'''#!{sys.executable}
import json, os, subprocess, sys
@@ -116,6 +122,9 @@
PATH=f'{self.bin}:{os.environ["PATH"]}', FAKE_LOG=str(self.log),
FAKE_HOME=str(self.home), FAKE_RUN=str(self.run_file))
e.pop('CONFIG', None)
+ if os.name == 'nt':
+ # only the refusals run here: a REMOTE_DIR they accept, and no real ssh or rsync on PATH
+ e.update(REMOTE_DIR='/tmp/hil-remote-test', PATH=str(self.bin))
return e
def hil_remote(self, *argv, **env):
@@ -210,6 +219,7 @@
self.refused(self.hil_remote('-b', 'alpha', '-B', bad), '-B must be')
+@needs_fake_rig
@needs_rsync
@needs_flock
class Staging(Rig):
@@ -321,6 +331,7 @@
self.assertFalse((self.remote / 'stale').exists())
+@needs_fake_rig
@needs_rsync
@needs_flock
class TheRun(Rig):
@@ -373,11 +384,13 @@
for a in rsync), rsync)
+@needs_fake_rig
@needs_flock
class RemoteLock(Rig):
"""A second run sharing REMOTE_DIR must not rm -rf the tree the first one is running from."""
def lock(self):
+ import fcntl
lk = open(f'{self.remote}.lock', 'a')
self.addCleanup(lk.close)
fcntl.flock(lk, fcntl.LOCK_EX | fcntl.LOCK_NB)
@@ -476,6 +489,7 @@
self.assertEqual([c['tool'] for c in self.calls()][-1], 'ssh', 'copy_back ran after the refusal')
+@needs_fake_rig
@needs_rsync
@needs_flock
class Reports(Rig):
@@ -546,6 +560,115 @@
self.assertEqual(hil_remote.sidecar_matches(p, CONFIG), ok, body)
p.write_text('{')
self.assertFalse(hil_remote.sidecar_matches(p, CONFIG))
+ p.write_text(json.dumps({'rows': [{'board': 'alpha', 'note': '\u2714'}]}, ensure_ascii=False), encoding='utf-8')
+ self.assertTrue(hil_remote.sidecar_matches(p, CONFIG), 'the rig writes it as UTF-8')
+
+ def test_staged_files_are_named_as_rsync_and_the_rig_name_them(self):
+ with tempfile.TemporaryDirectory() as td, mock.patch.object(hil_remote, 'ROOT', Path(td)):
+ d = Path(td) / 'cmake-build' / 'cmake-build-alpha'
+ (d / 'device' / 'cdc_msc').mkdir(parents=True)
+ for f in ('cdc_msc.elf', 'cdc_msc.map'):
+ (d / 'device' / 'cdc_msc' / f).write_text(f)
+ files = hil_remote.staged_files([d])
+ self.assertEqual(list(files), ['cmake-build/cmake-build-alpha/device/cdc_msc/cdc_msc.elf'])
+ self.assertEqual(hil_remote.example_firmware(d, files), {'device/cdc_msc'})
+
+ def test_no_rsync_is_refused_before_the_lease(self):
+ def popen(*a, **k):
+ raise AssertionError('ssh started')
+ cases = [('none', None, None)]
+ if os.name == 'nt': # subprocess cannot start a .cmd by bare name, and the cwd is not PATH
+ cases += [('a .cmd on PATH', 'path', 'rsync.cmd'), ('an .exe in the cwd', 'cwd', 'rsync.exe')]
+ for case, where, name in cases:
+ with self.subTest(case), tempfile.TemporaryDirectory() as path, tempfile.TemporaryDirectory() as cwd:
+ if name:
+ (Path(path if where == 'path' else cwd) / name).write_bytes(b'')
+ old = os.getcwd()
+ os.chdir(cwd)
+ try:
+ with mock.patch.dict(os.environ, {'PATH': os.pathsep.join(['', path])}), \
+ mock.patch.object(hil_remote.subprocess, 'Popen', popen), \
+ contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit) as e:
+ with hil_remote.remote_lease('rig', '/tmp/x', 'cmake-build'):
+ pass
+ finally:
+ os.chdir(old)
+ self.assertIn('rsync is not on PATH', str(e.exception.code))
+
+ def test_windows_takes_rsync_exe_only_from_an_absolute_path_dir(self):
+ # the os module as Windows' rsync_program sees it, on any host
+ with tempfile.TemporaryDirectory() as path, tempfile.TemporaryDirectory() as cwd:
+ for d in (path, cwd):
+ (Path(d) / 'rsync.exe').write_bytes(b'')
+ old = os.getcwd()
+ os.chdir(cwd)
+ try:
+ for entries, found in ((['.', 'bin', ''], None), (['.', path], os.path.join(path, 'rsync.exe'))):
+ nt = types.SimpleNamespace(name='nt', path=os.path, pathsep=os.pathsep,
+ environ={'PATH': os.pathsep.join(entries)})
+ with self.subTest(entries), mock.patch.object(hil_remote, 'os', nt):
+ self.assertEqual(hil_remote.rsync_program(), found)
+ finally:
+ os.chdir(old)
+
+ @unittest.skipUnless(os.name == 'nt', 'rsync.exe is the Windows name')
+ def test_windows_runs_the_rsync_exe_found_on_path(self):
+ calls = []
+ with tempfile.TemporaryDirectory() as path:
+ (Path(path) / 'rsync.exe').write_bytes(b'')
+ with mock.patch.dict(os.environ, {'PATH': path}), \
+ mock.patch.object(hil_remote.subprocess, 'run', lambda argv, **kw: calls.append(argv) or mock.Mock(returncode=0)):
+ hil_remote.rsync(('/tmp/t', 'tok'), '-q', 'rig:/tmp/t/x', hil_remote.ROOT / 'x')
+ self.assertEqual(calls[0][0], os.path.join(path, 'rsync.exe'))
+
+ def test_rsync_names_local_files_from_the_checkout_and_passes_remote_specs_as_they_are(self):
+ # a one-letter ssh host must not be read as a drive letter, the checkout's own or another
+ calls = []
+ local = hil_remote.ROOT / 'hil_report.json'
+ hosts = ('rig', hil_remote.ROOT.drive[:1].lower() or 'e', 'r')
+ with mock.patch.object(hil_remote, 'rsync_program', return_value='rsync'), \
+ mock.patch.object(hil_remote.subprocess, 'run', lambda argv, **kw: calls.append((argv, kw)) or mock.Mock(returncode=0)):
+ for host in hosts:
+ hil_remote.rsync(('/tmp/t', 'tok'), '-q', local, f'{host}:/tmp/t/')
+ hil_remote.rsync(('/tmp/t', 'tok'), '-q', f'{host}:/tmp/t/hil_report.json', local)
+ name = 'hil_report.json' if os.name == 'nt' else str(local) # Windows rsync reads C: as a host
+ self.assertEqual([argv[-2:] for argv, _ in calls],
+ [x for h in hosts for x in ([name, f'{h}:/tmp/t/'], [f'{h}:/tmp/t/hil_report.json', name])])
+ self.assertEqual({kw['cwd'] for _, kw in calls}, {hil_remote.ROOT})
+
+ @unittest.skipUnless(os.name == 'nt', 'drive letters are Windows paths')
+ def test_a_config_rsync_cannot_name_is_refused_before_the_lease(self):
+ leased = []
+
+ @contextlib.contextmanager
+ def lease(*a):
+ leased.append(a)
+ yield '/tmp/t', 'tok'
+ with tempfile.TemporaryDirectory() as td:
+ root = Path(td)
+ other = 'Q:' if root.drive.upper() != 'Q:' else 'R:'
+ args = types.SimpleNamespace(build=False, build_dir='cmake-build', accumulate=False)
+ with mock.patch.object(hil_remote, 'ROOT', root), mock.patch.object(hil_remote, 'helper'), \
+ mock.patch.object(hil_remote, 'parse_run', return_value=(args, CONFIG, Path(f'{other}/rig/tinyusb.json'))), \
+ mock.patch.object(hil_remote, 'resolve_firmware', return_value=[root / 'cmake-build' / 'cmake-build-alpha']), \
+ mock.patch.object(hil_remote, 'remote_lease', lease), \
+ mock.patch.object(hil_remote.subprocess, 'run', side_effect=AssertionError('rsync or ssh ran')), \
+ contextlib.redirect_stdout(io.StringIO()), self.assertRaises(SystemExit) as e:
+ hil_remote.run([])
+ self.assertIn('is not on the drive of the checkout', str(e.exception.code))
+ self.assertEqual(leased, [], 'refused after the lease wiped the rig tree')
+
+ def test_the_pin_check_passes_where_the_lock_primitive_cannot_import_fcntl(self):
+ # agentrc.load() checks the pin before running agentrc's hil_lock, whose fcntl Windows lacks
+ args = types.SimpleNamespace(build=True, board=['alpha'], flasher=None, exclude_flasher=None,
+ build_dir='cmake-build')
+ for error, refusal in ((ModuleNotFoundError("No module named 'fcntl'", name='fcntl'), '--build would build'),
+ (ImportError('lib/agentrc must be at the pin'), 'lib/agentrc must be at the pin')):
+ with self.subTest(error), mock.patch.object(hil_remote, 'helper', side_effect=error), \
+ mock.patch.object(hil_remote, 'parse_run', return_value=(args, CONFIG, hil_remote.ROOT / 'x.json')), \
+ self.assertRaises(SystemExit) as e:
+ hil_remote.run([])
+ self.assertIn(refusal, str(e.exception.code))
if __name__ == '__main__':
diff --git a/.claude/test/test_release.py b/.claude/test/test_release.py
index d5d32f1..179bc12 100644
--- a/.claude/test/test_release.py
+++ b/.claude/test/test_release.py
@@ -3,13 +3,16 @@
refuses a previous tag that is not an ancestor and any first-parent commit
that is not a merged PR; gh failures surface instead of thinning the list."""
import importlib.util
+import io
import json
import os
import subprocess
import sys
import tempfile
+import types
import unittest
from pathlib import Path
+from unittest import mock
SCRIPT = Path(__file__).resolve().parents[1] / 'skills' / 'make-release' / 'scripts' / 'release.py'
spec = importlib.util.spec_from_file_location('release', SCRIPT)
@@ -86,6 +89,20 @@
self.assertEqual(text.count('"0.22.0": "0.22.0"'), 1)
self.assertIn('"0-latest": "0.22.0"', text)
+ def test_files_are_read_and_written_as_utf8_with_lf_whatever_the_locale(self):
+ # bump runs no git or gh, so the script can run as a process here, outside UTF-8 mode and,
+ # on POSIX, in an ASCII locale
+ with tempfile.TemporaryDirectory() as d:
+ root = Path(d)
+ fixture(root)
+ lib = LIBRARY_JSON.replace('"name": "TinyUSB",', '"name": "TinyUSB",\n "authors": "ø ✔",')
+ (root / 'library.json').write_bytes(lib.encode('utf-8'))
+ env = {**{k: v for k, v in os.environ.items() if k != 'PYTHONUTF8'}, 'LC_ALL': 'C', 'PYTHONCOERCECLOCALE': '0'}
+ r = subprocess.run([sys.executable, '-X', 'utf8=0', str(SCRIPT), '--root', d, 'bump', '0.22.1'],
+ capture_output=True, env=env, timeout=60)
+ self.assertEqual(r.returncode, 0, r.stderr)
+ self.assertEqual((root / 'library.json').read_bytes(), lib.replace('0.21.0', '0.22.1').encode('utf-8'))
+
class PrNumbersTest(unittest.TestCase):
def test_merge_button_and_squash_subjects_dedupe_and_sort(self):
@@ -99,34 +116,49 @@
class CliTest(unittest.TestCase):
- """git and gh are stubs on PATH: git answers the ancestor check and the log, gh answers per PR."""
+ """main() in-process with git and gh answered by a stand-in for subprocess.run, so no real
+ git or gh can run: git answers the ancestor check and the log, gh answers per PR. Their output
+ is UTF-8 bytes, decoded the way subprocess.run would with the caller's arguments under a
+ legacy code page's locale, and stdout and stderr are a console in a legacy code page."""
def run_cli(self, args, ancestor=True, gh_fail=()):
- with tempfile.TemporaryDirectory() as d:
- bin_dir = Path(d) / 'bin'
- bin_dir.mkdir()
- (bin_dir / 'git').write_text(f'''#!/bin/sh
-case "$*" in
- *merge-base*) exit {0 if ancestor else 1} ;;
- *log*) printf 'Merge pull request #10 from a/b\\nSquashed (#11)\\n' ;;
-esac
-''')
- (bin_dir / 'gh').write_text(f'''#!/bin/sh
-n=$3
-case " {' '.join(map(str, gh_fail))} " in *" $n "*) echo "boom $n" >&2; exit 1 ;; esac
-author=hathach; [ "$n" = 11 ] && author='dependabot[bot]'
-printf '{{"number": %s, "title": "PR %s", "labels": [{{"name": "Port"}}], "author": {{"login": "%s"}}}}' "$n" "$n" "$author"
-''')
- for f in (bin_dir / 'git', bin_dir / 'gh'):
- f.chmod(0o755)
- env = dict(os.environ, PATH=f'{bin_dir}:{os.environ["PATH"]}')
- return subprocess.run([sys.executable, str(SCRIPT), '--root', d, *args],
- capture_output=True, text=True, env=env)
+ def run(argv, capture_output=False, text=False, encoding=None, cwd=None):
+ out, err, rc = '', '', 0
+ if argv[0] == 'git':
+ rc = 0 if 'merge-base' not in argv or ancestor else 1
+ out = 'Merge pull request #10 from a/b\nSquashed (#11)\n' if 'log' in argv else ''
+ elif argv[0] == 'gh':
+ n = int(argv[3])
+ if n in gh_fail:
+ err, rc = f'boom {n}\n', 1
+ else:
+ out = json.dumps({'number': n, 'title': f'PR {n} \u2714', 'labels': [{'name': 'Port \U0001f6a8'}],
+ 'author': {'login': 'dependabot[bot]' if n == 11 else 'hathach'}}, ensure_ascii=False)
+ else:
+ raise AssertionError(f'unexpected command {argv}')
+ out, err = out.encode('utf-8'), err.encode('utf-8')
+ if encoding or text:
+ out, err = (s.decode(encoding or 'cp1252') for s in (out, err))
+ return subprocess.CompletedProcess(argv, rc, out, err)
+
+ # Windows prints UTF-8 whatever the console; elsewhere stdout keeps the encoding and errors it was given
+ console = [io.TextIOWrapper(io.BytesIO(), encoding='cp1252' if os.name == 'nt' else 'ascii',
+ errors='strict' if os.name == 'nt' else 'backslashreplace', newline='\n')
+ for _ in range(2)]
+ with tempfile.TemporaryDirectory() as d, mock.patch.object(release.subprocess, 'run', run), \
+ mock.patch.object(sys, 'argv', [str(SCRIPT), '--root', d, *args]), \
+ mock.patch.object(sys, 'stdout', console[0]), mock.patch.object(sys, 'stderr', console[1]):
+ code = release.main()
+ for c in console:
+ c.flush()
+ return types.SimpleNamespace(returncode=code, stdout=console[0].buffer.getvalue().decode('utf-8'),
+ stderr=console[1].buffer.getvalue().decode('utf-8'))
def test_prs_lists_number_title_and_labels(self):
r = self.run_cli(['prs', '--prev', '0.20.0'])
self.assertEqual(r.returncode, 0, r.stderr)
- self.assertEqual(r.stdout, '#10\tPR 10\t[Port]\n#11\tPR 11\t[Port]\n')
+ title, label = ('PR {} \u2714', 'Port \U0001f6a8') if os.name == 'nt' else ('PR {} \\u2714', 'Port \\U0001f6a8')
+ self.assertEqual(r.stdout, f'#10\t{title.format(10)}\t[{label}]\n#11\t{title.format(11)}\t[{label}]\n')
def test_contributors_drops_bots(self):
r = self.run_cli(['contributors', '--prev', '0.20.0'])
@@ -145,5 +177,36 @@
self.assertEqual(r.stdout, '')
+@unittest.skipIf(os.name == 'nt', 'the stubs are #! scripts Windows cannot exec: a miss would run the real git or gh')
+class ProcessTest(unittest.TestCase):
+ """The script as a process, git and gh stubs on PATH: main()'s status is the exit status."""
+
+ def run_script(self, args, ancestor=True):
+ with tempfile.TemporaryDirectory() as d:
+ bin_dir = Path(d) / 'bin'
+ bin_dir.mkdir()
+ (bin_dir / 'git').write_text(f'''#!/bin/sh
+case "$*" in
+ *merge-base*) exit {0 if ancestor else 1} ;;
+ *log*) printf 'Merge pull request #10 from a/b\\n' ;;
+esac
+''')
+ (bin_dir / 'gh').write_text('''#!/bin/sh
+printf '{"number": %s, "title": "PR %s", "labels": [], "author": {"login": "hathach"}}' "$3" "$3"
+''')
+ for f in (bin_dir / 'git', bin_dir / 'gh'):
+ f.chmod(0o755)
+ env = dict(os.environ, PATH=f'{bin_dir}{os.pathsep}{os.environ["PATH"]}')
+ return subprocess.run([sys.executable, str(SCRIPT), '--root', d, *args],
+ capture_output=True, text=True, env=env, timeout=60)
+
+ def test_the_exit_status_is_mains(self):
+ ok = self.run_script(['prs', '--prev', '0.20.0'])
+ self.assertEqual((ok.returncode, ok.stdout), (0, '#10\tPR 10\t[]\n'), ok.stderr)
+ refused = self.run_script(['prs', '--prev', '0.20.0'], ancestor=False)
+ self.assertEqual(refused.returncode, 1)
+ self.assertIn('not an ancestor of HEAD', refused.stderr)
+
+
if __name__ == '__main__':
unittest.main()
diff --git a/.claude/test/test_update_sponsor.py b/.claude/test/test_update_sponsor.py
new file mode 100644
index 0000000..af3e694
--- /dev/null
+++ b/.claude/test/test_update_sponsor.py
@@ -0,0 +1,75 @@
+"""Tests for the update-sponsor skill's update_sponsor.py, with gh and codespell answered
+in-process (no real CLI runs): gh's JSON is read as UTF-8 whatever the locale, and the codespell
+guard reads the words codespell reports from a probe file it could open."""
+import importlib.util
+import subprocess
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+SCRIPT = Path(__file__).resolve().parents[1] / 'skills' / 'update-sponsor' / 'update_sponsor.py'
+spec = importlib.util.spec_from_file_location('update_sponsor', SCRIPT)
+update_sponsor = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(update_sponsor)
+
+
+def completed(argv, out, kw, rc=0):
+ """What subprocess.run returns for UTF-8 output `out`, decoded as the caller asked under a
+ legacy code page's locale: bytes when it asked for no text."""
+ out = out.encode('utf-8')
+ if kw.get('encoding') or kw.get('text'):
+ out = out.decode(kw.get('encoding') or 'cp1252', kw.get('errors') or 'strict')
+ return subprocess.CompletedProcess(argv, rc, out, out[:0])
+
+
+class GhTest(unittest.TestCase):
+ def test_label_names_come_back_as_github_sent_them(self):
+ def run(argv, **kw):
+ self.assertEqual(argv[0], 'gh')
+ return completed(argv, f'{update_sponsor.L_PRIO_TOP}\n', kw)
+ with mock.patch.object(update_sponsor.subprocess, 'run', run):
+ self.assertEqual(update_sponsor.gh('label', 'list'), f'{update_sponsor.L_PRIO_TOP}\n')
+
+
+TYPO = 'te' + 'h' # spelled out so the repository's own codespell hook leaves it alone
+
+
+class CodespellTest(unittest.TestCase):
+ def test_the_words_codespell_flags_in_the_probe(self):
+ seen = []
+
+ def run(argv, **kw):
+ self.assertEqual(argv[0], 'codespell')
+ probe = argv[-1]
+ seen.append(probe)
+ # codespell opens the probe itself: it must be closed and still there
+ self.assertIn(f'+{TYPO} \u00f8sterberg\n', Path(probe).read_text(encoding='utf-8'))
+ return completed(argv, f'{probe}:1: {TYPO} ==> the\n', kw, rc=65)
+ with mock.patch.object(update_sponsor.subprocess, 'run', run):
+ self.assertEqual(update_sponsor.codespell_collisions('a\n', f'{TYPO} \u00f8sterberg\n'), [TYPO])
+ self.assertFalse(Path(seen[0]).exists(), 'the probe is removed afterwards')
+
+ def test_a_probe_that_could_not_be_written_is_removed(self):
+ made, real = [], tempfile.NamedTemporaryFile
+
+ def full(*a, **kw):
+ f = real(*a, **kw)
+ made.append(f.name)
+ f.write = mock.Mock(side_effect=OSError(28, 'No space left on device'))
+ return f
+ with mock.patch.object(update_sponsor.tempfile, 'NamedTemporaryFile', full), self.assertRaises(OSError):
+ update_sponsor.codespell_collisions('a\n', 'b\n')
+ self.assertFalse(Path(made[0]).exists())
+
+ def test_no_codespell_is_a_note_not_a_failure(self):
+ def run(argv, **kw):
+ raise FileNotFoundError(argv[0])
+ with mock.patch.object(update_sponsor.subprocess, 'run', run), \
+ mock.patch('builtins.print') as note:
+ self.assertEqual(update_sponsor.codespell_collisions('a\n', 'b\n'), [])
+ self.assertIn('UNCHECKED', note.call_args[0][0])
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/.claude/test/test_usbtest_host.py b/.claude/test/test_usbtest_host.py
new file mode 100644
index 0000000..eda1fff
--- /dev/null
+++ b/.claude/test/test_usbtest_host.py
@@ -0,0 +1,47 @@
+"""run_case.py and wedge_drill.py drive usbfs, /proc and the usbtest kernel module: on a host
+that is not Linux they parse their arguments (so --help works) and then refuse through their
+verdict line, exit 2 (refused before any hardware action), before importing the harness."""
+import json
+import subprocess
+import sys
+import unittest
+from pathlib import Path
+
+SCRIPTS = Path(__file__).resolve().parents[1] / 'skills' / 'usbtest' / 'scripts'
+ARGS = {'run_case.py': ['--config', 'rig.json', '--board', 'x', '--tests', '1', '--after', 'leave'],
+ 'wedge_drill.py': ['--config', 'rig.json', '--board', 'x']}
+
+
+@unittest.skipIf(sys.platform == 'linux', 'the refusal is for hosts that are not Linux')
+class NotLinux(unittest.TestCase):
+ def run_script(self, name, *args):
+ return subprocess.run([sys.executable, str(SCRIPTS / name), *args],
+ capture_output=True, encoding='utf-8', timeout=60)
+
+ def test_each_script_refuses_with_its_verdict_line(self):
+ for name, args in ARGS.items():
+ with self.subTest(name):
+ r = self.run_script(name, *args)
+ error = f'{name} needs a Linux host (usbfs, /proc, the usbtest kernel module)'
+ self.assertEqual(r.returncode, 2, r.stderr)
+ self.assertEqual(r.stderr, f'error: {error}\n')
+ verdict = json.loads(r.stdout.splitlines()[-1])
+ self.assertEqual((verdict['board'], verdict['boardState'], verdict['error']), ('x', 'untouched', error))
+
+ def test_help_still_works(self):
+ for name in ARGS:
+ with self.subTest(name):
+ r = self.run_script(name, '--help')
+ self.assertEqual(r.returncode, 0, r.stderr)
+ self.assertIn('--board', r.stdout)
+
+ def test_a_bad_command_line_is_argparse_s_error(self):
+ for name in ARGS:
+ with self.subTest(name):
+ r = self.run_script(name, '--board', 'x')
+ self.assertEqual((r.returncode, r.stdout), (2, ''))
+ self.assertIn('the following arguments are required: --config', r.stderr)
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/.claude/test/test_usbtest_kernel_src.py b/.claude/test/test_usbtest_kernel_src.py
index d1cbaa6..17aed7a 100644
--- a/.claude/test/test_usbtest_kernel_src.py
+++ b/.claude/test/test_usbtest_kernel_src.py
@@ -177,10 +177,10 @@
self.server = Server(src)
self.addCleanup(self.server.close)
- def run_script(self, *args):
- env = dict(os.environ, KERNEL_SRC_BASE_URL=self.server.url, PYTHONDONTWRITEBYTECODE='1')
+ def run_script(self, *args, **env):
+ env = dict(os.environ, KERNEL_SRC_BASE_URL=self.server.url, PYTHONDONTWRITEBYTECODE='1', **env)
return subprocess.run([sys.executable, str(SCRIPT), '--cache', str(self.tmp / 'cache'), *args],
- capture_output=True, text=True, env=env, timeout=60)
+ capture_output=True, encoding='utf-8', env=env, timeout=60)
def test_fetch_prints_case_and_waits(self):
r = self.run_script('--release', '6.12.107+deb13-amd64', '--case', '16')
@@ -198,6 +198,25 @@
self.assertEqual(sorted(self.server.hits),
['/drivers/usb/misc/usbtest.c?h=v6.12.107', '/tools/usb/testusb.c?h=v6.12.107'])
+ def test_utf8_source_prints_as_utf8_whatever_the_locale(self):
+ # a comment in the case block that no legacy code page decodes or encodes
+ text = USBTEST_C.replace('\tcase 16:\n', '\tcase 16:\t/* \u2192 \u2714 \u00f8 */\n')
+ (self.src / 'drivers/usb/misc/usbtest.c').write_text(text, encoding='utf-8')
+ env = {'PYTHONUTF8': '0', 'LC_ALL': 'C', 'PYTHONCOERCECLOCALE': '0'} # an ASCII locale on POSIX
+ if os.name != 'nt': # stdout stays as given there, so pin it; Windows reconfigures its own
+ env['PYTHONIOENCODING'] = 'utf-8'
+ r = self.run_script('--tag', 'v6.12.107', '--case', '16', **env)
+ self.assertEqual(r.returncode, 0, r.stderr)
+ self.assertIn('\tcase 16:\t/* \u2192 \u2714 \u00f8 */', r.stdout)
+
+ @unittest.skipIf(os.name == 'nt', 'Windows prints UTF-8 whatever the console; elsewhere stdout stays as given')
+ def test_posix_stdout_keeps_the_encoding_and_errors_it_was_given(self):
+ text = USBTEST_C.replace('\tcase 16:\n', '\tcase 16:\t/* \u2714 */\n')
+ (self.src / 'drivers/usb/misc/usbtest.c').write_text(text, encoding='utf-8')
+ r = self.run_script('--tag', 'v6.12.107', '--case', '16', PYTHONIOENCODING='ascii:backslashreplace')
+ self.assertEqual(r.returncode, 0, r.stderr)
+ self.assertIn('\tcase 16:\t/* \\u2714 */', r.stdout)
+
def test_cache_is_reused(self):
self.assertEqual(self.run_script('--tag', 'v6.12.107').returncode, 0)
self.assertEqual(self.run_script('--tag', 'v6.12.107').returncode, 0)
diff --git a/.claude/test/test_usbtest_run_case.py b/.claude/test/test_usbtest_run_case.py
index 24cf9f0..7d790c7 100644
--- a/.claude/test/test_usbtest_run_case.py
+++ b/.claude/test/test_usbtest_run_case.py
@@ -15,6 +15,8 @@
from pathlib import Path
from unittest import mock
+if os.name == 'nt':
+ raise unittest.SkipTest('run_case.py imports test/hil/usbtest.py, which needs fcntl')
REPO = Path(__file__).resolve().parents[2]
SCRIPT = REPO / '.claude' / 'skills' / 'usbtest' / 'scripts' / 'run_case.py'
spec = importlib.util.spec_from_file_location('run_case', SCRIPT)
diff --git a/.claude/test/test_usbtest_wedge_drill.py b/.claude/test/test_usbtest_wedge_drill.py
index eaeba9a..b4f843f 100644
--- a/.claude/test/test_usbtest_wedge_drill.py
+++ b/.claude/test/test_usbtest_wedge_drill.py
@@ -16,6 +16,8 @@
from pathlib import Path
from unittest import mock
+if os.name == 'nt':
+ raise unittest.SkipTest('wedge_drill.py imports test/hil/usbtest.py, which needs fcntl')
REPO = Path(__file__).resolve().parents[2]
SCRIPTS = REPO / '.claude' / 'skills' / 'usbtest' / 'scripts'
sys.path.insert(0, str(SCRIPTS))
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index bdf1ec8..58ba920 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -115,7 +115,7 @@
# relying on it being installed system-wide.
- id: claude-scripts
name: claude-scripts
- files: ^(\.claude/(skills/.*/(scripts/|SKILL\.md)|test/)|tools/get_deps\.py$)
+ files: ^(\.claude/(skills/.*/(scripts/|SKILL\.md|[^/]*\.py$)|test/)|tools/(build_utils|family_json|get_deps)\.py$)
entry: python3 -m unittest discover -s .claude/test -p 'test_*.py'
pass_filenames: false
language: python
diff --git a/test/hil/test/test_family_json.py b/test/hil/test/test_family_json.py
index 259f655..2aae7af 100644
--- a/test/hil/test/test_family_json.py
+++ b/test/hil/test/test_family_json.py
@@ -3,12 +3,15 @@
No cmake here: the build dirs are fixtures. The last test is the check itself over the
real tree, which is what the pre-commit hook guarantees."""
import contextlib
+import errno
import io
import json
import os
import pathlib
+import subprocess
import sys
import tempfile
+import types
import unittest
from unittest import mock
@@ -115,6 +118,75 @@
self.assertEqual(fj.defines_of(fj.preprocess_argv(args)), {'A': '2'})
+class SplitCommand(unittest.TestCase):
+ """CMake on Windows writes `command` for CreateProcess, which hands it to the compiler's C
+ runtime to split: backslashes are path separators, only quotes are special."""
+ split = staticmethod(fj.split_command)
+
+ def windows(self, cmd):
+ with mock.patch.object(fj.os, 'name', 'nt'):
+ return self.split(cmd)
+
+ def test_the_c_runtime_rules(self):
+ for args in (['C:\\Tool\\bin\\arm-none-eabi-gcc.exe', '-IE:/mcu/tinyusb/src', '-c', 'E:\\mcu\\src\\tusb.c'],
+ ['C:\\Program Files\\gcc.exe', '-DX=a b', ''],
+ ['-DCFG_TUD_MEM_SECTION=__attribute__((section("m_usb_global")))'],
+ ['a\\\\b', 'trailing\\', 'quoted trailing\\', 'q\\"', '\\\\"x"', '\t']):
+ self.assertEqual(self.windows(subprocess.list2cmdline(args)), args)
+ # as lpc55's configure writes it, and the doubled quote inside quotes
+ self.assertEqual(self.windows('gcc -DS=__attribute__((section(\\"m_usb_global\\"))) "a""b"'),
+ ['gcc', '-DS=__attribute__((section("m_usb_global")))', 'a"b'])
+
+ def test_posix_commands_stay_shell_words(self):
+ with mock.patch.object(fj.os, 'name', 'posix'):
+ self.assertEqual(fj.split_command("gcc '-DA=1 2' -DB=\\\"x\\\""), ['gcc', '-DA=1 2', '-DB="x"'])
+
+ def test_a_windows_command_preprocesses_with_its_own_compiler_and_source(self):
+ with tempfile.TemporaryDirectory() as d:
+ src = str(pathlib.Path(d, 'src', 'tusb.c'))
+ e = {'directory': d, 'file': src, 'output': 'x.obj',
+ 'command': subprocess.list2cmdline(['C:\\Tool\\bin\\gcc.exe', '-DCFG_TUSB_MCU=OPT_MCU_STM32F4',
+ '-o', 'dir\\x.obj', '-c', src])}
+ with mock.patch.object(fj, 'split_command', self.windows): # pathlib stays native
+ argv = fj.preprocess_argv(e)
+ self.assertEqual(argv, ['C:\\Tool\\bin\\gcc.exe', '-DCFG_TUSB_MCU=OPT_MCU_STM32F4'])
+
+
+MSVCRT = types.SimpleNamespace(LK_UNLCK=0, LK_LOCK=1, LK_NBLCK=2)
+
+
+class WindowsLock(unittest.TestCase):
+ """_locked's msvcrt path on any host: no fcntl to import, and an msvcrt whose locking() is the test's."""
+ def rewrite(self, locking):
+ msvcrt = types.SimpleNamespace(**vars(MSVCRT), locking=locking)
+ with tempfile.TemporaryDirectory() as d:
+ root = pathlib.Path(d)
+ (root / 'family.json').write_text('{}\n', encoding='utf-8')
+ with mock.patch.object(fj, 'ROOT', root), mock.patch.dict(sys.modules, {'fcntl': None, 'msvcrt': msvcrt}), \
+ mock.patch.object(fj.time, 'sleep'):
+ return fj.rewrite(lambda data: data.update(x={}), root / 'family.json')
+
+ def test_a_writer_waits_for_the_lock_as_long_as_it_is_held(self):
+ calls = []
+
+ def locking(fd, mode, n):
+ calls.append(mode)
+ if mode == MSVCRT.LK_LOCK: # it gives up after about 10 s, and the retry loop would spin on that
+ raise AssertionError('LK_LOCK')
+ if mode == MSVCRT.LK_UNLCK:
+ os.fstat(fd) # EBADF once the file is closed
+ elif calls.count(mode) < 30:
+ raise OSError(errno.EACCES, 'Permission denied')
+ self.assertEqual(self.rewrite(locking), 'updated')
+ self.assertEqual(calls, [MSVCRT.LK_NBLCK] * 30 + [MSVCRT.LK_UNLCK], 'released before the close')
+
+ def test_a_lock_error_that_is_not_contention_is_raised(self):
+ def locking(fd, mode, n):
+ raise OSError(errno.EBADF, 'Bad file descriptor')
+ with self.assertRaises(OSError):
+ self.rewrite(locking)
+
+
class Observe(unittest.TestCase):
def test_a_configured_dir_yields_the_row(self):
with tempfile.TemporaryDirectory() as d:
@@ -301,6 +373,18 @@
path.write_text(fj.canonical_text(data))
return path
+ @unittest.skipIf(os.name == 'nt', 'a backslash separates there, so no one file has this name')
+ def test_a_backslash_portable_entry_is_refused_on_windows_even_when_posix_finds_the_file(self):
+ with tempfile.TemporaryDirectory() as d:
+ root = fixture_root(d)
+ p = 'synopsys\\dwc2\\dcd_dwc2.c'
+ (root / 'src' / 'portable' / p).write_text('')
+ row = dict(ROW, portable=[p])
+ self.assertEqual(fj.row_violations(row, {'OPT_MCU_STM32F4'}, root), [])
+ with mock.patch.object(fj, 'os', types.SimpleNamespace(name='nt')): # row_violations' os, as on Windows
+ self.assertEqual(fj.row_violations(row, {'OPT_MCU_STM32F4'}, root),
+ [f'portable entry {p!r} is not a relative path to a file under src/portable/'])
+
def test_a_complete_catalog_passes_and_every_gap_is_named_with_its_remedy(self):
with tempfile.TemporaryDirectory() as d:
root = fixture_root(d)
@@ -328,6 +412,7 @@
({**good, 'stm32f4': {'stm32f407disco': {'cmake': dict(ROW, roles=[{}])}}}, 'roles must be a sorted list'),
({**good, 'stm32f4': {'stm32f407disco': {'cmake': dict(ROW, portable=[str(root / 'src' / 'portable' / 'synopsys/dwc2/dcd_dwc2.c')])}}}, 'is not a relative path'),
({**good, 'stm32f4': {'stm32f407disco': {'cmake': dict(ROW, portable=['../portable/synopsys/dwc2/dcd_dwc2.c'])}}}, 'is not a relative path'),
+ ({**good, 'stm32f4': {'stm32f407disco': {'cmake': dict(ROW, portable=['synopsys\\dwc2\\dcd_dwc2.c'])}}}, 'is not a relative path'),
({**good, 'stm32f4': {'stm32f407disco': {'cmake': dict(ROW, mcu='OPT_MCU_BOGUS')}}}, 'is not an OPT_MCU_* defined in src/tusb_option.h'),
({**good, 'stm32f4': {'stm32f407disco': {'cmake': ROW, 'make': 1}}}, 'is {"cmake": ...} and nothing else'),
({**good, 'nrf': {'x': {'cmake': ROW}}}, 'nrf: rows for a family dir that does not exist'),
diff --git a/test/hil/test/test_hil_remote.py b/test/hil/test/test_hil_remote.py
index cf0fcaf..c1c8ad8 100644
--- a/test/hil/test/test_hil_remote.py
+++ b/test/hil/test/test_hil_remote.py
@@ -116,11 +116,61 @@
self.assertEqual(self.wait('r6', '--timeout', '20'), (0, {'state': 'done', 'runId': 'r6', 'exit': 0, 'reports': ['hil_report.md']}))
def test_a_wrapper_killed_before_its_done_record_is_dead(self):
- self.launch('os.kill(os.getpid(), signal.SIGKILL)', 'r7').wait()
+ # Windows has no SIGKILL; its SIGTERM is TerminateProcess, as abrupt
+ self.launch('os.kill(os.getpid(), getattr(signal, "SIGKILL", signal.SIGTERM))', 'r7').wait()
rc, status = self.wait('r7', '--timeout', '5')
self.assertEqual((rc, status['state']), (hil_remote.DEAD, 'dead'))
self.assertFalse(hil_remote.run_paths('r7')[1].exists())
+ @unittest.skipUnless(os.name == 'nt', 'msvcrt has no shared lock: two waiters probes contend there')
+ def test_another_waiters_probe_does_not_read_as_a_live_run(self):
+ self.started('r12')
+ started = hil_remote.run_paths('r12')[0]
+ fd = os.open(started, os.O_RDONLY)
+ self.addCleanup(os.close, fd)
+ hil_remote.win_lock(fd, block=False) # the other waiter, mid-probe
+ other = threading.Timer(0.05, hil_remote.win_unlock, (fd,))
+ other.start()
+ self.addCleanup(other.join) # before the close: it seeks and unlocks fd
+ # a 5 s retry budget at the usual step, so a late Timer on a loaded host still lands inside it
+ with mock.patch.object(hil_remote, 'PROBE_TRIES', int(5 / hil_remote.PROBE_SECS)), \
+ mock.patch.object(hil_remote, 'win_unlock', wraps=hil_remote.win_unlock) as unlock:
+ self.assertFalse(hil_remote.run_alive(started))
+ unlock.assert_called_once() # the probe released its hold, not left it to the close
+
+ def test_a_refused_unlink_of_the_hidden_name_keeps_the_runs_exit(self):
+ # Windows refuses it while a waiter's probe holds the record open; the next sweep takes it
+ self.stub_run(lambda argv, receipt=None: (3, []))
+ real_unlink = Path.unlink
+
+ def unlink(path, missing_ok=False):
+ if path.name.startswith('.r14.started.json.'):
+ raise PermissionError(13, 'The process cannot access the file')
+ return real_unlink(path, missing_ok=missing_ok)
+ with mock.patch.object(hil_remote, 'fcntl', None), mock.patch.object(hil_remote, 'win_lock'), \
+ mock.patch.object(hil_remote, 'win_unlock'), mock.patch.object(Path, 'unlink', unlink):
+ self.assertEqual(hil_remote.main(['--run-id', 'r14']), 3)
+ self.assertEqual(self.wait('r14')[1], {'state': 'done', 'runId': 'r14', 'exit': 3, 'reports': []})
+
+ @unittest.skipUnless(os.name == 'nt', 'only Windows defers the unlink of the hidden name')
+ def test_a_killed_runs_hidden_record_name_is_swept_at_the_next_run(self):
+ runs = self.root / hil_remote.RUNS_DIR
+ runs.mkdir()
+ hidden = {}
+ for run_id, pid in (('killed', 4242), ('live', 4343)):
+ started = runs / f'{run_id}.started.json'
+ started.write_text('{}')
+ hidden[run_id] = runs / f'.{started.name}.{pid}.tmp'
+ os.link(started, hidden[run_id])
+ fd = os.open(hidden['live'], os.O_RDONLY) # its run still holds it
+ self.addCleanup(os.close, fd)
+ writing = runs / '.new.started.json.4444.tmp' # a launch that has not linked its record yet
+ writing.write_text('{}')
+ self.stub_run(lambda argv, receipt=None: (0, []))
+ self.assertEqual(hil_remote.main(['--run-id', 'r13']), 0)
+ self.assertEqual(sorted(p.name for p in runs.glob('.*.tmp')), sorted([hidden['live'].name, writing.name]))
+ self.assertTrue((runs / 'killed.started.json').is_file(), 'the record itself stays')
+
def test_a_wait_begun_before_the_launch_wrote_its_record_waits_for_it(self):
self.stub_run(lambda argv, receipt=None: (0, []))
timer = threading.Timer(0.1, hil_remote.main, [['--run-id', 'r8']])
@@ -136,6 +186,22 @@
self.assertIn('no run never was started', e.exception.code)
self.assertLess(time.monotonic() - begun, 1, 'neither the grace nor its poll outlasts --timeout')
+ def test_a_killed_runs_hidden_name_under_a_reused_pid_keeps_its_record(self):
+ self.started('r13')
+ started = hil_remote.run_paths('r13')[0]
+ os.link(started, started.with_name(f'.{started.name}.{os.getpid()}.tmp')) # left by a kill on Windows
+ before = started.read_bytes()
+ with self.assertRaises(FileExistsError):
+ hil_remote.publish_locked(started, {'runId': 'r13', 'pid': os.getpid(), 'startedAt': 1})
+ self.assertEqual(started.read_bytes(), before)
+
+ def test_an_unlinked_leftover_hidden_name_is_replaced(self):
+ started = hil_remote.run_paths('r14')[0]
+ started.parent.mkdir(exist_ok=True)
+ started.with_name(f'.{started.name}.{os.getpid()}.tmp').write_text('stale')
+ os.close(hil_remote.publish_locked(started, {'runId': 'r14', 'pid': os.getpid(), 'startedAt': 1}))
+ self.assertEqual(json.loads(started.read_text())['runId'], 'r14')
+
def test_a_timeout_past_the_tool_cap_is_refused(self):
for bad in ('571', '-1', 'nan', 'inf'):
with self.subTest(timeout=bad), contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit) as e:
@@ -338,8 +404,8 @@
def rsync(guard, *args, optional=False):
if '--prune-empty-dirs' not in args:
return 0
- for src in (Path(a) for a in args if not a.startswith('-') and ':' not in a):
- sent.update({str(f.relative_to(src.parent.parent)): hil_remote.digest(f) for f in src.rglob('*')
+ for src in (a for a in args if isinstance(a, Path)): # local files are Paths
+ sent.update({f.relative_to(src.parent.parent).as_posix(): hil_remote.digest(f) for f in src.rglob('*')
if f.suffix in hil_remote.STAGED_SUFFIXES or f.name in hil_remote.STAGED_NAMES})
return 1 # stop the run once the firmware transfer is seen
with mock.patch.object(hil_remote, 'remote_lease', lease), mock.patch.object(hil_remote, 'rsync', rsync), \
@@ -363,7 +429,7 @@
def rsync(guard, *args, optional=False):
if '--prune-empty-dirs' in args:
- state['sources'] += [Path(a) for a in args if not a.startswith('-') and ':' not in a]
+ state['sources'] += [a for a in args if isinstance(a, Path)]
return 0
def run_command(*a, **k):
diff --git a/tools/build_utils.py b/tools/build_utils.py
index 7124d95..fa814d4 100755
--- a/tools/build_utils.py
+++ b/tools/build_utils.py
@@ -561,9 +561,9 @@
if not hits:
raise BoardInfoError(f'unknown board {board}')
if len(hits) > 1:
- raise BoardInfoError(f'{board} exists in several families: ' + ', '.join(map(str, hits)))
+ raise BoardInfoError(f'{board} exists in several families: ' + ', '.join(h.as_posix() for h in hits))
board_dir = hits[0]
- return board_dir, (str(board_dir / 'board.cmake'), str(board_dir.parent.parent / 'family.cmake'))
+ return board_dir, ((board_dir / 'board.cmake').as_posix(), (board_dir.parent.parent / 'family.cmake').as_posix())
def _checked_device(value, source):
@@ -610,8 +610,8 @@
hits = sorted((board_dir / 'ozone').glob('*.jdebug'))
if len(hits) > 1:
raise BoardInfoError(f'{board}: several reference projects: '
- + ', '.join(str(h) for h in hits))
- return str(hits[0]) if hits else None
+ + ', '.join(h.as_posix() for h in hits))
+ return hits[0].as_posix() if hits else None
def main(argv=None):
diff --git a/tools/family_json.py b/tools/family_json.py
index 13a056a..f400bde 100644
--- a/tools/family_json.py
+++ b/tools/family_json.py
@@ -25,6 +25,8 @@
tools/build.py get_examples) is not detected.
"""
import argparse
+import contextlib
+import errno
import json
import os
import re
@@ -32,6 +34,7 @@
import shutil
import subprocess
import sys
+import time
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
@@ -66,9 +69,9 @@
try:
data = json.loads(text)
except ValueError as e:
- raise Failure(f'{path.relative_to(ROOT)} is not valid JSON ({e}); fix it by hand')
+ raise Failure(f'{path.relative_to(ROOT).as_posix()} is not valid JSON ({e}); fix it by hand')
if not isinstance(data, dict):
- raise Failure(f'{path.relative_to(ROOT)} is not a JSON object')
+ raise Failure(f'{path.relative_to(ROOT).as_posix()} is not a JSON object')
for family, rows in data.items():
if not isinstance(rows, dict):
kind = 'null' if rows is None else f'a {type(rows).__name__}'
@@ -93,9 +96,54 @@
return names
+def split_command(cmd):
+ """A compile database `command` as its compiler's argv: shell words, or on Windows, where
+ the command line reaches CreateProcess as it is, the argv rules of the Microsoft C runtime
+ (msvcr90 and later, the UCRT): backslashes are literal except before a quote, 2k of them
+ before a quote are k and the quote delimits, 2k+1 are k and a literal quote, and `""`
+ inside quotes is a literal quote. CMake writes an embedded quote as `\\"`. The program name
+ is split by these rules too, though the CRT reads it with quotes only toggling; the two
+ differ only on a backslash before a quote or a `""`, which no compiler path holds.
+ The dialect follows the Python reading the file (os.name), which suits native and MinGW
+ Pythons on Windows; a Cygwin or MSYS-runtime Python (os.name 'posix') reading a Windows
+ build's file is not supported."""
+ if os.name != 'nt':
+ return shlex.split(cmd)
+ out, cur, quoted, i, n = [], None, False, 0, len(cmd)
+ while i < n:
+ c = cmd[i]
+ if c == '\\':
+ j = i
+ while j < n and cmd[j] == '\\':
+ j += 1
+ if j < n and cmd[j] == '"':
+ # 2k backslashes before a quote are k and the quote delimits; 2k+1 escape it
+ cur = (cur or '') + '\\' * ((j - i) // 2) + ('"' if (j - i) % 2 else '')
+ i = j + (j - i) % 2
+ else:
+ cur, i = (cur or '') + cmd[i:j], j
+ elif c == '"':
+ if quoted and cmd[i + 1:i + 2] == '"': # "" inside quotes is a literal quote
+ cur, i = cur + '"', i + 2
+ else:
+ quoted, cur, i = not quoted, cur or '', i + 1
+ elif c in ' \t' and not quoted:
+ if cur is not None:
+ out.append(cur)
+ cur = None
+ i += 1
+ else:
+ cur, i = (cur or '') + c, i + 1
+ if cur is not None:
+ out.append(cur)
+ return out
+
+
def compile_tokens(entry):
- """One compile command's tokens, response files expanded."""
- tokens = list(entry['arguments']) if 'arguments' in entry else shlex.split(entry['command'])
+ """One compile command's tokens, response files expanded. gcc reads a response file with
+ libiberty's buildargv on every host; shlex is an approximation of that, close enough for
+ the files CMake writes, which name paths with forward slashes on Windows too."""
+ tokens = list(entry['arguments']) if 'arguments' in entry else split_command(entry['command'])
out = []
for t in tokens:
if t.startswith('@') and len(t) > 1:
@@ -277,8 +325,8 @@
each other's change. Returns 'updated' or 'unchanged'."""
lock = ROOT / 'cmake-build' / '.family.json.lock'
lock.parent.mkdir(exist_ok=True)
- with open(lock, 'w') as lf:
- _lock(lf)
+ # not 'w': a second opener must not truncate the holder's file
+ with open(lock, 'a') as lf, _locked(lf):
data = load(path)
edit(data)
text = canonical_text(data)
@@ -294,13 +342,33 @@
return 'updated'
-def _lock(f):
+@contextlib.contextmanager
+def _locked(f):
+ """Hold the empty lock file f exclusively."""
try:
import fcntl
- fcntl.flock(f, fcntl.LOCK_EX)
- except ImportError:
+ except ImportError: # Windows
+ fcntl = None
import msvcrt
- msvcrt.locking(f.fileno(), msvcrt.LK_LOCK, 1)
+ if fcntl:
+ fcntl.flock(f, fcntl.LOCK_EX)
+ yield # the close releases it
+ return
+ fd = f.fileno()
+ os.lseek(fd, 0, os.SEEK_SET) # every holder locks the first byte
+ while True:
+ try:
+ msvcrt.locking(fd, msvcrt.LK_NBLCK, 1)
+ break
+ except OSError as e:
+ if e.errno not in (errno.EACCES, errno.EDEADLOCK):
+ raise
+ time.sleep(0.05) # LK_LOCK gives up after about 10 s
+ try:
+ yield
+ finally: # Windows drops a closed handle's lock only "after a delay"
+ os.lseek(fd, 0, os.SEEK_SET)
+ msvcrt.locking(fd, msvcrt.LK_UNLCK, 1)
def join(build_dirs):
@@ -360,9 +428,9 @@
data = load(path)
text = path.read_text(encoding='utf-8')
except (Failure, OSError) as e:
- return [f'{path.relative_to(root)}: {e}']
+ return [f'{path.relative_to(root).as_posix()}: {e}']
if text != canonical_text(data):
- out.append(f'{path.relative_to(root)} is not in canonical form')
+ out.append(f'{path.relative_to(root).as_posix()} is not in canonical form')
tree = inventory(root)
option_h = (root / 'src' / 'tusb_option.h').read_text(encoding='utf-8')
mcus = set(re.findall(r'^#define (OPT_MCU_\w+)', option_h, re.M))
@@ -545,6 +613,7 @@
for p in row['portable']:
# a plain relative path: joining an absolute one would discard the base
if not isinstance(p, str) or p.startswith('/') or '..' in p.split('/') \
+ or (os.name == 'nt' and ('\\' in p or Path(p).anchor)) \
or not (root / 'src' / 'portable' / p).is_file():
out.append(f'portable entry {p!r} is not a relative path to a file under src/portable/')
if isinstance(row['roles'], list) and all(isinstance(x, str) for x in row['roles']):
@@ -579,7 +648,7 @@
for line in problems:
print(line)
if changed:
- print(f'{CATALOG.relative_to(ROOT)} updated: stage it and commit again')
+ print(f'{CATALOG.relative_to(ROOT).as_posix()} updated: stage it and commit again')
return 1 if changed or problems else 0
if a.cmd == 'refresh':
asked, failed, problems = refresh(CATALOG, ROOT, subprocess.run)
diff --git a/tools/get_deps.py b/tools/get_deps.py
index 49e141d..3abace0 100755
--- a/tools/get_deps.py
+++ b/tools/get_deps.py
@@ -436,7 +436,7 @@
# argparse error here reds the Get Dependencies step of every scoped PR
parser.add_argument('-e', '--example', action='append', default=[], help='Have no effect')
parser.add_argument('--gen-doc', nargs='?', const=DEPS_RST, metavar='RST',
- help=f'write the dependency table for the docs ({DEPS_RST.relative_to(TOP)}) and exit, fetching nothing')
+ help=f'write the dependency table for the docs ({DEPS_RST.relative_to(TOP).as_posix()}) and exit, fetching nothing')
args = parser.parse_args()
if args.gen_doc:
