summaryrefslogtreecommitdiff
path: root/test/hil/hil_select.py
diff options
context:
space:
mode:
authorHa Thach <[email protected]>2026-07-30 02:29:32 +0700
committerGitHub <[email protected]>2026-07-30 02:29:32 +0700
commiteef5af86aa26fe3d72e41156a586a6ed3ffce9f8 (patch)
treefa5ebe3a8d46738338eaee2d4a80717a7a10f464 /test/hil/hil_select.py
parent5d8afbb23248e9d1d5d1fdf434c9c58cc54ba5b7 (diff)
hil, ci: scope HIL builds and tests to the boards a PR affects (#3797)
hil, ci: scope HIL builds and tests to the boards a PR affects Add test/hil/hil_select.py, a stdlib-only selector that maps a PR diff to the rig boards, tests and BSP families a change can affect, and wire it into CI so pull requests build and run only those. A port change picks its families' boards, a class change picks the examples enabling that class, and device/host changes prune the other role. Anything unclassified — infra, an unmapped port, a selector error — falls back to the full matrix, and push/schedule runs are untouched. Move the shared example lists to hil_examples.py; 54 hardware-free tests cover the rules.
Diffstat (limited to 'test/hil/hil_select.py')
-rwxr-xr-xtest/hil/hil_select.py520
1 files changed, 520 insertions, 0 deletions
diff --git a/test/hil/hil_select.py b/test/hil/hil_select.py
new file mode 100755
index 000000000..3ac3f1fdb
--- /dev/null
+++ b/test/hil/hil_select.py
@@ -0,0 +1,520 @@
+#!/usr/bin/env python3
+# SPDX-License-Identifier: MIT
+"""PR-diff -> HIL selection: which rig boards and which tests a change can affect.
+
+Stdlib-only (runs on bare CI runners; never imports hil_test/hil_flash/hil_lock).
+Fail-open: any file no rule classifies forces the full matrix. See
+docs/superpowers/specs/2026-07-29-hil-pr-scoped-selection-design.md.
+
+JSON: full, boards (name -> 'all' | [tests]), families (bsp families the diff
+touches, including ones with no rig board - build-only consumers such as /pre-pr
+sample from these), args (hil_test.py args per config) and args_flasher (the same
+args split by each board's flasher, for CI legs that split one rig by flasher).
+"""
+import argparse
+import functools
+import glob
+import json
+import os
+import re
+import subprocess
+import sys
+
+from hil_examples import device_tests, dual_tests, host_test
+
+ALL_TESTS = {'device': device_tests, 'dual': dual_tests, 'host': host_test}
+
+# class dir -> config macro suffix exceptions (rule 3); dfu is per-file, handled inline
+NET_MACROS = ('ECM_RNDIS', 'NCM')
+
+_NONCODE_RE = re.compile(
+ r'^(docs/|\.claude/|.*\.(md|rst)$|LICENSE)')
+_FULL_RE = re.compile(
+ r'^(src/common/|src/osal/|src/tusb\.c$|src/tusb\.h$|src/tusb_option\.h$|'
+ r'test/hil/|\.github/workflows/build.*\.yml$|\.github/actions/|'
+ r'tools/build\.py$|tools/get_deps\.py$|tools/cmake/|hw/mcu/|lib/|'
+ r'hw/bsp/(family_support\.cmake|board_api\.h|board\.c|ansi_escape\.h)$|'
+ r'examples/build_system/|examples/CMakeLists\.txt$|'
+ # board_test is HIL infrastructure, not a test: hil_test.py flashes it to park
+ # every board (variant boundary + end-of-board teardown), so every board depends on it
+ r'examples/device/board_test/)')
+
+# --no-renames: with rename detection git reports only a rename's destination, so code
+# moved out of an HIL-relevant path would be classified by its new path alone
+GIT_DIFF_ARGV = ['git', 'diff', '--no-renames', '--name-only']
+
+
+def test_role(test: str) -> str:
+ return test.split('/', 1)[0] # 'device' | 'dual' | 'host'
+
+
+def board_roles(board: dict) -> set:
+ t = board.get('tests', {})
+ roles = set()
+ if t.get('device'):
+ roles.add('device')
+ if t.get('host'):
+ roles.add('host')
+ if t.get('dual'):
+ roles.update(('device', 'host'))
+ for only in t.get('only', []):
+ r = test_role(only)
+ roles.update(('device', 'host') if r == 'dual' else (r,))
+ return roles
+
+
+def board_tests(board: dict) -> list:
+ """Every test this board would run today (mirrors hil_test.test_board's default)."""
+ t = board.get('tests', {})
+ if 'only' in t:
+ run = list(t['only'])
+ else:
+ run = []
+ if t.get('device'):
+ run += device_tests
+ if t.get('dual'):
+ run += dual_tests
+ if t.get('host'):
+ run += host_test
+ return [x for x in run if x not in t.get('skip', [])]
+
+
+# cached: called per changed file x roster board, and the tree doesn't change mid-run
[email protected]_cache(maxsize=None)
+def board_family(board_name: str, repo_root: str):
+ hits = glob.glob(os.path.join(repo_root, 'hw/bsp/*/boards', board_name))
+ return os.path.basename(os.path.dirname(os.path.dirname(hits[0]))) if hits else None
+
+
+# `if (OPTION STREQUAL "1")` guards in family_support.cmake, and the option tokens
+# a roster entry passes to the build (NAME=VALUE / -DNAME=VALUE)
+_CM_IF_RE = re.compile(r'if\s*\(')
+_CM_ELSE_RE = re.compile(r'else(if)?\s*\(')
+_CM_ENDIF_RE = re.compile(r'endif\s*\(')
+_CM_OPT_RE = re.compile(r'if\s*\(\s*\$?\{?([A-Za-z_]\w*)\}?\s+STREQUAL\s+"?1"?\s*\)')
+_CM_PORT_RE = re.compile(r'src/portable/((?:[^/\s]+/)?[^/\s]+)/')
+_FALSY = ('', '0', 'off', 'false', 'no')
+
+
[email protected]_cache(maxsize=None)
+def port_option_gates(repo_root: str) -> dict:
+ """port dir -> build options that compile it regardless of the board's family
+ file, e.g. {'analog/max3421': {'MAX3421_HOST'}} from family_support.cmake."""
+ gates = {}
+ try:
+ text = open(os.path.join(repo_root, 'hw/bsp/family_support.cmake')).read()
+ except OSError:
+ return gates
+ stack = [] # one entry per open if(): its option, or None
+ for line in text.splitlines():
+ line = line.strip()
+ if _CM_IF_RE.match(line):
+ m = _CM_OPT_RE.match(line)
+ stack.append(m.group(1) if m else None)
+ elif _CM_ELSE_RE.match(line):
+ if stack:
+ stack[-1] = None # the guard doesn't hold in this branch
+ elif _CM_ENDIF_RE.match(line):
+ if stack:
+ stack.pop()
+ opts = {o for o in stack if o}
+ m = _CM_PORT_RE.search(line)
+ if opts and m:
+ gates.setdefault(m.group(1), set()).update(opts)
+ return gates
+
+
+_CM_SET_RE = re.compile(r'set\s*\(\s*([A-Za-z_]\w*)\s+([^)\s]+)\s*\)')
+
+
+# cached: called per changed portable file x roster board
[email protected]_cache(maxsize=None)
+def bsp_board_options(board_name: str, repo_root: str) -> frozenset:
+ """Build options a board turns on in its own BSP: `set(<OPT> <value>)` in
+ hw/bsp/<family>/boards/<board>/board.cmake, e.g. MAX3421_HOST on the espressif
+ and rp2040 max3421 boards. CMake only - HIL CI builds nothing with Make, so a
+ board.mk-only option (e.g. nrf5340dk's MAX3421_HOST) compiles no port here."""
+ fam = board_family(board_name, repo_root)
+ if not fam:
+ return frozenset()
+ path = os.path.join(repo_root, 'hw/bsp', fam, 'boards', board_name, 'board.cmake')
+ try:
+ text = open(path).read()
+ except OSError:
+ return frozenset()
+ out = set()
+ for line in text.splitlines():
+ line = line.strip()
+ if line.startswith('#'):
+ continue
+ m = _CM_SET_RE.match(line)
+ if m and m.group(2).strip('"').lower() not in _FALSY:
+ out.add(m.group(1))
+ return frozenset(out)
+
+
+def board_options(board: dict, repo_root: str) -> set:
+ """Build options a board has truthy: the roster entry's build.args plus each
+ variant's defines (NAME=VALUE) and raw CFLAGS (-DNAME=VALUE), plus whatever its
+ own board.cmake sets (a board can enable a gated port without the roster saying so)."""
+ toks = list(board.get('build', {}).get('args', []))
+ for v in board.get('variant', []):
+ toks += list(v.get('defines', []))
+ toks += v.get('flags', '').split()
+ out = set(bsp_board_options(board['name'], repo_root))
+ for t in toks:
+ name, _, val = (t[2:] if t.startswith('-D') else t).partition('=')
+ if name and val.strip().strip('"').lower() not in _FALSY:
+ out.add(name.strip())
+ return out
+
+
[email protected]_cache(maxsize=None)
+def port_families(port_dir: str, repo_root: str) -> set:
+ """Board families that compile this src/portable dir. CMake only: HIL CI builds
+ every board with CMake, so a port wired up in family.mk alone is compiled for no
+ HIL board and must not select one. family.cmake lists portable sources directly
+ for most families; espressif instead references them from a nested component
+ CMakeLists.txt (hw/bsp/espressif/components/tinyusb_src/CMakeLists.txt)."""
+ fams = set()
+ bsp_root = os.path.join(repo_root, 'hw/bsp')
+ # trailing '/' so a port dir is not a prefix of a sibling: bare 'microchip/pic'
+ # would otherwise match '.../microchip/pic32mz/...' and inherit its families
+ needle = port_dir + '/'
+ for f in glob.glob(os.path.join(bsp_root, '*/family.cmake')) + \
+ glob.glob(os.path.join(bsp_root, '*/components/*/CMakeLists.txt')):
+ try:
+ if needle in open(f).read():
+ fam = os.path.relpath(f, bsp_root).split(os.sep, 1)[0]
+ fams.add(fam)
+ except OSError:
+ pass
+ return fams
+
+
+_CLS_INC_RE = re.compile(r'#\s*include\s*[<"]class/([^/"<>]+)/([^"<>]+)[">]')
+
+
[email protected]_cache(maxsize=None)
+def class_include_edges(repo_root: str) -> dict:
+ """'<class>/<header>' -> the other class dirs that include it. A class header
+ pulled in by a second class ships in every firmware enabling that second class:
+ src/class/midi/midi{,2}_{device,host}.h include class/audio/audio.h, and
+ net_device.h includes class/cdc/cdc.h. The class rule derives macros from the
+ directory name alone, so without this edge a change to the included header
+ selects only its own class's examples - and on a board that skips those (e.g.
+ metro_m4_express skips audio_test_freertos), nothing at all.
+
+ Derived from the actual #include lines rather than a hand-written table so it
+ cannot rot when a class picks up or drops a cross-class include."""
+ edges = {}
+ for f in sorted(glob.glob(os.path.join(repo_root, 'src/class/*/*.[ch]'))):
+ cls = os.path.basename(os.path.dirname(f))
+ try:
+ text = open(f).read()
+ except OSError:
+ continue
+ for inc_cls, inc_hdr in _CLS_INC_RE.findall(text):
+ if inc_cls != cls:
+ edges.setdefault(f'{inc_cls}/{inc_hdr}', set()).add(cls)
+ return edges
+
+
+def class_macros(cls: str, base: str, prefix: str) -> list:
+ """Config macros that compile a class dir's code, for role prefix TUD/TUH.
+ `base` refines dfu only (it splits DFU from DFU_RUNTIME per file); pass '' for
+ a class reached through an include edge, where the widest set is correct."""
+ if cls == 'net':
+ return [f'CFG_{prefix}_{m}' for m in NET_MACROS]
+ if cls == 'dfu':
+ if base.startswith('dfu_rt'):
+ return [f'CFG_{prefix}_DFU_RUNTIME']
+ if base.startswith('dfu_device') or base.startswith('dfu_host'):
+ return [f'CFG_{prefix}_DFU']
+ return [f'CFG_{prefix}_DFU', f'CFG_{prefix}_DFU_RUNTIME']
+ return [f'CFG_{prefix}_{cls.upper()}']
+
+
+def _config_enables(cfg_path: str, macros) -> bool:
+ try:
+ text = open(cfg_path).read()
+ except OSError:
+ return False
+ return any(re.search(rf'#define\s+{m}\s+\(?\s*0*[1-9]', text) for m in macros)
+
+
+def roster_only_tests(all_boards) -> set:
+ """Test paths that only appear in a roster board's tests.only list (e.g.
+ espressif boards), not in the shared device/dual/host_test lists."""
+ out = set()
+ for b in all_boards:
+ out.update(b.get('tests', {}).get('only', []))
+ return out
+
+
+def class_examples(macros, role: str, repo_root: str, extra_tests: set) -> set:
+ """Tests (from role's + dual lists, plus roster-only-list tests of that role)
+ whose example config enables any macro."""
+ pool = role_tests({role}, extra_tests)
+ out = set()
+ for test in pool:
+ cfg = os.path.join(repo_root, 'examples', test, 'src', 'tusb_config.h')
+ if _config_enables(cfg, macros):
+ out.add(test)
+ return out
+
+
+def role_tests(roles: set, extras: set) -> set:
+ """Every test for the given role(s): each role's own list + dual tests,
+ plus roster-only-list tests (extras) matching those roles or 'dual'."""
+ pool = set(dual_tests)
+ for r in roles:
+ pool |= set(ALL_TESTS[r])
+ pool |= {t for t in extras if test_role(t) in roles or test_role(t) == 'dual'}
+ return pool
+
+
+class _Sel:
+ """Accumulates contributions. board->set(tests) plus 'all-board' markers."""
+ def __init__(self):
+ self.full = False
+ self.by_board = {} # name -> set of tests, or 'all'
+ self.roles = set() # roles touched by any contribution
+ self.families = set() # bsp families touched (incl. off-rig ones: build-only consumers)
+ self.reasons = []
+
+ def add(self, boards, tests, reason):
+ """tests: 'all' or iterable of test paths."""
+ self.reasons.append(reason)
+ for b in boards:
+ cur = self.by_board.get(b)
+ if tests == 'all' or cur == 'all':
+ self.by_board[b] = 'all'
+ else:
+ self.by_board[b] = (cur or set()) | set(tests)
+
+ def force_full(self, reason):
+ self.full = True
+ self.reasons.append(reason)
+
+
+def _classify_one(path, repo_root, roster_boards, extras: set, s: _Sel):
+ base = os.path.basename(path)
+ if _NONCODE_RE.match(path):
+ s.reasons.append(f'{path}: non-code, no contribution')
+ return
+ if _FULL_RE.match(path):
+ s.force_full(f'{path}: core/infra -> full matrix')
+ return
+
+ m = re.match(r'src/portable/((?:[^/]+/)?[^/]+)/', path)
+ if m:
+ port = m.group(1)
+ if re.match(r'(dcd_|.*_device)', base):
+ roles = {'device'}
+ elif re.match(r'(hcd_|.*_host)', base):
+ roles = {'host'}
+ else:
+ roles = {'device', 'host'}
+ fams = port_families(port, repo_root)
+ if not fams:
+ # no family references this port: either a new/renamed port dir or a
+ # family.cmake layout the scan misses - widen instead of contributing nothing
+ s.force_full(f'{path}: port {port} maps to no board family -> full matrix')
+ return
+ s.families.update(fams)
+ # a board can also pull the port in through a build option (e.g. MAX3421_HOST=1
+ # from the roster on metro_m4_express, or from its own board.cmake), which its
+ # family file never names
+ gates = port_option_gates(repo_root).get(port, set())
+ boards = [b['name'] for b in roster_boards
+ if (board_family(b['name'], repo_root) in fams or
+ (gates and board_options(b, repo_root) & gates)) and (board_roles(b) & roles)]
+ tests = role_tests(roles, extras)
+ s.roles.update(roles)
+ why = f'{path}: port {port} -> families {sorted(fams)}'
+ if gates:
+ why += f' + option {sorted(gates)}'
+ s.add(boards, tests, f'{why} -> boards {boards} ({"/".join(sorted(roles))})')
+ return
+
+ m = re.match(r'src/class/([^/]+)/', path)
+ if m:
+ cls = m.group(1)
+ if re.search(r'_device\.[ch]$', base):
+ roles = {'device'}
+ elif re.search(r'_host\.[ch]$', base):
+ roles = {'host'}
+ else:
+ roles = {'device', 'host'}
+ # this file's own class, plus any class whose headers include it
+ via = sorted(class_include_edges(repo_root).get(f'{cls}/{base}', ()))
+
+ def macros(prefix):
+ return (class_macros(cls, base, prefix) +
+ [m2 for c in via for m2 in class_macros(c, '', prefix)])
+ tests = set()
+ if 'device' in roles:
+ tests |= class_examples(macros('TUD'), 'device', repo_root, extras)
+ if 'host' in roles:
+ tests |= class_examples(macros('TUH'), 'host', repo_root, extras)
+ boards = [b['name'] for b in roster_boards if board_roles(b) & roles]
+ s.roles.update(roles)
+ why = f'{path}: class {cls}' + (f' (+ included by {via})' if via else '')
+ s.add(boards, tests, f'{why} -> {sorted(tests)} ({"/".join(sorted(roles))})')
+ return
+
+ m = re.match(r'src/(device|host)/', path)
+ if m:
+ role = m.group(1)
+ boards = [b['name'] for b in roster_boards if role in board_roles(b)]
+ s.roles.add(role)
+ s.add(boards, role_tests({role}, extras), f'{path}: core {role} stack -> all {role} tests')
+ return
+
+ m = re.match(r'hw/bsp/([^/]+)/(?:boards/([^/]+)/)?', path)
+ if m:
+ fam, brd = m.group(1), m.group(2)
+ s.families.add(fam)
+ if brd:
+ boards = [b['name'] for b in roster_boards if b['name'] == brd]
+ why = f'{path}: bsp board {brd}'
+ else:
+ boards = [b['name'] for b in roster_boards
+ if board_family(b['name'], repo_root) == fam]
+ why = f'{path}: bsp family {fam}'
+ s.roles.update(('device', 'host'))
+ s.add(boards, 'all', f'{why} -> boards {boards}')
+ return
+
+ m = re.match(r'examples/(device|host|dual)/([^/]+)/', path)
+ if m:
+ test = f'{m.group(1)}/{m.group(2)}'
+ known = any(test in pool for pool in ALL_TESTS.values()) or test in extras
+ if known:
+ boards = [b['name'] for b in roster_boards]
+ role = test_role(test)
+ s.roles.update(('device', 'host') if role == 'dual' else (role,))
+ s.add(boards, [test], f'{path}: example -> {test} on all boards')
+ else:
+ s.reasons.append(f'{path}: example not in HIL lists, no contribution')
+ return
+
+ s.force_full(f'{path}: unclassified -> full matrix')
+
+
+def classify(changed_files, repo_root, rosters):
+ all_boards = []
+ seen = set()
+ for _, boards in rosters:
+ for b in boards:
+ if b['name'] not in seen:
+ seen.add(b['name'])
+ all_boards.append(b)
+
+ extras = roster_only_tests(all_boards)
+ s = _Sel()
+ # no early exit once full: keep classifying so `families` still reports every
+ # family the diff touches (build-only consumers need it). Nothing after the first
+ # force_full can change full/boards/args - the full branch below ignores by_board.
+ for path in changed_files:
+ _classify_one(path, repo_root, all_boards, extras, s)
+
+ if s.full:
+ return {'full': True, 'boards': {b['name']: 'all' for b in all_boards},
+ 'families': sorted(s.families), 'reasons': s.reasons}
+
+ # role pruning: single-role selections drop the other role's tests and boards
+ by_name = {b['name']: b for b in all_boards}
+ out = {}
+ for name, tests in s.by_board.items():
+ allowed = board_tests(by_name[name])
+ if tests == 'all':
+ kept = list(allowed)
+ else:
+ kept = [t for t in allowed if t in tests]
+ if s.roles and s.roles != {'device', 'host'}:
+ role = next(iter(s.roles))
+ kept = [t for t in kept if test_role(t) in (role, 'dual')]
+ if kept:
+ out[name] = 'all' if set(kept) == set(allowed) else sorted(kept)
+ return {'full': False, 'boards': out, 'families': sorted(s.families),
+ 'reasons': s.reasons}
+
+
+def _board_args(name, chosen) -> list:
+ parts = [f'-b {name}']
+ if chosen != 'all':
+ parts.append(f'-bt {name}:{",".join(chosen)}')
+ return parts
+
+
+def selection_args(sel, rosters):
+ """hil_test.py args per config. Empty means either 'full matrix' or 'nothing
+ selected' - callers must read sel['full'] to tell them apart."""
+ args = {}
+ for cfg_path, boards in rosters:
+ parts = []
+ if not sel['full']:
+ for b in boards:
+ chosen = sel['boards'].get(b['name'])
+ if chosen is not None:
+ parts += _board_args(b['name'], chosen)
+ args[os.path.basename(cfg_path)] = ' '.join(parts)
+ return args
+
+
+def selection_args_by_flasher(sel, rosters):
+ """{config: {flasher name: args}}. CI runs one rig as several jobs split by
+ flasher (esptool vs the rest); each must gate on its own subset, otherwise the
+ other leg runs a filter matching zero boards and reports a vacuous green."""
+ out = {}
+ for cfg_path, boards in rosters:
+ per = {}
+ if not sel['full']:
+ for b in boards:
+ chosen = sel['boards'].get(b['name'])
+ if chosen is None:
+ continue
+ per.setdefault(b.get('flasher', {}).get('name', ''), []).extend(
+ _board_args(b['name'], chosen))
+ out[os.path.basename(cfg_path)] = {f: ' '.join(p) for f, p in per.items()}
+ return out
+
+
+def changed_files_from_git(base, repo_root):
+ mb = subprocess.run(['git', 'merge-base', 'HEAD', base], cwd=repo_root,
+ capture_output=True, text=True, check=True).stdout.strip()
+ diff = subprocess.run(GIT_DIFF_ARGV + [f'{mb}..HEAD'], cwd=repo_root,
+ capture_output=True, text=True, check=True).stdout
+ return [l for l in diff.splitlines() if l.strip()]
+
+
+def main():
+ ap = argparse.ArgumentParser(description=__doc__)
+ g = ap.add_mutually_exclusive_group(required=True)
+ g.add_argument('--base', help='git ref to diff against (merge-base..HEAD)')
+ g.add_argument('--diff-file', help='newline-separated changed-file list')
+ ap.add_argument('configs', nargs='+', help='rig roster JSON file(s)')
+ a = ap.parse_args()
+
+ repo_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+ rosters = []
+ for c in a.configs:
+ with open(c) as f:
+ rosters.append((c, json.load(f)['boards']))
+
+ files = (open(a.diff_file).read().splitlines() if a.diff_file
+ else changed_files_from_git(a.base, repo_root))
+ files = [f for f in files if f.strip()]
+
+ s = classify(files, repo_root, rosters)
+ s['args'] = selection_args(s, rosters)
+ s['args_flasher'] = selection_args_by_flasher(s, rosters)
+ for r in s['reasons']:
+ print(f'hil_select: {r}', file=sys.stderr)
+ print(json.dumps(s))
+
+
+if __name__ == '__main__':
+ main()