diff options
Diffstat (limited to 'test/hil/hil_test.py')
| -rwxr-xr-x | test/hil/hil_test.py | 211 |
1 files changed, 145 insertions, 66 deletions
diff --git a/test/hil/hil_test.py b/test/hil/hil_test.py index 5f4cef7a6..bfb73234f 100755 --- a/test/hil/hil_test.py +++ b/test/hil/hil_test.py @@ -66,6 +66,51 @@ import ctypes from pymtp import MTP import string +# --- per-board dev-session locks (see test/hil/board_lock.py) ------------ +BOARD_LOCK_DIR = '/tmp/tinyusb-hil-locks' + +def acquire_board_lock(board_name): + """Take this board's flock for the duration of its flash+test. + Returns an open file handle (keep it referenced; closing releases it), + or None when HIL_NO_BOARD_LOCK=1 or the lock dir is unusable (fail-open: + locking must never break a test run by itself). + Raises RuntimeError only when another session holds the board.""" + import fcntl + if os.environ.get('HIL_NO_BOARD_LOCK') == '1': + return None # user-authorized bypass — see board_lock.py / hil skill + try: + os.makedirs(BOARD_LOCK_DIR, exist_ok=True) + fd = os.open(os.path.join(BOARD_LOCK_DIR, f'{board_name}.lock'), + os.O_RDWR | os.O_CREAT, 0o666) + fh = os.fdopen(fd, 'r+') + except OSError as e: + # odd lock dir (perms, path collision): proceed unlocked, but say so — + # a silent fail-open is indistinguishable from the intentional bypass + print(f'warning: board lock unavailable for {board_name} ({e}); proceeding unlocked', + flush=True) + return None + try: + fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB) + except OSError: + try: + info = fh.read(500).strip() + except (OSError, UnicodeDecodeError): + info = '' + fh.close() + raise RuntimeError(f'board locked: {info or "unknown holder"}') + # announce ourselves so the other side's conflict message is truthful; + # best-effort — the flock itself is already held + try: + fh.truncate(0) + fh.seek(0) + json.dump({'pid': os.getpid(), 'reason': 'hil_test.py', + 'since': time.strftime('%Y-%m-%dT%H:%M:%S%z')}, fh) + fh.flush() + except OSError: + pass + return fh + + # Enumeration wait budget. The first attempt gets ENUM_TIMEOUT; retry attempts get the # shorter ENUM_TIMEOUT_RETRY - the board was just re-flashed again, and a device that is # going to enumerate shows up within a few seconds, so a failing test costs ~3-5x a @@ -1885,77 +1930,96 @@ def test_board(board: Board) -> tuple[str, int, list[str], list]: name = board['name'] flasher = board['flasher'] - # default to all tests - test_list = [] + try: + _lock_fh = acquire_board_lock(name) + except RuntimeError as e: + log_line(f'{name:25} {STATUS_FAILED}: {e}') + # visible report row so the ❌ matches the exit code; failed-tests stays + # empty so a re-run repeats the whole board (no bogus -bt test filter) + return name, 1, [], [(name, {'board-locked': 'fail'})] + try: + # default to all tests + test_list = [] - if name in board_test: - test_list = board_test[name] - elif len(test_only) > 0: - # Explicit -t: filter against the board's capabilities so a device-only - # board doesn't try to run host/dual tests (the test functions need a - # `dev_attached` entry in the board config that won't exist). - board_tests = board.get('tests', {}) - if 'only' in board_tests: - allowed = set(board_tests['only']) - test_list = [t for t in test_only if t in allowed] - else: - for t in test_only: - category = t.split('/', 1)[0] - if board_tests.get(category) is True: - test_list.append(t) - else: - if 'tests' in board: - board_tests = board['tests'] - if board_tests.get('device') is True: - test_list += list(device_tests) - if board_tests.get('dual') is True: - test_list += dual_tests - if board_tests.get('host') is True: - test_list += host_test + if name in board_test: + test_list = board_test[name] + elif len(test_only) > 0: + # Explicit -t: filter against the board's capabilities so a device-only + # board doesn't try to run host/dual tests (the test functions need a + # `dev_attached` entry in the board config that won't exist). + board_tests = board.get('tests', {}) if 'only' in board_tests: - test_list = board_tests['only'] - if 'skip' in board_tests: - for skip in board_tests['skip']: - if skip in test_list: - test_list.remove(skip) - log_line(f'{name:25} {skip:30} ... Skip') + allowed = set(board_tests['only']) + test_list = [t for t in test_only if t in allowed] + else: + for t in test_only: + category = t.split('/', 1)[0] + if board_tests.get(category) is True: + test_list.append(t) + else: + if 'tests' in board: + board_tests = board['tests'] + if board_tests.get('device') is True: + test_list += list(device_tests) + if board_tests.get('dual') is True: + test_list += dual_tests + if board_tests.get('host') is True: + test_list += host_test + if 'only' in board_tests: + test_list = board_tests['only'] + if 'skip' in board_tests: + for skip in board_tests['skip']: + if skip in test_list: + test_list.remove(skip) + log_line(f'{name:25} {skip:30} ... Skip') - err_count = 0 - failed_tests = [] - rows = [] # list of (row_label, {example: status}) — one row per build variant - variants = board.get('variant') or [{'name': name, 'flags': ''}] + err_count = 0 + failed_tests = [] + rows = [] # list of (row_label, {example: status}) — one row per build variant + variants = board.get('variant') or [{'name': name, 'flags': ''}] - prev_last = None # last test of the previous variant: the variant boundary is an adjacency too - for v in variants: - vname = v['name'] - # Shuffle each (board, variant)'s run order — de-synchronizes the worker pool so - # usbtest batteries and flash churn spread across the timeline instead of convoying, - # and surfaces order-dependent bugs. Seeded for replay (HIL_SHUFFLE_SEED, logged by - # main). Unique per-example PIDs make any two different examples re-enumerate; only - # the variant boundary can repeat the same example (same PID) — swap it away. - run_list = list(test_list) - if shuffle_seed is not None and len(run_list) > 1: - random.Random(f'{shuffle_seed}:{name}:{vname}').shuffle(run_list) - if run_list[0] == prev_last: - run_list[0], run_list[-1] = run_list[-1], run_list[0] - log_line(f'{vname:40} test order: {", ".join(t.rsplit("/", 1)[-1] for t in run_list)}') - if run_list: - prev_last = run_list[-1] - cells = {} - for test in run_list: - ec, status, metric = test_example(board, vname, test) - err_count += ec - cells[test] = metric if metric else status - if ec > 0: - failed_tests.append(test) - rows.append((vname, cells)) + prev_last = None # last test of the previous variant: the variant boundary is an adjacency too + for v in variants: + vname = v['name'] + # Shuffle each (board, variant)'s run order — de-synchronizes the worker pool so + # usbtest batteries and flash churn spread across the timeline instead of convoying, + # and surfaces order-dependent bugs. Seeded for replay (HIL_SHUFFLE_SEED, logged by + # main). Unique per-example PIDs make any two different examples re-enumerate; only + # the variant boundary can repeat the same example (same PID) — swap it away. + run_list = list(test_list) + if shuffle_seed is not None and len(run_list) > 1: + random.Random(f'{shuffle_seed}:{name}:{vname}').shuffle(run_list) + if run_list[0] == prev_last: + run_list[0], run_list[-1] = run_list[-1], run_list[0] + log_line(f'{vname:40} test order: {", ".join(t.rsplit("/", 1)[-1] for t in run_list)}') + if run_list: + prev_last = run_list[-1] + cells = {} + for test in run_list: + ec, status, metric = test_example(board, vname, test) + err_count += ec + cells[test] = metric if metric else status + if ec > 0: + failed_tests.append(test) + rows.append((vname, cells)) - # flash board_test last to disable board's usb (skipped when --skip-flash is set); - # this is teardown/park, not a test — not recorded in the report - if not skip_flash: - test_example(board, variants[0]['name'], 'device/board_test') + # flash board_test last to disable board's usb (skipped when --skip-flash is set); + # this is teardown/park, not a test — not recorded in the report + if not skip_flash: + test_example(board, variants[0]['name'], 'device/board_test') - return name, err_count, sorted(set(failed_tests)), rows + return name, err_count, sorted(set(failed_tests)), rows + finally: + if _lock_fh: + try: + # clear our pid record before dropping the flock: this worker + # process lives on (pool reuse), so a stale record would make + # board_lock.py's pid-liveness checks report a freed board as + # still locked for the rest of the run + _lock_fh.truncate(0) + except OSError: + pass + _lock_fh.close() REPORT_MD = 'hil_report.md' @@ -2035,7 +2099,17 @@ def accumulate_report(mret: list, report_dir: Path, fresh: bool) -> str: pass # corrupt/old sidecar: start fresh # merge this run: current cells override prior for boards/tests that ran - for _, _, _, rows in mret: + for name, _, _, rows in mret: + if rows and not any('board-locked' in cells for _, cells in rows): + # board ran for real this time: clear a stale lock-failure cell + # (its row is keyed by board name; test rows may be variant names) + stale = acc.get(name) + if stale is not None: + stale.pop('board-locked', None) + if not stale: + # variant-keyed boards never repopulate the board-name row — + # drop it or it renders as a blank ghost row + del acc[name] for row_label, cells in rows: acc.setdefault(row_label, {}).update(cells) @@ -2098,6 +2172,11 @@ def main() -> None: if len(boards) == 0: config_boards = [e for e in config['boards'] if e['name'] not in skip_boards] else: + unknown = [b for b in boards if b not in {e['name'] for e in config['boards']}] + if unknown: + # exiting 0 with 'No tests were run.' would read as a green HIL run + print(f'ERROR: board(s) not in {config_file.name}: {", ".join(unknown)}') + sys.exit(1) config_boards = [e for e in config['boards'] if e['name'] in boards] build_err = 0 |