| #!/usr/bin/env python3 |
| # SPDX-License-Identifier: MIT |
| # Bottom layer of the HIL harness: the bounded command runner plus the shared helpers and |
| # data every other module needs. Stays stdlib-only and imports nothing local -- everything |
| # else imports this, including the unit tests on GitHub's bare runner; never import them |
| # from here. Callers set the module global `verbose`. |
| |
| from __future__ import annotations |
| |
| import glob |
| import os |
| import signal |
| import subprocess |
| import unicodedata |
| import threading |
| import sys |
| from pathlib import Path |
| from typing import Any |
| |
| |
| # ------------------------------------------------------------- |
| # HIL example test lists, shared by hil_test.py (runner) and ci_select.py (PR-diff |
| # selector). Run order is shuffled per board (see test_board); every example carries a |
| # unique hardcoded idProduct (see its usb_descriptors.c). |
| # ------------------------------------------------------------- |
| |
| # device tests |
| device_tests = [ |
| 'device/cdc_dual_ports', |
| 'device/cdc_msc', |
| 'device/dfu', |
| 'device/cdc_msc_throughput', |
| 'device/audio_test_freertos', |
| 'device/dfu_runtime', |
| 'device/cdc_msc_freertos', |
| 'device/hid_boot_interface', |
| 'device/msc_dual_lun', |
| 'device/hid_generic_inout', |
| 'device/printer_to_cdc', |
| 'device/midi_test', |
| 'device/mtp', |
| 'device/usbtest', # cafe:4010, unique PID; runs the Linux testusb tier-4 battery via usbtest.py |
| # 'device/net_lwip_webserver', # disabled for PR #3605: USB net iface enum is flaky on the CI HIL host |
| ] |
| |
| dual_tests = [ |
| 'dual/host_info_to_device_cdc', |
| ] |
| |
| host_test = [ |
| 'host/cdc_msc_hid', |
| 'host/msc_file_explorer', |
| 'host/msc_file_explorer_freertos', |
| 'host/device_info', |
| ] |
| |
| verbose = False |
| |
| def pos_int_env(name: str, default: int) -> int: |
| # One parsing policy for every HIL_* knob: a bare int() crashes every run at import |
| # on a malformed value, and 0/negative silently removes the bound the knob enforces. |
| try: |
| v = int(os.getenv(name, str(default))) |
| except ValueError: |
| print(f'warning: {name} is not an integer; using {default}', |
| file=sys.stderr, flush=True) |
| return default |
| if v <= 0: |
| print(f'warning: {name}={v} is not usable; using {default}', |
| file=sys.stderr, flush=True) |
| return default |
| return v |
| |
| |
| def pos_float_env(name: str, default: float) -> float: |
| try: |
| v = float(os.getenv(name, str(default))) |
| except ValueError: |
| print(f'warning: {name} is not a number; using {default}', |
| file=sys.stderr, flush=True) |
| return default |
| # float() accepts 'inf'/'nan': an infinite serial timeout is an unbounded read, the |
| # very thing these knobs exist to prevent, and nan fails every comparison silently |
| if not (v > 0 and v < float('inf')): |
| print(f'warning: {name}={v} is not usable; using {default}', |
| file=sys.stderr, flush=True) |
| return default |
| return v |
| |
| |
| CMD_TIMEOUT = pos_int_env('HIL_CMD_TIMEOUT', 180) |
| |
| TINYUSB_ROOT = Path(__file__).resolve().parents[3] # test/hil/helper/ -> repo root |
| |
| |
| def display_width(s: str) -> int: |
| """Terminal COLUMNS, not characters. |
| |
| The status marks the reports use -- ✅ ❌ ⚪ ⚠ 🔒 -- are one Python character and TWO |
| columns wide. Measuring with len() pads every cell containing one a column short, so |
| the pipes drift out of line against the header rule for the whole table. |
| """ |
| return sum(2 if unicodedata.east_asian_width(c) in 'WF' else 1 for c in s) |
| |
| |
| def pad(s: str, width: int, center: bool = False) -> str: |
| """str.ljust/center, measured in display columns. See display_width.""" |
| room = max(0, width - display_width(s)) |
| if not center: |
| return s + ' ' * room |
| left = room // 2 |
| return ' ' * left + s + ' ' * (room - left) |
| |
| |
| def cmd_stdout_text(out: Any) -> str: |
| if out is None: |
| return '' |
| if isinstance(out, bytes): |
| return out.decode('utf-8', errors='ignore') |
| return str(out) |
| |
| |
| def _banner_body(out: Any, err: Any) -> str: |
| # split_stderr callers keep the diagnostic in stderr — a banner of stdout alone |
| # would be blank exactly when something went wrong |
| body = cmd_stdout_text(out) |
| err_text = cmd_stdout_text(err) |
| if err_text: |
| body = f'{body}\n{err_text}' if body else err_text |
| return body |
| |
| |
| # Shared with compact_output's stripper in hil_test: duplicated literals let the two |
| # layers drift and reintroduce literal marker noise mid-row in the GitHub log. |
| GROUP_MARK, ENDGROUP_MARK = '::group::', '::endgroup::' |
| |
| |
| def strip_workflow_markers(line: str) -> str: |
| # run_cmd only ever emits markers at line start; mid-line is not a real case. |
| return line.removeprefix(GROUP_MARK).removeprefix(ENDGROUP_MARK) |
| |
| |
| def _ci_log_groups() -> bool: |
| # GitHub folds ::group::/::endgroup:: only at line start of the JOB's real stdout; a |
| # pool worker's capture is compacted into one row line, where they render literally. |
| return bool(os.getenv('CI')) and sys.stdout is sys.__stdout__ |
| |
| |
| def _print_banner(title: str, out: Any, err: Any) -> None: |
| print() |
| if _ci_log_groups(): |
| print(f'{GROUP_MARK}{title}') |
| print(_banner_body(out, err)) |
| print(ENDGROUP_MARK) |
| else: |
| print(title) |
| print(_banner_body(out, err)) |
| |
| |
| SYSFS_READ_GRACE = 2.0 # bound on one attribute read of a possibly-wedged device |
| SYSFS_STUCK_MAX = 4 # stranded readers tolerated before read_sysfs goes blind |
| _sysfs_stuck = 0 # each costs a thread + an fd for the life of the process |
| _sysfs_stuck_lock = threading.Lock() |
| _sysfs_blind_logged = False |
| |
| |
| class _SysfsUnknown: |
| """Sentinel: the read did not answer. NOT "the attribute is absent" -- reading it as |
| absence turns a healthy board into a firmware regression in the report.""" |
| __slots__ = () |
| |
| def __bool__(self) -> bool: |
| return False |
| |
| def __repr__(self) -> str: |
| return 'SYSFS_UNKNOWN' |
| |
| |
| SYSFS_UNKNOWN = _SysfsUnknown() |
| |
| |
| def sysfs_blind() -> bool: |
| """True once this process has stranded SYSFS_STUCK_MAX readers: every later read |
| answers SYSFS_UNKNOWN, so nothing it reports about a device is a fact any more.""" |
| return _sysfs_stuck >= SYSFS_STUCK_MAX |
| |
| |
| def sysfs_blind_note() -> str: |
| """Suffix for a failure message, so a blind worker's verdict never reads as hardware.""" |
| return (f' (this worker is blind: {SYSFS_STUCK_MAX} sysfs reads stranded on a wedged ' |
| f'device, so the check could not see the bus)') if sysfs_blind() else '' |
| |
| |
| def read_sysfs(path: str, grace: float = SYSFS_READ_GRACE) -> str | None | _SysfsUnknown: |
| """Read a sysfs attribute with a WALL-CLOCK bound. |
| |
| The value, None when the attribute is genuinely unreadable (OSError), or SYSFS_UNKNOWN |
| when the read did not answer -- it timed out, or this process is already blind. Callers |
| MUST keep those apart: absence is a fact, unknown is not. |
| |
| usb_string_attr (serial/product/manufacturer) is served under the device lock a wedged |
| usbfs ioctl holds, so a plain open().read() blocks for as long as the wedge lasts, on |
| exactly the board an incident is about. The reader sleeps INTERRUPTIBLY (every read |
| takes usb_lock_device_interruptible, v6.12.96 sysfs.c:124-139 -- uninterruptible is the |
| ioctl holder, not us), so it dies with a SIGKILLed worker; what it costs meanwhile is a |
| thread and an fd for this process's life, because on sysfs the open() SUCCEEDS and only |
| the read blocks. Measured: 20 blocking reads leave 20 live threads. |
| |
| Hence the cap: callers rescan (hil_lock's controller_of re-reads every unresolved |
| device on EVERY permit), and hitting RLIMIT_NOFILE or the thread ceiling raises inside |
| the worker and loses every board's result -- worse than the hang this prevents. |
| """ |
| if sysfs_blind(): |
| return SYSFS_UNKNOWN |
| # Known-stranded? Re-reading costs another permanent thread+fd and a blindness credit |
| # to learn what we already know. Lives HERE, not at the call sites: a call-site memo |
| # has to be remembered by every new scanner, and twice it was not. |
| was = _sysfs_stranded.get(path, _STRAND_MISS) |
| if was is not _STRAND_MISS: |
| if was is None: |
| return SYSFS_UNKNOWN # stranded, inode unknown: never re-read it |
| try: |
| if os.stat(path).st_ino == was: |
| return SYSFS_UNKNOWN # same node, still wedged |
| except OSError: |
| pass # gone: fall through, the read reports it |
| _sysfs_stranded.pop(path, None) # replaced or gone -> re-read it |
| out: dict = {} |
| |
| def _read(): |
| try: |
| with open(path) as f: |
| out['v'] = f.read().strip() |
| except (OSError, ValueError): |
| pass # no such attribute, or not text: unreadable, and that IS a fact |
| |
| t = threading.Thread(target=_read, daemon=True) |
| t.start() |
| t.join(grace) |
| # `out` FIRST, not is_alive() alone: a reader can deposit its value and still be alive |
| # for a moment afterwards, and counting that as a strand memoises a healthy attribute as |
| # unreadable and spends one of four blindness credits. bounded_open has always checked |
| # its box for the same reason. |
| if t.is_alive() and 'v' not in out: |
| # Count the PATH once, not once per reader. hil_pool_check runs -j4 by default, |
| # which equals SYSFS_STUCK_MAX, so four threads hitting ONE wedged device used to |
| # spend the entire blindness budget between them -- latching blind on the single |
| # wedge the tool was run to find. The strand is real for each thread, but the |
| # DEVICE is what the cap is about. |
| # Under the SAME lock as the counter: check-then-act here is a race, and |
| # hil_pool_check runs a ThreadPoolExecutor of exactly SYSFS_STUCK_MAX workers in |
| # ONE process, so four threads on one wedged path could each see `first` before any |
| # of them recorded it -- spending the whole blindness budget on a single device, |
| # which is what this memo exists to prevent. note_sysfs_strand takes the lock |
| # itself, so call it after releasing. |
| with _sysfs_stuck_lock: |
| first = path not in _sysfs_stranded |
| if first: |
| try: |
| # stat, never the thread's own open(): stat does not call ->show(), so |
| # it cannot block on the device lock the reader is stuck behind |
| _sysfs_stranded[path] = os.stat(path).st_ino |
| except OSError: |
| _sysfs_stranded[path] = None # unstattable, but still known-stranded |
| if first: |
| note_sysfs_strand() |
| return SYSFS_UNKNOWN |
| return out.get('v') |
| |
| |
| def note_sysfs_strand() -> None: |
| """Record ONE stranded sysfs reader. Shared by read_sysfs and bounded_open so both |
| account against a single counter -- the report caveat keys off it.""" |
| global _sysfs_stuck, _sysfs_blind_logged |
| with _sysfs_stuck_lock: |
| _sysfs_stuck += 1 |
| announce = sysfs_blind() and not _sysfs_blind_logged |
| _sysfs_blind_logged = _sysfs_blind_logged or announce |
| if announce: |
| # once per process, on stderr: a worker's stdout is compacted into one report |
| # row, where this would be lost among the test output |
| print(f'warning: {SYSFS_STUCK_MAX} sysfs reads stranded on a wedged device; ' |
| f'this process is now blind and answers SYSFS_UNKNOWN for every ' |
| f'attribute -- its verdicts about device presence are not evidence', |
| file=sys.stderr, flush=True) |
| |
| |
| # path -> the inode it had when its read stranded. A stranded attribute stays |
| # stranded until the DEVICE is replaced, and a re-enumeration destroys the kernfs |
| # node and makes a new one -- so a changed inode is the all-clear. Keyed by path |
| # alone it would outlive the wedge: a busport does not change when a board comes |
| # back on the same port, so the HUNG reflash this branch performs would recover a |
| # board the harness could then never see again. |
| _sysfs_stranded: dict = {} |
| # A stranded path whose inode could not be read is stored as None, so a plain .get() cannot |
| # tell 'known stranded, inode unknown' from 'never seen' -- and treating the first as the |
| # second re-reads it, stranding another permanent thread and fd every call. Distinct miss |
| # sentinel, so None keeps its own meaning. |
| _STRAND_MISS = object() |
| |
| |
| def usb_scan(vid_pid=None, serial=None, vid=None) -> tuple[list, bool]: |
| """Enumerated USB devices matching the filters, and whether anything is unknown. |
| |
| Returns ([{busport, dir, vid, pid, serial}], unknown). `unknown` True means a bounded |
| read did not answer, so absence is NOT proven -- the same contract as read_sysfs. |
| |
| Three rules, one implementation for every caller: |
| |
| * Root hubs excluded (glob `*-*`): no DUT is one, and scans including them measured |
| seconds slower (observation, no mechanism -- the "autosuspend wake" explanation was |
| wrong; usb_string_attr reads a cached string, sysfs.c:124-139). |
| * idVendor/idProduct first: lock-free `sysfs_emit` from udev->descriptor |
| (sysfs.c:688-705), so they rule out nearly every device for free. |
| * `serial` last and bounded: it is served under the lock a wedged ioctl holds, and a |
| path that already stranded is never re-read (each strand costs a thread and an fd |
| for this process's life). |
| """ |
| out = [] |
| unknown = False |
| for d in glob.glob('/sys/bus/usb/devices/*-*'): |
| # Interfaces are '<busport>:<cfg>.<ifnum>' (e.g. 2-4:1.0) -- they CONTAIN the |
| # colon, they do not end with it, so the original endswith() never fired and every |
| # scan opened idVendor/idProduct on all of them (measured: 31 of 44 matches). |
| if ':' in os.path.basename(d): |
| continue |
| try: |
| with open(os.path.join(d, 'idVendor')) as f: |
| dev_vid = f.read().strip() |
| with open(os.path.join(d, 'idProduct')) as f: |
| dev_pid = f.read().strip() |
| except OSError: |
| continue # vanished mid-walk, or not a device dir: a fact, not unknown |
| if vid_pid is not None and (dev_vid, dev_pid) != tuple(vid_pid): |
| continue # ruled out for free, without touching the locked attribute |
| if vid is not None and dev_vid != vid: |
| continue # same, for callers that know the VID but not the PID |
| sn = read_sysfs(os.path.join(d, 'serial')) |
| if sn is SYSFS_UNKNOWN: |
| unknown = True # read_sysfs memoises it; a repeat scan costs nothing |
| continue |
| if sn is None: |
| continue # no serial attribute: a fact |
| if serial is not None and sn.lower() != serial.lower(): |
| continue |
| out.append({'busport': os.path.basename(d), 'dir': d, |
| 'vid': dev_vid, 'pid': dev_pid, 'serial': sn}) |
| return out, unknown |
| |
| |
| def bounded_open(path: str, flags: int, timeout: float = SYSFS_READ_GRACE): |
| """os.open() with a wall-clock bound. |
| |
| The fd, None when the open genuinely FAILED (OSError: EBUSY, ENOENT, EACCES), or |
| SYSFS_UNKNOWN when it did not answer -- the same three-valued contract as read_sysfs, |
| and for the same reason: folding a fact into an unknown made an ordinary EBUSY read as |
| a wedged device and sent the operator hunting hardware that is healthy. |
| |
| An open CAN block on a wedged device -- not on O_NONBLOCK, which usblp_open never |
| consults, but on usb_autopm_get_interface(), a runtime-PM resume that does I/O |
| (v6.12.96 drivers/usb/class/usblp.c). It holds usblp_mutex while it waits, and that |
| mutex is driver-GLOBAL, so one wedged printer blocks opens of every usblp node. |
| |
| Unlike read_sysfs the stranded thread cleans up after itself: if we have given up it |
| closes the fd it eventually got, so only the thread leaks. Both sides take `handoff` |
| -- "store or close" and "abandon and drain" are a check-then-act pair that can |
| interleave into an fd stored after the box was drained, which would leak it into a |
| node that allows a SINGLE opener (usblp_open returns -EBUSY when usblp->used). |
| """ |
| # Same short-circuit as read_sysfs: once blind, another stranded thread buys nothing |
| # and the cap exists precisely to stop them accumulating. |
| if sysfs_blind(): |
| return SYSFS_UNKNOWN |
| # Known-stranded? Re-opening costs another thread, another fd and another blindness |
| # credit to learn what we already know -- and the printer test re-opens ONE lp node on |
| # every retry. Same memo and same inode check as read_sysfs. |
| was = _sysfs_stranded.get(path, _STRAND_MISS) |
| if was is not _STRAND_MISS: |
| if was is None: |
| return SYSFS_UNKNOWN # stranded, inode unknown: never re-read it |
| try: |
| if os.stat(path).st_ino == was: |
| return SYSFS_UNKNOWN |
| except OSError: |
| pass |
| _sysfs_stranded.pop(path, None) |
| box: dict = {} |
| done, abandoned = threading.Event(), threading.Event() |
| handoff = threading.Lock() |
| |
| def _open(): |
| try: |
| fd = os.open(path, flags) |
| except OSError: |
| done.set() |
| return |
| with handoff: |
| stored = not abandoned.is_set() |
| if stored: |
| box['fd'] = fd |
| if not stored: |
| try: |
| os.close(fd) |
| except OSError: |
| pass |
| done.set() |
| |
| threading.Thread(target=_open, daemon=True).start() |
| if not done.wait(timeout): |
| with handoff: |
| abandoned.set() |
| fd = box.pop('fd', None) # completed in the gap between timeout and flag |
| if fd is not None: |
| # It DID open, just after our deadline -- the thread finished, so nothing is |
| # stranded. Report unknown (we already gave up on it) but do not spend a |
| # blindness credit, and do not call a merely-slow node wedged. |
| try: |
| os.close(fd) |
| except OSError: |
| pass |
| return SYSFS_UNKNOWN |
| # counted like a stranded read_sysfs: the thread and (eventually) its fd are gone |
| # for the life of the process, and the cap exists to stop that reaching the |
| # thread/fd ceiling -- an exception there escapes the worker and loses every board. |
| # Memoised by inode so a retry of the same node does not pay again. |
| # same lock as read_sysfs, same reason |
| with _sysfs_stuck_lock: |
| first = path not in _sysfs_stranded |
| if first: |
| try: |
| _sysfs_stranded[path] = os.stat(path).st_ino |
| except OSError: |
| _sysfs_stranded[path] = None |
| if first: |
| note_sysfs_strand() |
| return SYSFS_UNKNOWN |
| return box.get('fd') |
| |
| |
| def _close_pipes(p: subprocess.Popen) -> None: |
| """Close OUR ends of an abandoned child's pipes. Never raises.""" |
| for pipe in (p.stdout, p.stderr, p.stdin): |
| try: |
| if pipe is not None: |
| pipe.close() |
| except OSError: |
| pass |
| |
| |
| def run_alongside(argv: list, work, timeout: int) -> subprocess.CompletedProcess: |
| """Run `argv` alongside `work()`, which runs in THIS thread, then reap it -- bounded. |
| |
| The read-while-we-write shape run_cmd cannot express: the caller needs the child |
| RUNNING while it does something else. Everything else about the contract is run_cmd's |
| -- own session, killpg, bounded reap, our pipe ends closed, rc 124 on the kill. |
| |
| A PROCESS, not a thread: an abandoned thread keeps the fd, and usblp_open returns |
| -EBUSY while usblp->used (v6.12.96 usblp.c), so every later open in this long-lived |
| worker would read as a wedged device. A killed process takes its fd with it. |
| |
| stdout is captured as BYTES and kept CLEAN -- a caller byte-compares it against the |
| payload it sent, so a single stderr byte (a PYTHONWARNINGS chirp, a sitecustomize |
| print, a .pth deprecation from a venv) would read as USB data corruption. stderr gets |
| its own pipe; communicate() drains both, so the split cannot deadlock. |
| `work` runs even if the child dies immediately -- the caller's own asserts decide. |
| """ |
| p = subprocess.Popen(argv, stdout=subprocess.PIPE, stderr=subprocess.PIPE, |
| start_new_session=True) |
| |
| def _reap() -> subprocess.CompletedProcess: |
| try: |
| out, err = p.communicate(timeout=timeout) |
| return subprocess.CompletedProcess(argv, p.returncode, out, err) |
| except subprocess.TimeoutExpired: |
| try: |
| os.killpg(p.pid, signal.SIGKILL) |
| except OSError: |
| p.kill() |
| try: |
| out, err = p.communicate(timeout=5) |
| except subprocess.TimeoutExpired: |
| # Outlasted SIGKILL: uninterruptible, still holding whatever it opened. |
| # Abandoned like any other stray -- but as a real child in its own |
| # session, so the containment sweep FINDS it (child_procs walks the ppid |
| # tree) and the report names it. That is the whole difference from a |
| # blocked thread, which no sweep can see and no signal can reach. |
| out, err = b'', b'' |
| _close_pipes(p) # our own fds must not leak either |
| return subprocess.CompletedProcess(argv, 124, out, err) |
| |
| try: |
| work() |
| except BaseException: |
| # Reap first so the child never outlives us, then let the caller's error through. |
| # A `return` inside a `finally` would SWALLOW it -- an assert in `work` would |
| # vanish and the caller would compare data it never finished sending. |
| _reap() |
| raise |
| return _reap() |
| |
| |
| def _cmd_label(cmd) -> str: |
| """A one-line name for a banner. An argv whose payload is a `python3 -c` program would |
| otherwise dump the whole body into the CI log, where run_cmd's banners are already the |
| noisiest thing in a failing row.""" |
| if isinstance(cmd, str): |
| return cmd |
| parts = [a if len(a) <= 60 else f'<{len(a)}-char program>' for a in cmd] |
| return ' '.join(parts) |
| |
| |
| def run_cmd(cmd: str | list, cwd: str | None = None, timeout: int | None = None, |
| binary: bool = False, split_stderr: bool = False, |
| quiet: bool = False) -> subprocess.CompletedProcess: |
| """Bounded subprocess: own session, killpg on expiry, rc 124 when it had to be killed. |
| |
| `cmd` is a shell STRING or an argv LIST. argv exists for a program that cannot survive |
| a trip through the shell -- a multi-line `python3 -c` body -- which is how the harness |
| runs a library call that no in-process bound can contain. A daemon thread cannot bound |
| a C call that holds the GIL, so for those the child process IS the bound. |
| """ |
| if timeout is None: |
| timeout = CMD_TIMEOUT |
| # binary: raw bytes (text mode's errors='replace' mangles non-UTF-8 file content). |
| # split_stderr: keep stderr out of stdout, for callers that parse stdout. quiet: no |
| # COMMAND FAILED banner, for retry loops that report failures themselves (timeouts |
| # still print: a killed child is always noteworthy). |
| popen_kwargs = { |
| 'cwd': cwd, |
| # a list goes straight to execve; only a string needs a shell to parse it |
| 'shell': isinstance(cmd, str), |
| 'stdout': subprocess.PIPE, |
| 'stderr': subprocess.PIPE if split_stderr else subprocess.STDOUT, |
| } |
| if not binary: |
| popen_kwargs.update({'text': True, 'encoding': 'utf-8', 'errors': 'replace'}) |
| # C-level setsid, same process-group semantics as preexec_fn=os.setsid but safe when |
| # called from threads (pool_check runs flashes from a thread pool) |
| popen_kwargs['start_new_session'] = True |
| |
| p = subprocess.Popen(cmd, **popen_kwargs) |
| try: |
| out, err = p.communicate(timeout=timeout) |
| r = subprocess.CompletedProcess(args=cmd, returncode=p.returncode, stdout=out, stderr=err) |
| except subprocess.TimeoutExpired as ex: |
| try: |
| os.killpg(p.pid, signal.SIGKILL) |
| except OSError: |
| # ProcessLookupError: already gone. PermissionError: an all-root group refuses |
| # the group kill -- letting either escape would skip the bounded reap, the pipe |
| # close and the rc-124 return this handler exists for. |
| pass |
| try: |
| out, err = p.communicate(timeout=10) |
| except subprocess.TimeoutExpired: |
| # Something in the group outlived SIGKILL: D state (truly unkillable), or |
| # root-owned because sudo FORKS rather than execs, so the wrapper dies and its |
| # root child does not. Abandon it and let the report name it; the harness never |
| # sudo-kills its way out. Our ends of its pipes must not leak, though: a pool |
| # worker lives for the whole run, so every wedged command would cost it two fds. |
| out, err = None, None |
| _close_pipes(p) |
| # prefer the post-kill buffers (supersets of the exception's), falling back to ex.* |
| # when the child was unkillable. TimeoutExpired carries BYTES even for a text-mode |
| # Popen, so the fallbacks must be decoded or a text-mode caller gets bytes exactly |
| # when the child wedged in D state. |
| def _typed(v): |
| if not binary and isinstance(v, bytes): |
| return v.decode('utf-8', errors='replace') |
| return v |
| |
| timeout_out = _typed(out or ex.stdout) or (b'' if binary else '') |
| # ...and never None: with split_stderr the SUCCESS path always yields a str/bytes, |
| # so a caller that does `r.stderr.strip()` works everywhere except the timeout -- |
| # the one path it was written for. Without split_stderr stderr stays None, as on |
| # the success path (it was merged into stdout). |
| timeout_err = _typed(err if err is not None else ex.stderr) |
| if split_stderr and timeout_err is None: |
| timeout_err = b'' if binary else '' |
| _print_banner(f'COMMAND TIMEOUT ({timeout}s): {_cmd_label(cmd)}', timeout_out, timeout_err) |
| return subprocess.CompletedProcess(args=cmd, returncode=124, stdout=timeout_out, stderr=timeout_err) |
| except BaseException: |
| # BaseException, not Exception (as in CPython's own subprocess.run): |
| # KeyboardInterrupt is the case that matters, and start_new_session put the child in |
| # its OWN group, so it never got the terminal's SIGINT -- without this, Ctrl-C |
| # leaves the flasher or testusb holding the probe and its usbfs node. Kill and |
| # close, never wait: this path must not add a hang of its own. |
| try: |
| os.killpg(p.pid, signal.SIGKILL) |
| except OSError: |
| pass |
| _close_pipes(p) |
| raise |
| |
| if r.returncode != 0 and not quiet: |
| _print_banner(f'COMMAND FAILED: {_cmd_label(cmd)}', r.stdout, r.stderr) |
| elif verbose: |
| print(cmd) |
| print(cmd_stdout_text(r.stdout)) |
| return r |
| |
| |
| # get usb serial by id |
| def get_serial_dev(id, vendor_str, product_str, ifnum): |
| if vendor_str and product_str: |
| # known vendor and product |
| vendor_str = vendor_str.replace(' ', '_') |
| product_str = product_str.replace(' ', '_') |
| return f'/dev/serial/by-id/usb-{vendor_str}_{product_str}_{id}-if{ifnum:02d}' |
| else: |
| # just use id: mostly for cp210x/ftdi flasher |
| pattern = f'/dev/serial/by-id/usb-*_{id}-if*' |
| port_list = glob.glob(pattern) |
| if len(port_list) == 0: |
| raise RuntimeError(f'No serial device found for {pattern}') |
| return port_list[0] |
