diff options
Diffstat (limited to 'test/hil')
| -rw-r--r-- | test/hil/helper/hil_pool_check.py | 42 | ||||
| -rw-r--r-- | test/hil/helper/hil_util.py | 26 | ||||
| -rw-r--r-- | test/hil/hil_ci.sh | 3 | ||||
| -rwxr-xr-x | test/hil/hil_test.py | 152 | ||||
| -rw-r--r-- | test/hil/test/test_ci_select.py | 5 | ||||
| -rw-r--r-- | test/hil/test/test_hil_rtt.py | 506 | ||||
| -rw-r--r-- | test/hil/test/test_hil_util.py | 6 |
7 files changed, 711 insertions, 29 deletions
diff --git a/test/hil/helper/hil_pool_check.py b/test/hil/helper/hil_pool_check.py index b92f0aee0..4623ce45f 100644 --- a/test/hil/helper/hil_pool_check.py +++ b/test/hil/helper/hil_pool_check.py @@ -360,7 +360,47 @@ def check_host_serial(board: dict, do_reset: bool = True, want_hello: bool = Fal do_reset=False listens to the firmware as-is: used right after a flash whose own reset already started it — a second openocd/JLink session back-to-back on - the same probe can fail transiently and leave the target halted.""" + the same probe can fail transiently and leave the target halted. + + "logger": "rtt" boards have no VCOM: the same check runs over the probe's RTT + console instead. The reset happens BEFORE the console opens (it owns the probe), + which also zeroes the .bss ring — so pre-reset backlog cannot count as life, and + without a reset Commander delivers the boot burst the preceding flash left.""" + if board.get('logger') == 'rtt': + if do_reset: + # a failed reset leaves the previous run's ring intact: attaching anyway would + # score stale output as life, so bail to host_alive's board_test reflash ladder + rc, err = call_flasher(getattr(hil_flash, f'reset_{board["flasher"]["name"].lower()}'), board) + if rc: + say(f'{board["name"]:26} reset failed: {err}') + return None + try: + ser = hil_util.JlinkRtt(board, timeout=0.3) + except hil_util.RttError as e: + say(f'{board["name"]:26} no RTT console: {e}') + return None + try: + data = b'' + deadline = time.monotonic() + SERIAL_WAIT + while time.monotonic() < deadline: + ser.write(b'U') + data += ser.read(256) + # JLinkExe's banner arrives whether or not the target is alive -- + # judged unfiltered it scores a dead board 'alive'. Same shared filter + # as test_host_device_info; complete_only drops a trailing partial + # line, so a banner FRAGMENT split by this read boundary cannot count + # as target output either. + td = hil_util.strip_banner(data, complete_only=True) + if want_hello: + if b'Hello from TinyUSB' in td: + return td + elif td and not boardtest_output(td): + return td + return hil_util.strip_banner(data) + except hil_util.RttError: + return None # console died mid-poll (server exited, probe dropped) + finally: + ser.close() import serial try: port = hil_util.get_serial_dev(board['flasher']['uid'], None, None, 0) diff --git a/test/hil/helper/hil_util.py b/test/hil/helper/hil_util.py index 03d01270f..f279cfa77 100644 --- a/test/hil/helper/hil_util.py +++ b/test/hil/helper/hil_util.py @@ -1,7 +1,8 @@ #!/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 +# data every other module needs. Stays stdlib-only; its one local dependency is +# tools/rtt.py (the RTT console, loaded by path below) -- everything # else imports this, including the unit tests on GitHub's bare runner; never import them # from here. Callers set the module global `verbose`. @@ -499,6 +500,29 @@ def run_alongside(argv: list, work, timeout: int) -> subprocess.CompletedProcess return _reap() +# The RTT console implementation lives in tools/rtt.py (importable classes + CLI, +# stdlib-only, harness-critical — see its module docstring). Loaded by file path so +# no sys.path entry for tools/ can shadow other imports; re-exported here so the +# harness keeps addressing hil_util.JlinkRtt. +import importlib.util as _ilu + +_rtt_path = TINYUSB_ROOT / 'tools' / 'rtt.py' +if not _rtt_path.exists(): + # name the real cause: a bare FileNotFoundError out of an exec_module here reads + # as a harness bug, when the actual problem is an incompletely staged tree + raise ImportError(f'{_rtt_path} is missing — the RTT console lives there and the ' + f'harness depends on it; stage it alongside test/hil (hil_ci.sh does)') +_rtt_spec = _ilu.spec_from_file_location('tinyusb_tools_rtt', _rtt_path) +_rtt = _ilu.module_from_spec(_rtt_spec) +sys.modules[_rtt_spec.name] = _rtt # registered: RttError must be picklable across the fork Pool +_rtt_spec.loader.exec_module(_rtt) +JlinkRtt = _rtt.JlinkRtt +OpenocdRtt = _rtt.OpenocdRtt +RttError = _rtt.RttError +RTT_BANNER_RE = _rtt.RTT_BANNER_RE +strip_banner = _rtt.strip_banner + + 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 diff --git a/test/hil/hil_ci.sh b/test/hil/hil_ci.sh index daa787242..43ede5795 100644 --- a/test/hil/hil_ci.sh +++ b/test/hil/hil_ci.sh @@ -288,6 +288,9 @@ scp -q "$ROOT_DIR/test/hil/helper/__init__.py" \ "$ROOT_DIR/test/hil/helper/hil_lock.py" \ "$ROOT_DIR/test/hil/helper/hil_report.py" \ "$REMOTE:$REMOTE_DIR/test/hil/helper/" +# the rtt console/capture tool (rtt skill), harness-critical: hil_util imports it +ssh "$REMOTE" mkdir -p "$REMOTE_DIR/tools" +scp -q "$ROOT_DIR/tools/rtt.py" "$REMOTE:$REMOTE_DIR/tools/" # Copy only firmware binaries (elf/bin/hex) plus esptool metadata # (config.env + flash_args needed by the esptool flasher), preserving structure diff --git a/test/hil/hil_test.py b/test/hil/hil_test.py index e32998420..233627ec7 100755 --- a/test/hil/hil_test.py +++ b/test/hil/hil_test.py @@ -206,6 +206,7 @@ class Board(TypedDict): # needs one carries a single variant named after itself (metro_m4_express / # MAX3421_HOST=1), which is exactly what the `or [...]` default below synthesises variant: NotRequired[list[VariantCfg]] + logger: NotRequired[str] # "rtt": console = the debug probe's RTT channel 0, not a VCOM (rtt skill) toolchain: NotRequired[str] # CI build bucket override, e.g. "riscv-gcc" (consumed by hil_ci_set_matrix.py) @@ -292,6 +293,25 @@ def open_serial_dev(port: str): return ser +def open_board_console(board: Board): + """The board's log console: its probe's VCOM, or RTT when the probe has none. + + Both ends expose the same read/in_waiting/write/close surface, so the tests read one + the same way they read the other.""" + if board.get('logger') == 'rtt': + # JlinkRtt speaks JLinkExe only; an openocd/stlink flasher would yield + # `-device ''` and fail 15 s later with a misleading port error. The OpenOCD + # RTT route is validated manually on native probes but has no harness backend + # yet (rtt skill; followup doc) — and never point it at ea4088's LPC-Link2 + # (measured: knocks that probe off USB; other J-Link-OB probes untested) + assert board['flasher']['name'].lower() == 'jlink', \ + f'{board["name"]}: "logger": "rtt" needs a jlink flasher, not {board["flasher"]["name"]}' + return hil_util.JlinkRtt(board) + ser = open_serial_dev(hil_util.get_serial_dev(board['flasher']["uid"], None, None, 0)) + ser.timeout = 0.1 + return ser + + def serial_write_all(ser: serial.Serial, data: bytes): # write_timeout is a deadline for the whole call. A timeout means the device stopped # draining, and it is fatal: pyserial loses the partial-write count on raise, so @@ -300,8 +320,19 @@ def serial_write_all(ser: serial.Serial, data: bytes): ser.write(data) except serial.SerialTimeoutException: raise AssertionError(f'Serial write timeout after {SERIAL_WRITE_TIMEOUT:.1f}s') + except hil_util.RttError as e: + # the RTT console's failure contract (stall/closed/peer death): same + # drain-stopped meaning as the serial timeout -- a test failure, not a harness + # crash. Deliberately NOT bare RuntimeError: NotImplementedError and CPython's + # own 'dictionary changed size during iteration' are RuntimeErrors too, and a + # harness bug must not be reported as this board misbehaving. + raise AssertionError(f'Console write failed: {e}') +# J-Link Commander's telnet greeting: never target output (defined with the console +# in tools/rtt.py; hil_pool_check strips it through the same object) +RTT_BANNER_RE = hil_util.RTT_BANNER_RE + LP_OPEN_TIMEOUT = 5 # bound on opening the printer lp node; see test_device_printer_to_cdc # Runs under hil_util.run_alongside as `python3 -c`. Inline rather than a file so hil_ci.sh's # staging list does not need another entry to keep the rig working. @@ -508,34 +539,53 @@ def test_host_device_info(board): flasher = board['flasher'] declared_devs = [f'{d["vid_pid"]}_{d["serial"]}' for d in board['tests']['dev_attached']] - port = hil_util.get_serial_dev(flasher["uid"], None, None, 0) - ser = open_serial_dev(port) - ser.timeout = 0.1 - - # reset device since we can miss the first line - ret = getattr(hil_flash, f'reset_{flasher["name"].lower()}')(board) - assert ret.returncode == 0, 'Failed to reset device' + if board.get('logger') == 'rtt': + # The RTT console owns the probe, so reset BEFORE opening it (Commander then + # delivers the buffered boot burst). Unconditional, not only under --skip-flash: + # a previous run's console drained the ring, and the enumeration lines print + # only once — without this a re-run on unchanged firmware reads an empty ring. + ret = getattr(hil_flash, f'reset_{flasher["name"].lower()}')(board) + assert ret.returncode == 0, 'Failed to reset device' + ser = open_board_console(board) + try: + if board.get('logger') != 'rtt': + # reset device since we can miss the first line; on the VCOM the console + # survives the reset, so resetting after open catches the boot banner. + ret = getattr(hil_flash, f'reset_{flasher["name"].lower()}')(board) + assert ret.returncode == 0, 'Failed to reset device' - data = b'' - timeout = enum_timeout() - while timeout > 0: - new_data = ser.read(ser.in_waiting or 1) - if new_data: - data += new_data - enum_dev_sn = [] - for l in data.decode('utf-8', errors='ignore').splitlines(): - vid_pid_sn = re.search(r'ID ([0-9a-fA-F]+):([0-9a-fA-F]+) SN (\w+)', l) - if vid_pid_sn: - enum_dev_sn.append(f'{vid_pid_sn.group(1)}_{vid_pid_sn.group(2)}_{vid_pid_sn.group(3)}') - if set(declared_devs).issubset(set(enum_dev_sn)): - break - time.sleep(0.1) - timeout -= 0.1 - ser.close() + data = b'' + timeout = enum_timeout() + while timeout > 0: + # infra death is not a board failure: without this a dead JLinkExe/probe + # would burn the whole timeout and report as 'No data from device' + assert not getattr(ser, 'eof', False), \ + 'RTT console died (its server exited or the probe dropped off USB)' + new_data = ser.read(ser.in_waiting or 1) + if new_data: + data += new_data + enum_dev_sn = [] + for l in data.decode('utf-8', errors='ignore').splitlines(): + vid_pid_sn = re.search(r'ID ([0-9a-fA-F]+):([0-9a-fA-F]+) SN (\w+)', l) + if vid_pid_sn: + enum_dev_sn.append(f'{vid_pid_sn.group(1)}_{vid_pid_sn.group(2)}_{vid_pid_sn.group(3)}') + if set(declared_devs).issubset(set(enum_dev_sn)): + break + time.sleep(0.1) + timeout -= 0.1 + finally: + ser.close() - if len(data) == 0: - assert False, 'No data from device' lines = data.decode('utf-8', errors='ignore').splitlines() + if board.get('logger') == 'rtt': + # JLinkExe's telnet banner is delivered at connect, whether or not it ever + # finds the control block, so len(data) alone cannot tell "board said nothing" + # from "console never attached to the ring" -- drop the banner first + target_lines = hil_util.strip_banner(data).splitlines() + assert target_lines, ('No data from device: the RTT console attached but the target ' + 'produced nothing -- firmware built without LOGGER=rtt, or SWD lost') + elif len(data) == 0: + assert False, 'No data from device' enum_dev_sn = [] for l in lines: @@ -1729,7 +1779,7 @@ def test_example(board: Board, variant: str, example: str) -> tuple[int, str, st def build_board(board: Board) -> tuple[str, int]: """Build firmware for this board via tools/build.py. - Honors board config's variant list (name, defines, flags). + Honors board config's variant list. Output goes to cmake-build/cmake-build-<variant>/ (tools/build.py layout). Unbounded on purpose: --build is a local convenience (no CI workflow passes it), so @@ -2337,6 +2387,56 @@ def main() -> None: config_boards = [e for e in config['boards'] if e['name'] in boards] config_boards = [e for e in config_boards if e['flasher']['name'] not in args.exclude_flasher and (not args.flasher or e['flasher']['name'] in args.flasher)] + + # fail rtt misconfigurations before the first flash cycle -- but only for boards + # this run actually touches: one bad roster entry must not abort other runs' subsets + def _rtt_config_abort(msg: str): + # loud AND leaving evidence, like the no-boards branch below: exiting with no + # report at all lets the PR comment keep the previous push's stale table + print(f'ERROR: {msg}', flush=True) + rd = Path(os.environ.get('HIL_REPORT_DIR', '.')) + hil_report.mark_report_no_boards(rd, f'config error: {msg}', fresh=not args.accumulate) + sys.exit(1) + + bad_logger = [e['name'] for e in config_boards if e.get('logger') not in (None, 'rtt')] + if bad_logger: + # only the exact string activates RTT handling; anything else would silently + # mean VCOM and reproduce the misleading 'No serial device found' failure + _rtt_config_abort(f'unknown "logger" value (only "rtt" is supported): {", ".join(bad_logger)}') + bad_rtt = [e['name'] for e in config_boards + if e.get('logger') == 'rtt' and e['flasher']['name'].lower() != 'jlink'] + if bad_rtt: + # JlinkRtt speaks JLinkExe only (the OpenOCD RTT route is manual — rtt skill) + _rtt_config_abort(f'"logger": "rtt" needs a jlink flasher: {", ".join(bad_rtt)}') + rtt_no_logger_def = [e['name'] for e in config_boards + if e.get('logger') == 'rtt' + and any('LOGGER=rtt' not in (v.get('defines') or []) + for v in (e.get('variant') or [{}]))] + if rtt_no_logger_def: + # a prebuilt cmake-build-<board> configured with -DLOGGER=rtt is a legitimate + # build path the roster need not describe, so warn there -- but when this run is + # responsible for the firmware (--build, or CI where the hil-build job compiled + # the artifact from these same defines) the flashed image is UART-logger and every + # test times out as 'the target produced nothing'. An always-on define is + # expressed as a single self-named variant (see the Board comment). + msg = (f'"logger": "rtt" board has a variant without LOGGER=rtt in its defines ' + f'({", ".join(rtt_no_logger_def)})') + if args.build or os.environ.get('GITHUB_ACTIONS'): + _rtt_config_abort(f'{msg} -- the firmware built for this run cannot serve the ' + f'configured RTT console') + print(f'warning: {msg} -- fine for prebuilt example sets, wrong for --build/CI ' + f'builds', flush=True) + rtt_fixture = [e['name'] for e in config_boards + if e.get('logger') == 'rtt' + and any(d.get('is_cdc') or d.get('is_msc') + for d in e.get('tests', {}).get('dev_attached', []))] + if rtt_fixture: + # interim guard, removed when the followup lands: cdc_msc_hid/msc_file_explorer + # still open the flasher VCOM directly and would die mid-run on an rtt board + _rtt_config_abort(f'"logger": "rtt" boards cannot carry is_cdc/is_msc fixtures yet ' + f'(host cdc/msc tests bypass the RTT console — see ' + f'the rtt harness-adoption doc in docs/superpowers/followup/): {", ".join(rtt_fixture)}') + if not config_boards: # same reason the unknown -b board exits 1: 'No tests were run.' with rc 0 reads as # a green HIL leg, so a roster edit emptying a leg's filter stops testing silently diff --git a/test/hil/test/test_ci_select.py b/test/hil/test/test_ci_select.py index ace230246..22fbde17b 100644 --- a/test/hil/test/test_ci_select.py +++ b/test/hil/test/test_ci_select.py @@ -499,6 +499,8 @@ class TestPortAndCoreRoleUseExtras(unittest.TestCase): self.assertFalse(s['full']) for board in boards: tests = s['boards'][board] + if tests == 'all': + continue # a board whose whole allowed set is selected collapses to 'all' self.assertIn('device/hid_composite_freertos', tests) self.assertIn('device/cdc_msc_freertos', tests) self.assertIn('device/audio_test_freertos', tests) @@ -510,6 +512,8 @@ class TestPortAndCoreRoleUseExtras(unittest.TestCase): self.assertFalse(s['full']) for board in boards: tests = s['boards'][board] + if tests == 'all': + continue # a board whose whole allowed set is selected collapses to 'all' self.assertIn('device/hid_composite_freertos', tests) self.assertIn('device/cdc_msc_freertos', tests) self.assertIn('device/audio_test_freertos', tests) @@ -987,6 +991,7 @@ class TestTheHarnessTestsAreNotTheHarness(unittest.TestCase): 'test/hil/test/test_hil_bounded.py', 'test/hil/test/test_hil_health.py', 'test/hil/test/test_hil_report.py', + 'test/hil/test/test_hil_rtt.py', 'test/hil/test/test_hil_util.py', ], 'test/hil/test/ gained or lost a file; it is carved out of rule 2, so confirm ' 'the rig still does not read anything in there before updating this list') diff --git a/test/hil/test/test_hil_rtt.py b/test/hil/test/test_hil_rtt.py new file mode 100644 index 000000000..3a07f13ec --- /dev/null +++ b/test/hil/test/test_hil_rtt.py @@ -0,0 +1,506 @@ +#!/usr/bin/env python3 +# SPDX-License-Identifier: MIT +# Unit tests for hil_util.JlinkRtt and the rtt.py CLI against a fake JLinkExe +# on PATH — real subprocesses and sockets, no hardware, stdlib only, so the pre-commit +# hil-test hook can run this on GitHub's bare runner. Run directly: +# python3 test/hil/test/test_hil_rtt.py +import os +import subprocess +import sys +import tempfile +import time +import unittest +from contextlib import suppress as contextlib_suppress +from pathlib import Path + +# the module under test lives in the parent dir's helper/ package +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +from helper import hil_util + +CLI = Path(__file__).resolve().parents[3] / 'tools' / 'rtt.py' + +# Serves -RTTTelnetPort like J-Link Commander: greets, echoes input uppercased, exits on +# stdin 'exit' (JlinkRtt.close()'s contract). FAKE_JLINK_MODE=die_after_greet sends the +# greeting then drops the connection and exits — the probe-unplug/crash case; +# FAKE_JLINK_MODE=tick also streams a line every 50 ms — the continuous-capture case. +FAKE_JLINK = '''#!/usr/bin/env python3 +import os, socket, sys, threading, time +port = int(sys.argv[sys.argv.index('-RTTTelnetPort') + 1]) +srv = socket.socket(); srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) +srv.bind(('127.0.0.1', port)); srv.listen(1) +mode = os.environ.get('FAKE_JLINK_MODE', '') +def serve(): + conn, _ = srv.accept() + # the real server sends its banner AT CONNECT, before the control block is + # found — target data only flows later; the CLI's -i gate must not release + # on the banner + conn.sendall(b'SEGGER J-Link fake - Real time terminal output\\r\\n' + b'J-Link FakeProbe V1.0, SN=000\\r\\nProcess: JLinkExe\\r\\n') + if mode == 'banner_only': + while True: + if not conn.recv(4096): os._exit(0) + if mode == 'rst': + import struct + conn.recv(4096) # wait for the client to speak, then reset the connection + conn.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, struct.pack('ii', 1, 0)) + conn.close(); os._exit(0) + if mode == 'late_cb': + # models JLinkExe before it finds the control block: client bytes sent in + # this window are silently dropped, output starts only after the "attach" + end = time.time() + 1.0 + conn.setblocking(False) + while time.time() < end: + try: + conn.recv(4096) # discard early input like the real server + except OSError: + pass + time.sleep(0.05) + conn.setblocking(True) + conn.sendall(b'hello from target\\r\\n') + if mode == 'die_after_greet': + conn.close(); os._exit(0) + if mode == 'tick': + def tick(): + try: + while True: + time.sleep(0.05); conn.sendall(b'tick\\r\\n') + except OSError: + pass + threading.Thread(target=tick, daemon=True).start() + while True: + d = conn.recv(4096) + if not d: return + conn.sendall(d.upper()) +threading.Thread(target=serve, daemon=True).start() +for line in sys.stdin: + if line.strip() == 'exit': break +''' + +BOARD = {'flasher': {'uid': '000', 'args': '-device FAKE'}} + + [email protected](os.name == 'nt', 'POSIX PATH/exec semantics') +class JlinkRttFakeProbe(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls._dir = tempfile.TemporaryDirectory() + fake = Path(cls._dir.name) / 'JLinkExe' + fake.write_text(FAKE_JLINK) + fake.chmod(0o755) + cls._path = f'{cls._dir.name}{os.pathsep}{os.environ["PATH"]}' + + @classmethod + def tearDownClass(cls): + cls._dir.cleanup() + + def _fake_path(self): + # register the restore BEFORE mutating, then prepend the fake tool dir + self.addCleanup(os.environ.__setitem__, 'PATH', os.environ['PATH']) + os.environ['PATH'] = self._path + + def _console(self, mode=''): + self._fake_path() + if mode: + os.environ['FAKE_JLINK_MODE'] = mode + self.addCleanup(os.environ.pop, 'FAKE_JLINK_MODE', None) + con = hil_util.JlinkRtt(BOARD, timeout=0.1) + self.addCleanup(con.close) + return con + + def _read_until(self, con, want, timeout=3): + out = b'' + end = time.monotonic() + timeout + while want not in out and time.monotonic() < end: + out += con.read(con.in_waiting or 1) + return out + + def test_read_and_echo_write(self): + con = self._console() + self.assertIn(b'hello from target', self._read_until(con, b'hello from target')) + self.assertEqual(con.write(b'ping'), 4) + self.assertIn(b'PING', self._read_until(con, b'PING')) + + def test_eof_latched_when_server_dies(self): + con = self._console(mode='die_after_greet') + self._read_until(con, b'hello from target') + end = time.monotonic() + 3 + while not con.eof and time.monotonic() < end: + time.sleep(0.05) + self.assertTrue(con.eof) # dead server is detected, not spun on + t0 = time.monotonic() + self.assertEqual(con.read(64), b'') # empty, paced like a serial timeout + elapsed = time.monotonic() - t0 + self.assertLess(elapsed, 0.5) # bounded by the 0.1 s timeout, not hung + self.assertGreater(elapsed, 0.02) # ...but not a busy-spin fast return + con.timeout = None # pyserial's block-forever mode must + t0 = time.monotonic() # ALSO pace (0.1 s default), not spin + self.assertEqual(con.read(64), b'') + elapsed = time.monotonic() - t0 + self.assertLess(elapsed, 0.5) + self.assertGreater(elapsed, 0.02) + con.timeout = 0.1 + + def test_reset_input_buffer(self): + con = self._console() + self._read_until(con, b'hello from target') + con.write(b'x') + time.sleep(0.3) + con.reset_input_buffer() + self.assertEqual(con.in_waiting, 0) + + def test_write_after_close_raises_runtimeerror(self): + con = self._console() + con.close() + with self.assertRaises(RuntimeError): + con.write(b'x') + + def test_write_after_server_death_raises(self): + # TCP accepts one send after peer death — write() must refuse instead of + # "succeeding" into the void + con = self._console(mode='die_after_greet') + self._read_until(con, b'hello from target') + end = time.monotonic() + 3 + while not con.eof and time.monotonic() < end: + time.sleep(0.05) + with self.assertRaises(RuntimeError): + con.write(b'ping') + + def test_read_after_close_raises_runtimeerror(self): + con = self._console() + self._read_until(con, b'hello from target') + con.close() + with self.assertRaises(RuntimeError): + con.read(1) + + def test_missing_jlinkexe_raises_runtimeerror(self): + self._fake_path() + os.environ['PATH'] = self._dir.name # no python3 either, but JLinkExe fails first + os.rename(f'{self._dir.name}/JLinkExe', f'{self._dir.name}/JLinkExe.off') + self.addCleanup(os.rename, f'{self._dir.name}/JLinkExe.off', f'{self._dir.name}/JLinkExe') + with self.assertRaises(RuntimeError): + hil_util.JlinkRtt(BOARD, timeout=0.1) + + def test_close_reaps_the_server(self): + con = self._console() + proc = con._proc + con.close() + self.assertIsNotNone(proc.poll()) # no zombie, no probe held + + def test_cli_exits_when_server_dies(self): + # --seconds 0 must end on server EOF (rc 1), not hang forever + env = dict(os.environ, PATH=self._path, FAKE_JLINK_MODE='die_after_greet') + r = subprocess.run([sys.executable, str(CLI), + '--backend', 'jlink', '--probe', '000', '--device', 'FAKE', '--seconds', '0'], + env=env, capture_output=True, timeout=20) + self.assertEqual(r.returncode, 1) + self.assertIn(b'hello from target', r.stdout) + self.assertIn(b'server closed', r.stderr) + + def test_peer_reset_latches_eof(self): + # a killed server closes with RST when bytes are unread; the read side must + # LATCH eof (so the harness's `assert not ser.eof` triage fires) and never + # leak ConnectionResetError/ValueError to in_waiting/eof callers + con = self._console(mode='rst') + # rst mode sends only the banner (it RSTs on first input) -- wait for the + # banner tail, not target output that never comes + self._read_until(con, b'Process: JLinkExe') + con.write(b'x') # fake resets the connection on input + end = time.monotonic() + 3 + try: + while not con.eof and time.monotonic() < end: + con.in_waiting # must not raise across the RST + time.sleep(0.05) + except Exception as e: # noqa: BLE001 - the regression this guards + self.fail(f'{type(e).__name__} escaped the latch-only contract: {e}') + self.assertTrue(con.eof) + with self.assertRaises(hil_util.RttError): + con.write(b'y') # dead server refuses writes + + def test_write_timeout_env_rejects_inf(self): + # hil_util's twin rejects inf for the same reason: an unbounded write is what + # this knob exists to bound + import importlib.util as ilu + from pathlib import Path as _P + spec = ilu.spec_from_file_location('rtt_env_probe', _P(CLI)) + mod = ilu.module_from_spec(spec) + old = os.environ.get('HIL_SERIAL_WRITE_TIMEOUT') + os.environ['HIL_SERIAL_WRITE_TIMEOUT'] = 'inf' + self.addCleanup(lambda: os.environ.__setitem__('HIL_SERIAL_WRITE_TIMEOUT', old) + if old is not None else os.environ.pop('HIL_SERIAL_WRITE_TIMEOUT', None)) + spec.loader.exec_module(mod) + self.assertEqual(mod.RTT_WRITE_TIMEOUT, 10) + + def test_cli_rejects_bad_seconds_and_jlink_channel(self): + def run(*a): + return subprocess.run([sys.executable, str(CLI), *a], capture_output=True, timeout=15) + for bad in ('-5', 'nan'): + r = run('--backend', 'jlink', '--probe', '000', '--device', 'FAKE', '--seconds', bad) + self.assertEqual(r.returncode, 2, f'--seconds {bad} was accepted') + # the jlink telnet route serves channel 0 only; asking for another is an error, + # not silence (--dump can read any ring, so it stays allowed there) + r = run('--backend', 'jlink', '--probe', '000', '--device', 'FAKE', '--channel', '1') + self.assertEqual(r.returncode, 2) + self.assertIn(b'channel 0 only', r.stderr) + # a negative index would walk backwards off aUp[] (dump route included) + r = run('--backend', 'jlink', '--probe', '000', '--device', 'FAKE', '--channel', '-1') + self.assertEqual(r.returncode, 2) + self.assertIn(b'>= 0', r.stderr) + + def test_pyserial_surface_contracts(self): + con = self._console() + self._read_until(con, b'hello from target') + con.write(b'abcdef') + self._read_until(con, b'ABC') # echo queued + before = con.in_waiting + self.assertEqual(con.read(0), b'') # pyserial: consumes nothing + self.assertEqual(con.read(-1), b'') # never hand over/destroy bytes + self.assertEqual(con.in_waiting, before) + con.timeout = None # pyserial: block until satisfied + con.write(b'xy') # fresh echo guarantees the read returns + self.assertEqual(len(con.read(2)), 2) + con.timeout = 0.1 + con.close() + with self.assertRaises(hil_util.RttError): + con.in_waiting # closed console reports closed, not healthy + self.assertTrue(con.eof) + + def test_context_manager_closes(self): + self._fake_path() + with hil_util.JlinkRtt(BOARD, timeout=0.1) as con: + proc = con._proc + self.assertIsNotNone(proc.poll()) # __exit__ released the probe + + def test_staging_and_banner_coupling(self): + # tripwires for couplings no import-walk can see: + # (a) hil_ci.sh must stage tools/rtt.py -- hil_util exec_module's it, so an + # unstaged rig tree kills every harness import + hil_ci = (Path(__file__).resolve().parents[1] / 'hil_ci.sh').read_text() + self.assertIn('tools/rtt.py', hil_ci) + # (b) the shared RTT banner filter must drop ALL THREE J-Link banner lines, + # including the middle one, which is the PROBE MODEL string and in + # libjlinkarm carries no 'SEGGER ' prefix (J-Link OH3, J-Trace H9...) + banner_re = hil_util.RTT_BANNER_RE + for line in ('SEGGER J-Link V9.66 - Real time terminal output', + 'SEGGER J-Link LPC-Link 2 V1.0, SN=611000000', + 'J-Link OH3 V1.0, SN=123456789', + 'J-Trace H9 V2.0, SN=123456789002', + 'Process: JLinkExe'): + self.assertTrue(banner_re.match(line), f'banner line not filtered: {line!r}') + for line in ('Hello from TinyUSB', 'USBD init on controller 0', + 'ID 1a86:8010 SN 7FD88F0604B5', 'echo:p'): + self.assertFalse(banner_re.match(line), f'target line wrongly filtered: {line!r}') + + def test_pool_check_dead_rtt_board_is_not_alive(self): + # JLinkExe's banner alone must not score a dead board 'alive': pool_check's + # rtt aliveness judges only target bytes (the bug: unfiltered, the banner + # made `not boardtest_output(data)` true on the first poll) + sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + from helper import hil_pool_check + # a dead board burns the whole poll window; the verdict is the same at 0.5 s + self.addCleanup(setattr, hil_pool_check, 'SERIAL_WAIT', hil_pool_check.SERIAL_WAIT) + hil_pool_check.SERIAL_WAIT = 0.5 + self._fake_path() + os.environ['FAKE_JLINK_MODE'] = 'banner_only' + self.addCleanup(os.environ.pop, 'FAKE_JLINK_MODE', None) + board = dict(BOARD, name='deadboard', logger='rtt') + got = hil_pool_check.check_host_serial(board, do_reset=False, want_hello=True) + self.assertEqual(got, b'') # dead, not "alive on banner" + + def test_cli_arg_contract(self): + # --backend is explicit (no default); vid-pid is openocd-only; the openocd + # backend accepts --addr instead of --elf and --vid-pid instead of --probe + def run(*a, inp=b''): + return subprocess.run([sys.executable, str(CLI), *a], + input=inp, capture_output=True, timeout=15) + r = run('--probe', '000', '--device', 'FAKE') # no --backend + self.assertEqual(r.returncode, 2) + self.assertIn(b'--backend', r.stderr) + r = run('--backend', 'jlink', '--probe', '000', '--device', 'FAKE', '--vid-pid', '0x1 0x2') + self.assertEqual(r.returncode, 2) # vid-pid is openocd-only + r = run('--backend', 'openocd', '--cfg', '-f x.cfg', '--addr', '0x20000000') + self.assertEqual(r.returncode, 2) # needs --probe or --vid-pid + self.assertIn(b'vid-pid', r.stderr) + r = run('--backend', 'openocd', '--probe', '000', '--cfg', '-f x.cfg', '--addr', 'nothex') + self.assertEqual(r.returncode, 2) + self.assertIn(b'hex', r.stderr) + + def test_cli_interactive_echo(self): + env = dict(os.environ, PATH=self._path) + r = subprocess.run([sys.executable, str(CLI), + '--backend', 'jlink', '--probe', '000', '--device', 'FAKE', '--seconds', '2', '-i'], + env=env, input=b'hi', capture_output=True, timeout=20) + self.assertEqual(r.returncode, 0) + self.assertIn(b'HI', r.stdout) # bytes forwarded without needing a newline + self.assertNotIn(b'never forwarded', r.stderr) # forwarding happened: no false alarm + + def test_cli_interactive_input_held_until_output(self): + # input piped at process start must survive the server's control-block hunt + # (the real JLinkExe drops client bytes until the block is found — measured + # on the rig: instant 'ping' lost, delayed 'ping' echoed) + env = dict(os.environ, PATH=self._path, FAKE_JLINK_MODE='late_cb') + r = subprocess.run([sys.executable, str(CLI), + '--backend', 'jlink', '--probe', '000', '--device', 'FAKE', '--seconds', '3', '-i'], + env=env, input=b'hi', capture_output=True, timeout=25) + self.assertEqual(r.returncode, 0) + self.assertIn(b'HI', r.stdout) + + def test_cli_interactive_no_input_diagnostic(self): + # -i with stdin closed immediately: the diagnostic must say stdin was never + # forwarded (true), keyed on actual forwarding -- not on the attach gate, + # which releases after 5 s and forwards anyway on longer runs + env = dict(os.environ, PATH=self._path, FAKE_JLINK_MODE='banner_only') + r = subprocess.run([sys.executable, str(CLI), + '--backend', 'jlink', '--probe', '000', '--device', 'FAKE', '--seconds', '1', '-i'], + env=env, input=b'', capture_output=True, timeout=20) + self.assertEqual(r.returncode, 0) + self.assertIn(b'never forwarded', r.stderr) + self.assertIn(b'no target output', r.stderr) + + def test_cli_downstream_pipe_close(self): + # a real `rtt.py | head`-style consumer: close the read end mid-stream + # and the CLI must exit 0 via its BrokenPipe path, not traceback (this test + # fails if the handler is removed — subprocess.run capture can't cover it) + env = dict(os.environ, PATH=self._path, FAKE_JLINK_MODE='tick') + p = subprocess.Popen([sys.executable, str(CLI), + '--backend', 'jlink', '--probe', '000', '--device', 'FAKE', '--seconds', '8'], + env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE) + p.stdout.read(10) # let it stream a little + p.stdout.close() # downstream hangs up + rc = p.wait(timeout=20) + err = p.stderr.read() + p.stderr.close() + self.assertEqual(rc, 0, err) + self.assertNotIn(b'Traceback', err) + + def test_cli_feeder_races_shutdown(self): + # a feeder still writing when --seconds expires must not crash the CLI + # (pump thread vs close() race: historically tracebacks and SIGABRT rc 134) + env = dict(os.environ, PATH=self._path) + for _ in range(3): + p = subprocess.Popen([sys.executable, str(CLI), + '--backend', 'jlink', '--probe', '000', '--device', 'FAKE', '--seconds', '1', '-i'], + env=env, stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, + stderr=subprocess.PIPE) + try: + while True: + p.stdin.write(b'hi\n') + p.stdin.flush() + time.sleep(0.01) + except (BrokenPipeError, OSError): + pass + rc = p.wait(timeout=20) + err = p.stderr.read() + p.stderr.close() + with contextlib_suppress(OSError, ValueError): + p.stdin.close() + self.assertEqual(rc, 0, err) + self.assertNotIn(b'Exception in thread', err) + + + +class StripBanner(unittest.TestCase): + # both harness consumers (device_info verdict, pool_check aliveness) judge + # target-aliveness through this ONE filter -- pin its shape here + def test_drops_banner_keeps_target(self): + raw = (b'SEGGER J-Link V9.66 - Real time terminal output\r\n' + b'J-Link OH3 V1.0, SN=123456789\r\nProcess: JLinkExe\r\n' + b'Hello from TinyUSB\r\n') + self.assertEqual(hil_util.strip_banner(raw), b'Hello from TinyUSB') + + def test_complete_only_drops_split_banner_fragment(self): + # a poll loop can catch the banner mid-line at a read boundary; the + # fragment must not defeat the prefix regex and score as target output + frag = b'SEGGER J-Link V9.66 - Real time terminal output\r\nProce' + self.assertEqual(hil_util.strip_banner(frag, complete_only=True), b'') + # the final verdict keeps a genuine unterminated target tail + self.assertEqual(hil_util.strip_banner(b'tud_task\r\nrunn'), b'tud_task\nrunn') + self.assertEqual(hil_util.strip_banner(b'', complete_only=True), b'') + + +# Serves like `openocd ... -c "rtt server start PORT CH"`: parses the port from its +# single shell-quoted command line, greets, echoes uppercased. No banner (matches the +# real openocd rtt server, which sends target data only). +FAKE_OPENOCD = '''#!/usr/bin/env python3 +import os, re, socket, sys, threading, time +if os.environ.get('FAKE_OPENOCD_ARGV'): + open(os.environ['FAKE_OPENOCD_ARGV'], 'w').write(' '.join(sys.argv)) +port = int(re.search(r'rtt server start (\\d+)', ' '.join(sys.argv)).group(1)) +srv = socket.socket(); srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) +srv.bind(('127.0.0.1', port)); srv.listen(1) +conn, _ = srv.accept() +conn.sendall(b'hello from target\\r\\n') +while True: + d = conn.recv(4096) + if not d: break + conn.sendall(d.upper()) +''' + + [email protected](os.name == 'nt', 'POSIX PATH/exec semantics') +class OpenocdRttFakeProbe(unittest.TestCase): + """The openocd-backend class shares its whole read/write/eof contract with + JlinkRtt via the base class (covered above); this exercises the parts it owns: + spawn/connect, echo round-trip, teardown.""" + + @classmethod + def setUpClass(cls): + cls._dir = tempfile.TemporaryDirectory() + fake = Path(cls._dir.name) / 'openocd' + fake.write_text(FAKE_OPENOCD) + fake.chmod(0o755) + cls._path = f'{cls._dir.name}{os.pathsep}{os.environ["PATH"]}' + + @classmethod + def tearDownClass(cls): + cls._dir.cleanup() + + def _fake_path(self): + self.addCleanup(os.environ.__setitem__, 'PATH', os.environ['PATH']) + os.environ['PATH'] = self._path + + def test_reset_before_attach_shapes_the_command(self): + # SystemView-style consumers need the server draining WHEN the target boots + # (its Init record is emitted once); the opt-in flag must put `reset run` + # between init and rtt setup, and must not appear otherwise + self._fake_path() + argv_file = os.path.join(self._dir.name, 'argv.txt') + os.environ['FAKE_OPENOCD_ARGV'] = argv_file + self.addCleanup(os.environ.pop, 'FAKE_OPENOCD_ARGV', None) + for flag, want in ((True, True), (False, False)): + con = hil_util.OpenocdRtt('-f fake.cfg', 0x20000000, 1, serial_no='000', + reset_before_attach=flag) + try: + argv = Path(argv_file).read_text() + finally: + con.close() + self.assertEqual('reset run' in argv, want, argv) + if want: # ordering is the whole point: reset, settle, THEN attach + self.assertLess(argv.index('reset run'), argv.index('rtt setup'), argv) + self.assertIn('sleep 2000', argv) + self.assertIn('rtt server start', argv) + self.assertTrue(argv.rstrip().endswith('1'), argv) # channel threaded through + + def test_openocd_route_echo_and_teardown(self): + self._fake_path() + con = hil_util.OpenocdRtt('-f fake.cfg', 0x20000000, 0, + serial_no='000', vid_pid='0x1234 0x5678') + self.addCleanup(con.close) + out = b'' + end = time.monotonic() + 3 + while b'hello from target' not in out and time.monotonic() < end: + out += con.read(con.in_waiting or 1) + self.assertIn(b'hello from target', out) + con.write(b'ping') + end = time.monotonic() + 3 + while b'PING' not in out and time.monotonic() < end: + out += con.read(con.in_waiting or 1) + self.assertIn(b'PING', out) + proc = con._proc + con.close() + self.assertIsNotNone(proc.poll()) # no zombie, no probe held + with self.assertRaises(RuntimeError): + con.write(b'x') # same post-close contract as JlinkRtt + + +if __name__ == '__main__': + unittest.main() diff --git a/test/hil/test/test_hil_util.py b/test/hil/test/test_hil_util.py index e06d2ba8b..1a283bda7 100644 --- a/test/hil/test/test_hil_util.py +++ b/test/hil/test/test_hil_util.py @@ -145,9 +145,13 @@ class BottomLayer(unittest.TestCase): # hil_pool_check included: test_hil_util_is_a_single_module_instance imports it # on the bare runner, and its `import serial` is function-local for exactly # this reason -- hoisting it must fail HERE, not on every PR's pre-commit CI + # ../../tools/rtt: hil_util exec_module's it at import (helper/hil_util.py's + # loader block), so a non-stdlib import THERE kills ci_select on the bare + # runner just as surely -- and the spec_from_file_location call is invisible to + # the ast.Import walk below, which is why it must be listed explicitly for mod in ('helper/hil_util', 'hil_flash', '../../tools/ci_select', 'helper/hil_health', 'helper/hil_lock', 'helper/hil_pool_check', - '../../tools/build', '../../tools/build_utils'): + '../../tools/build', '../../tools/build_utils', '../../tools/rtt'): tree = ast.parse((hil_dir / f'{mod}.py').read_text()) # module level only: a deferred import inside a function cannot break # importability (hil_pool_check keeps `import serial` function-local |