diff --git a/.claude/skills/hil-contract/scripts/hil_remote.py b/.claude/skills/hil-contract/scripts/hil_remote.py index 8c6c70faf9..7c9af900b2 100644 --- a/.claude/skills/hil-contract/scripts/hil_remote.py +++ b/.claude/skills/hil-contract/scripts/hil_remote.py @@ -75,6 +75,7 @@ def helper(name): 'test/hil/pymtp.py', 'test/hil/mtp_test.py', 'test/hil/mtp_raw.py', + 'test/hil/net_test.py', 'test/hil/sysview_ci.py', # hil_test.py imports it; --sysview-out itself is refused here 'test/hil/helper/__init__.py', 'test/hil/helper/agentrc.py', diff --git a/test/hil/helper/agentrc.py b/test/hil/helper/agentrc.py index 372dd52e14..bae28b70d6 100644 --- a/test/hil/helper/agentrc.py +++ b/test/hil/helper/agentrc.py @@ -44,6 +44,7 @@ 'skills/hil/scripts/hil_remote_core.py', 'skills/hil/scripts/hil_report_core.py', 'skills/hil/scripts/hil_report_notice.py', + 'skills/hil/scripts/hil_root.py', 'skills/hil/scripts/hil_tt.py', 'skills/hil/scripts/hil_usbtest.py', 'skills/hil/scripts/usb_observe.py', diff --git a/test/hil/helper/hil_util.py b/test/hil/helper/hil_util.py index e46f7c06fa..55a66f1f53 100644 --- a/test/hil/helper/hil_util.py +++ b/test/hil/helper/hil_util.py @@ -37,7 +37,7 @@ 'device/midi_test', 'device/mtp', 'device/usbtest', # 0525:a4a0 (Gadget Zero, usbtest's static id); 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 + 'device/net_lwip_webserver', ] dual_tests = [ diff --git a/test/hil/hil_test.py b/test/hil/hil_test.py index 7fb53727eb..517db8b9ca 100755 --- a/test/hil/hil_test.py +++ b/test/hil/hil_test.py @@ -23,12 +23,16 @@ # THE SOFTWARE. # Host setup (required: a missing tool fails its test rather than skipping it): -# - System packages: sudo apt install mtools libmtp9 libmtp-runtime alsa-utils iperf +# - System packages: sudo apt install mtools libmtp9 libmtp-runtime alsa-utils iproute2 iperf udev util-linux # mtools read_disk_file (device/cdc_msc, device/msc_dual_lun) # libmtp9 pymtp ctypes load (device/mtp); Debian 13 uses libmtp9t64 # libmtp-runtime mtp-probe and the completed-device /dev/libmtp-* marker # alsa-utils arecord (device/audio_test_freertos) -# iperf throughput tests (device/net_lwip_*) +# iproute2 isolated USB network test, run inside its own namespace +# iperf its host-to-device TCP rate, on builds with the lwiperf server +# util-linux unshare/nsenter for per-process user and network namespaces; the one root +# step, moving the interface in, is agentrc's hil_root netns (sudo) +# udev udevadm wait (systemd >= 251) for USB network initialization # openocd unified openocd from https://github.com/hathach/openocd (branch tinyusb) for wch, rp2040/rp2350, analog max32 # - device/usbtest: usbtest kernel module + testusb binary (kernel tools/usb/testusb.c) on PATH, # plus sudo for modprobe / sysfs writes @@ -1326,60 +1330,14 @@ def test_device_mtp(board): def test_device_net_lwip_webserver(board): - # MAC hard-coded in examples/device/net_lwip_webserver/src/main.c; Linux names the - # iface enx. Device IP 192.168.7.1, iperf2 TCP server on 5001 - # (INCLUDE_IPERF). - import socket - mac_no_colons = '0202846a9600' - iface = 'enx' + mac_no_colons - device_ip = '192.168.7.1' - iperf_port = 5001 - - # Wait for an IPv4 address in the device's subnet (it serves DHCP); 30s because USB - # enum + DHCP serve is slower on the CI HIL hardware than locally. - iface_timeout = 30 - deadline = time.monotonic() + iface_timeout - host_ip = None - while time.monotonic() < deadline: - ret = subprocess.run(['ip', '-o', '-4', 'addr', 'show', iface], - capture_output=True, text=True, timeout=2) - m = re.search(r'inet (192\.168\.7\.\d+)/', ret.stdout) if ret.returncode == 0 else None - if m: - host_ip = m.group(1) - break - time.sleep(0.5) - assert host_ip, f'USB net iface {iface} did not come up with 192.168.7.x within {iface_timeout}s' - - # Poll until the device accepts: the net stack and the iperf bind come up after DHCP. - deadline = time.monotonic() + enum_timeout() - last_err = None - while time.monotonic() < deadline: - try: - with socket.create_connection((device_ip, iperf_port), timeout=1): - last_err = None - break - except OSError as e: - last_err = e - time.sleep(0.3) - assert last_err is None, f'iperf TCP {device_ip}:{iperf_port} not accepting within {enum_timeout()}s: {last_err}' - - # 5-second iperf2 TCP test; -y C for stable parsing (final summary line is - # timestamp,src_ip,src_port,dst_ip,dst_port,id,interval,bytes,bps). - ret = hil_util.run_cmd(f'iperf -c {device_ip} -t 5 -y C', - timeout=30, split_stderr=True, quiet=True) - stderr = (ret.stderr or '').strip() - stdout = (ret.stdout or '').strip() - assert ret.returncode == 0, f'iperf rc={ret.returncode}: stderr={stderr!r} stdout={stdout!r}' - lines = [l for l in stdout.splitlines() if l] - assert lines, f'iperf produced no output (rc={ret.returncode}, stderr={stderr!r})' - try: - bps = int(lines[-1].split(',')[-1]) - except (ValueError, IndexError) as e: - raise AssertionError(f'could not parse iperf output: {lines[-1]!r} ({e})') - mbps = bps / 1e6 - print(f' iperf {mbps:5.1f} Mbps', end='') - - assert mbps >= 1.0, f'iperf throughput too low: {mbps:.2f} Mbps' + script = Path(__file__).resolve().parent / 'net_test.py' + cmd = [sys.executable, str(script), '--uid', board['uid']] + ret = hil_util.run_cmd(cmd, timeout=110, split_stderr=True, quiet=True) + assert ret.returncode == 0, (f'USB network test failed (rc={ret.returncode}): ' + f'{ret.stdout} {ret.stderr}') + print(ret.stdout.strip(), end='') + rate = re.search(r'iperf: ([\d.]+) Mbit/s', ret.stdout) + return f'{hil_report.REPORT_CELL["pass"]} {rate.group(1)} Mbit/s' if rate else None def test_device_msc_dual_lun(board): diff --git a/test/hil/net_test.py b/test/hil/net_test.py new file mode 100644 index 0000000000..4377f0b8f7 --- /dev/null +++ b/test/hil/net_test.py @@ -0,0 +1,253 @@ +#!/usr/bin/env python3 +# SPDX-License-Identifier: MIT +"""Exercise the webserver and iperf server on one USB device, independent of host DHCP and routing. + +Invoked by hil_test.py while it holds the board lock. Only the USB network interface +whose parent has the requested serial is moved. Each board gets its own namespace: +the example deliberately shares its MAC and 192.168.7.1 with every other board. +Requires iproute2, util-linux, iperf (2.x), udevadm wait, unprivileged user namespaces and sudo +on hil_root (agentrc), whose netns verb is the one root step; run it as a user other than root. +No persistent host configuration. +""" +import argparse +import hashlib +import http.client +import json +import os +from pathlib import Path +import signal +import socket +import sys +import time +import select +import subprocess + +sys.path.append(os.path.dirname(os.path.abspath(__file__))) # PYTHONSAFEPATH drops it +# hil_util is stdlib-only at import. usb_observe and hil_root load agentrc, which only the +# --uid parent uses: the --http client stays stdlib-only. +from helper import hil_util # noqa: E402 + +DEVICE_IP = '192.168.7.1' +IPERF_PORT = 5001 # lwiperf's LWIPERF_TCP_PORT_DEFAULT +IPERF_SECONDS = 3 +# Default assets in the pinned lwIP dependency's src/apps/http/fsdata.c. Verify +# complete responses, including the binary image, rather than just TCP connect/200. +ASSETS = ( + ('/index.html', 1751, '1da17ad4638e314329a47a9176973ea59ff0b19fc498f55f01c19053c7816fcc'), + ('/img/sics.gif', 724, '3b4eb378fdefac1e7dd6dbb6c84db20e1512463457626f5d991bf7c1ccb7cf63'), +) + + +def command(argv, timeout=5): + ret = hil_util.run_cmd(argv, timeout=timeout, split_stderr=True, quiet=True) + if ret.returncode: + raise RuntimeError(f'{" ".join(argv)}: rc={ret.returncode}: {ret.stderr} {ret.stdout}') + return ret.stdout + + +def interfaces_for_uid(uid): + from helper import usb_observe # the --uid parent only; see the import note above + matches = [] + for dev in usb_observe.usb_scan(vid='cafe', serial=uid): + matches.extend(p.name for p in Path(dev['dir']).glob('*:*/net/*')) + return sorted(set(matches)) + + +def busport_for_uid(uid): + from helper import usb_observe # the --uid parent only; see the import note above + devs = usb_observe.usb_scan(vid='cafe', serial=uid) + if len(devs) != 1: + raise RuntimeError(f'USB serial {uid}: expected one device, found {len(devs)}') + return Path(devs[0]['dir']).name + + +def hil_root(): + from helper import agentrc # the --uid parent only; see the import note above + return agentrc.load_file('skills/hil/scripts/hil_root.py') + + +def wait_interface(uid, timeout=30): + deadline = time.monotonic() + timeout + matches = [] + detail = '' + while time.monotonic() < deadline: + matches = interfaces_for_uid(uid) + if len(matches) == 1: + # A sysfs node exists BEFORE udev finishes initializing/renaming it. + # Wait for this device only: a global settle would also wait for peers' + # mtp-probe jobs. Rediscover by serial if the name changed during wait. + ret = hil_util.run_cmd(['udevadm', 'wait', '--timeout=1', + '/sys/class/net/' + matches[0]], + timeout=2, split_stderr=True, quiet=True) + detail = ret.stderr or '' + if ret.returncode == 0 and interfaces_for_uid(uid) == matches: + return matches[0] + time.sleep(0.2) + raise RuntimeError(f'USB serial {uid}: expected one initialized network interface within {timeout}s; ' + f'found {matches}; {detail}') + + +def interface_index(iface): + """The kernel ifindex of `iface` in our namespace; it survives a move into a fresh one.""" + return int(Path('/sys/class/net', iface, 'ifindex').read_text()) + + +def close_pipes(proc): + """Close our ends of a child's pipes. Never raises.""" + for pipe in (proc.stdout, proc.stderr, proc.stdin): + try: + if pipe is not None: + pipe.close() + except OSError: + pass + + +def stop_client(proc, grace=5): + # unshare execs the client, so proc is the client itself. Never killpg: it shares + # the HIL caller's group. + for sig in (signal.SIGTERM, signal.SIGKILL): + proc.send_signal(sig) + try: + proc.communicate(timeout=grace) + return + except subprocess.TimeoutExpired: + pass + close_pipes(proc) + raise RuntimeError('network client did not exit after SIGKILL') + + +def check_device(uid): + iface = wait_interface(uid) + # Bind the move to this device: hil_root moves the one interface at its busport and + # prints that interface's index, which must be the one read here. + ifindex = interface_index(iface) + busport = busport_for_uid(uid) + print(f'USB serial {uid}: interface {iface} (ifindex {ifindex}) at {busport}', flush=True) + # Anonymous user and network namespace, ours: its last process exiting releases the + # USB interface, even on SIGKILL. hil_root moves the interface in only for a namespace + # a user other than root (the sudo caller) created. + argv = ['unshare', '--user', '--map-root-user', '--net', sys.executable, + str(Path(__file__).resolve()), '--http'] + # Keep the child in our process group: hil_test's interruption sweep can kill + # the whole group without stranding a detached namespace owner. + proc = subprocess.Popen(argv, stdin=subprocess.PIPE, stdout=subprocess.PIPE, + stderr=subprocess.PIPE, text=True) + try: + if not select.select([proc.stdout], [], [], 5)[0]: + raise RuntimeError('network namespace client did not become ready') + ready = proc.stdout.readline(64).strip() + if not ready.isdecimal(): + raise RuntimeError(f'network namespace client failed to start: {ready!r}') + pid = int(ready) + moved = command(hil_root().argv('netns', busport, str(pid)), timeout=30).split() + if moved[:1] != [str(ifindex)]: + raise RuntimeError(f'hil_root netns {busport}: moved {moved}, expected ifindex {ifindex}') + ip = ['nsenter', '--preserve-credentials', f'--target={pid}', '--user', '--net', 'ip'] + # A fresh netns may also hold fallback tunnels (tunl0, sit0, ...) of loaded modules. + links = json.loads(command(ip + ['-j', 'link', 'show'])) + moved_links = [link for link in links if link.get('ifindex') == ifindex] + if len(moved_links) != 1: + found = [(link['ifname'], link.get('ifindex')) for link in links] + raise RuntimeError(f'pid {pid}: expected one network interface with ifindex {ifindex}, found {found}') + iface = moved_links[0]['ifname'] + command(ip + ['addr', 'flush', 'dev', iface]) + command(ip + ['addr', 'add', '192.168.7.2/24', 'dev', iface]) + command(ip + ['link', 'set', 'dev', iface, 'up']) + try: + stdout, stderr = proc.communicate(input='go\n', timeout=45) + except subprocess.TimeoutExpired as exc: + raise RuntimeError('network client timed out after 45s') from exc + if proc.returncode: + raise RuntimeError(f'network client failed (rc={proc.returncode}): {stdout} {stderr}') + print(stdout, end='', flush=True) + finally: + if proc.poll() is None: + stop_client(proc) + else: + close_pipes(proc) + + +def check_http(): + # Direct connections also ignore the host's HTTP proxy settings. + total = 0 + for _ in range(3): + for path, size, digest in ASSETS: + conn = http.client.HTTPConnection(DEVICE_IP, timeout=3) + try: + # Bringing the interface up triggers the class packet filter. Only + # the first connection may wait for that; data errors are failures. + deadline = time.monotonic() + (10 if total == 0 else 0) + while True: + try: + conn.connect() + break + except OSError: + conn.close() + if time.monotonic() >= deadline: + raise + time.sleep(0.5) + conn.request('GET', path, headers={'Connection': 'close'}) + response = conn.getresponse() + data = response.read(size + 1) + if response.status != 200 or len(data) != size or hashlib.sha256(data).hexdigest() != digest: + raise AssertionError(f'{path}: HTTP {response.status}, {len(data)} bytes; expected {size} bytes with SHA256 {digest}') + total += len(data) + finally: + conn.close() + print(f'HTTP: {3 * len(ASSETS)} responses verified, {total} bytes', flush=True) + + +def check_iperf(): + # lwiperf runs only on the high-throughput SRAM tier (INCLUDE_IPERF), so a refused port is + # reported, not failed: it cannot tell a build without lwiperf from one whose server failed + # to start. Its tradeoff (-r) and dual (-d) tests fail against iperf 2.2, so only host to + # device is measured. + try: + socket.create_connection((DEVICE_IP, IPERF_PORT), timeout=3).close() + except ConnectionRefusedError: + print(f'iperf: port {IPERF_PORT} refused, no server', flush=True) + return + try: + ret = subprocess.run(['iperf', '-c', DEVICE_IP, '-p', str(IPERF_PORT), '-t', str(IPERF_SECONDS), '-y', 'C'], + capture_output=True, text=True, timeout=IPERF_SECONDS + 15) + except subprocess.TimeoutExpired as exc: + raise RuntimeError(f'iperf timed out after {exc.timeout}s') from exc + # iperf 2 exits 0 even on a refused or broken connection: only one whole CSV report counts. + lines = ret.stdout.splitlines() + fields = lines[0].split(',') if len(lines) == 1 else [] + if ret.returncode or ret.stderr.strip() or len(fields) != 9: + raise RuntimeError(f'iperf failed (rc={ret.returncode}): {ret.stdout} {ret.stderr}') + *_, interval, nbytes, bps = fields + start, end = map(float, interval.split('-')) + if end - start < IPERF_SECONDS - 0.5 or int(nbytes) <= 0: + raise RuntimeError(f'iperf ended early: {lines[0]}') + print(f'iperf: {int(bps) / 1e6:.1f} Mbit/s host to device', flush=True) + + +def interrupted(signum, _frame): + raise TimeoutError(f'network test interrupted (signal {signum})') + + +def main(): + ap = argparse.ArgumentParser(description=__doc__) + mode = ap.add_mutually_exclusive_group(required=True) + mode.add_argument('--uid') + mode.add_argument('--http', action='store_true', help=argparse.SUPPRESS) + args = ap.parse_args() + # Each side bounds itself even if the other, or its caller, dies. + signal.signal(signal.SIGALRM, interrupted) + signal.signal(signal.SIGTERM, interrupted) + signal.alarm(90) + if args.http: + print(os.getpid(), flush=True) + if sys.stdin.readline().strip() != 'go': + raise RuntimeError('network setup owner exited before starting HTTP') + check_http() + check_iperf() + else: + check_device(args.uid) + signal.alarm(0) + + +if __name__ == '__main__': + main() diff --git a/test/hil/requirements.txt b/test/hil/requirements.txt index e3e0bdcfb7..2cfadf902b 100644 --- a/test/hil/requirements.txt +++ b/test/hil/requirements.txt @@ -1,10 +1,14 @@ # System packages (install separately): -# sudo apt install mtools libmtp9 libmtp-runtime alsa-utils iperf +# sudo apt install mtools libmtp9 libmtp-runtime alsa-utils iproute2 iperf udev util-linux # mtools - read_disk_file (device/cdc_msc, device/msc_dual_lun) # libmtp9 - pymtp ctypes load (device/mtp); Debian 13 uses libmtp9t64 # libmtp-runtime - mtp-probe and the completed-device /dev/libmtp-* marker # alsa-utils - arecord (device/audio_test_freertos) -# iperf - throughput tests (device/net_lwip_*) +# iproute2 - isolated USB network test, run inside its own namespace +# iperf - its host-to-device TCP rate, on builds with the lwiperf server +# util-linux - unshare/nsenter for per-process user and network namespaces; the one root +# step, moving the interface in, is agentrc's hil_root netns (sudo) +# udev - udevadm wait (systemd >= 251) for USB network initialization hidapi pyserial esptool diff --git a/test/hil/test/test_change_impact.py b/test/hil/test/test_change_impact.py index b66ab8e8a0..a184cdfef4 100644 --- a/test/hil/test/test_change_impact.py +++ b/test/hil/test/test_change_impact.py @@ -1086,6 +1086,7 @@ def test_nothing_rig_affecting_has_moved_into_the_carve_out(self): 'test/hil/test/test_hil_flash.py', 'test/hil/test/test_hil_lock.py', 'test/hil/test/test_hil_mtp_raw.py', + 'test/hil/test/test_hil_net.py', 'test/hil/test/test_hil_pr_comment.py', 'test/hil/test/test_hil_pyocd.py', 'test/hil/test/test_hil_remote.py', @@ -1456,12 +1457,13 @@ def test_hil_lib_selects_the_using_tests(self): self.assertNotIn('stm32f407disco', s['boards']) self.assertNotIn('espressif_s3_devkitm', s['boards']) - def test_hil_lib_used_only_by_a_disabled_test_selects_nothing(self): - # device/net_lwip_webserver is commented out of hil_util.device_tests, so the - # intersection with the HIL universe is empty - s = sel(['lib/networking/dhserver.c']) - self.assertFalse(s['full']) - self.assertEqual(s['boards'], {}) + def test_hil_network_changes_select_webserver(self): + for path in ('lib/networking/dhserver.c', 'src/class/net/ncm_device.c'): + with self.subTest(path=path): + s = sel([path]) + self.assertFalse(s['full']) + self.assertEqual(set(s['boards']['stm32f407disco']), + {'device/net_lwip_webserver'}) def test_hil_lib_nobody_builds_selects_nothing(self): s = sel(['lib/SEGGER_RTT/RTT/SEGGER_RTT.c']) diff --git a/test/hil/test/test_hil_bounded.py b/test/hil/test/test_hil_bounded.py index af34a9d2f4..75ed3ed7e2 100644 --- a/test/hil/test/test_hil_bounded.py +++ b/test/hil/test/test_hil_bounded.py @@ -428,7 +428,7 @@ def imported_paths(pyfile): else: yield node.module.replace('.', '/') + '.py' - seeds = ['hil_test.py', 'usbtest.py', 'mtp_test.py'] # CLI + spawned helpers + seeds = ['hil_test.py', 'usbtest.py', 'mtp_test.py', 'net_test.py'] # CLI + spawned helpers for f in seeds: # a renamed seed must fail loudly, not fall out of the walk self.assertTrue((hil_dir / f).exists(), f'stale RemoteStaging seed: {f}') todo, seen = list(seeds), set() diff --git a/test/hil/test/test_hil_net.py b/test/hil/test/test_hil_net.py new file mode 100644 index 0000000000..88503324c8 --- /dev/null +++ b/test/hil/test/test_hil_net.py @@ -0,0 +1,372 @@ +#!/usr/bin/env python3 +# SPDX-License-Identifier: MIT +"""Network regression coverage without USB hardware or privileged host changes.""" +import hashlib +import io +import json +import os +from pathlib import Path +import subprocess +import sys +from tempfile import TemporaryDirectory +import unittest +from unittest.mock import Mock, patch + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +import net_test as net + +SCRIPT = str(Path(net.__file__).resolve()) + + +def clean_env(tmp): + """A fresh interpreter's env with no agentrc to find: the --http client needs none.""" + env = {k: v for k, v in os.environ.items() + if k not in ('GITHUB_ACTIONS', 'CI') and not k.startswith('PYTHON')} + env.update(PYTHONSAFEPATH='1', AGENTRC_DIR=tmp, HOME=tmp, USERPROFILE=tmp) + return env + + +class Startup(unittest.TestCase): + # hil_test.py starts net_test.py in a fresh interpreter, and the --http child under + # unshare; the in-process tests above add test/hil to sys.path and would hide both. + def test_safe_path_startup_finds_helper(self): + with TemporaryDirectory() as tmp: + ret = subprocess.run([sys.executable, SCRIPT, '--help'], cwd=tmp, env=clean_env(tmp), + capture_output=True, text=True, timeout=30) + self.assertNotIn('ModuleNotFoundError', ret.stderr) + self.assertEqual(ret.returncode, 0, ret.stderr) + + @unittest.skipUnless(os.name == 'posix', 'the --http client arms SIGALRM') + def test_http_client_does_not_load_agentrc(self): + # runpy does not put the script's folder on sys.path: its own append must. + code = '''import io, runpy, signal, sys +sys.argv = [sys.argv[1], '--http'] +sys.stdin = io.StringIO('') +try: + runpy.run_path(sys.argv[0], run_name='__main__') +except RuntimeError as e: + print('stopped:', e) +signal.alarm(0) +print('loaded:', sorted(m for m in sys.modules if 'agentrc' in m or 'usb_observe' in m)) +''' + with TemporaryDirectory() as tmp: + ret = subprocess.run([sys.executable, '-c', code, SCRIPT], cwd=tmp, env=clean_env(tmp), + capture_output=True, text=True, timeout=30) + self.assertEqual(ret.returncode, 0, ret.stderr) + self.assertIn('stopped: network setup owner exited before starting HTTP', ret.stdout) + self.assertIn('loaded: []', ret.stdout) + + +class InterfaceSelection(unittest.TestCase): + @unittest.skipUnless(os.name == 'posix', 'sysfs interface names contain colons') + def test_interface_comes_from_usb_serial_not_shared_mac_or_name(self): + with TemporaryDirectory() as tmp: + target = Path(tmp) / '1-2' + other = Path(tmp) / '1-3' + (target / '1-2:1.0/net/usb7').mkdir(parents=True) + (other / '1-3:1.0/net/enx0202846a9600').mkdir(parents=True) + with patch('helper.usb_observe.usb_scan', return_value=[{'dir': str(target)}]) as scan: + self.assertEqual(net.interfaces_for_uid('ABC123'), ['usb7']) + scan.assert_called_once_with(vid='cafe', serial='ABC123') + + def test_busport_is_the_one_device_with_the_serial(self): + with patch('helper.usb_observe.usb_scan', return_value=[{'dir': '/sys/bus/usb/devices/1-2.3'}]) as scan: + self.assertEqual(net.busport_for_uid('ABC123'), '1-2.3') + scan.assert_called_once_with(vid='cafe', serial='ABC123') + for found in ([], [{'dir': '/x/1-2'}, {'dir': '/x/1-3'}]): + with self.subTest(found=len(found)), patch('helper.usb_observe.usb_scan', return_value=found): + with self.assertRaisesRegex(RuntimeError, f'expected one device, found {len(found)}'): + net.busport_for_uid('ABC123') + + def test_waits_for_driver_interface(self): + with patch.object(net, 'interfaces_for_uid', side_effect=[[], ['usb7'], ['usb7']]), \ + patch.object(net.hil_util, 'run_cmd', return_value=Mock(returncode=0, stderr='')), \ + patch.object(net.time, 'sleep'): + self.assertEqual(net.wait_interface('ABC123'), 'usb7') + + def test_udev_rename_is_followed_by_serial(self): + with patch.object(net, 'interfaces_for_uid', + side_effect=[['usb0'], ['enx123'], ['enx123'], ['enx123']]), \ + patch.object(net.hil_util, 'run_cmd', return_value=Mock(returncode=0, stderr='')), \ + patch.object(net.time, 'sleep'): + self.assertEqual(net.wait_interface('ABC123'), 'enx123') + + def test_uninitialized_interface_is_not_moved(self): + with patch.object(net, 'interfaces_for_uid', return_value=['usb0']), \ + patch.object(net.hil_util, 'run_cmd', return_value=Mock(returncode=1, stderr='not ready')), \ + patch.object(net.time, 'monotonic', side_effect=[0, 0, 2]), \ + patch.object(net.time, 'sleep'): + with self.assertRaisesRegex(RuntimeError, 'not ready'): + net.wait_interface('ABC123', timeout=1) + + def test_ambiguous_serial_never_selects_arbitrary_board(self): + with patch.object(net, 'interfaces_for_uid', return_value=['usb0', 'usb1']), \ + patch.object(net.time, 'monotonic', side_effect=[0, 0, 2]), \ + patch.object(net.time, 'sleep'): + with self.assertRaisesRegex(RuntimeError, 'found.*usb0.*usb1'): + net.wait_interface('ABC123', timeout=1) + + +class NamespaceLifetime(unittest.TestCase): + def setUp(self): + self.calls = [] + self.proc = Mock(returncode=0, pid=1234) + self.proc.stdout = io.StringIO('4321\n') + self.proc.poll.return_value = 0 + self.proc.communicate.return_value = ('HTTP verified\n', '') + + self.moved_ifindex = 7 + self.printed_ifindex = 7 + self.extra_links = [] + + def command(argv, timeout=5): + self.calls.append(argv) + if argv[:2] == ['hil_root', 'netns']: + return f'{self.printed_ifindex} usb7\n' + if '-j' in argv: + return json.dumps([{'ifname': 'lo', 'ifindex': 1}, *self.extra_links, + {'ifname': 'renamed0', 'ifindex': self.moved_ifindex}]) + return '' + + hil_root = Mock() + hil_root.argv.side_effect = lambda verb, *args: ['hil_root', verb, *args] + for name, value in [('wait_interface', Mock(return_value='usb7')), + ('interface_index', Mock(return_value=7)), + ('busport_for_uid', Mock(return_value='1-2')), + ('hil_root', Mock(return_value=hil_root)), + ('command', Mock(side_effect=command))]: + p = patch.object(net, name, value) + p.start() + self.addCleanup(p.stop) + for p in [patch.object(net.subprocess, 'Popen', return_value=self.proc), + patch.object(net.select, 'select', return_value=([self.proc.stdout], [], [])), + patch.object(net.signal, 'SIGKILL', 9, create=True), + patch('sys.stdout', new_callable=io.StringIO)]: + p.start() + self.addCleanup(p.stop) + + def test_client_runs_unprivileged_and_hil_root_moves_the_interface(self): + net.check_device('ABC123') + argv = net.subprocess.Popen.call_args.args[0] + self.assertEqual(argv, ['unshare', '--user', '--map-root-user', '--net', sys.executable, SCRIPT, '--http']) + self.assertNotIn('start_new_session', net.subprocess.Popen.call_args.kwargs) + self.assertEqual(self.calls[0], ['hil_root', 'netns', '1-2', '4321']) + self.assertFalse(any('sudo' in c for c in self.calls[1:])) + nsenter = ['nsenter', '--preserve-credentials', '--target=4321', '--user', '--net', 'ip'] + self.assertTrue(all(c[:6] == nsenter for c in self.calls[1:])) + self.assertTrue(any('renamed0' in c for c in self.calls)) + self.proc.communicate.assert_called_once_with(input='go\n', timeout=45) + + def test_fallback_tunnels_next_to_the_moved_interface_are_ignored(self): + self.extra_links = [{'ifname': 'tunl0', 'ifindex': 2}, {'ifname': 'sit0', 'ifindex': 3}] + net.check_device('ABC123') + self.assertIn(['set', 'dev', 'renamed0', 'up'], [c[-4:] for c in self.calls]) + self.assertFalse(any('tunl0' in c or 'sit0' in c for c in self.calls)) + self.proc.communicate.assert_called_once_with(input='go\n', timeout=45) + + def test_namespace_holding_another_ifindex_fails(self): + self.moved_ifindex = 8 + self.proc.poll.return_value = None + with self.assertRaisesRegex(RuntimeError, 'ifindex 7.*renamed0.*8'): + net.check_device('ABC123') + self.assertFalse(any('addr' in c for c in self.calls)) + self.proc.send_signal.assert_called_once_with(net.signal.SIGTERM) + + def test_hil_root_moving_another_index_fails_before_setup(self): + self.printed_ifindex = 9 + self.proc.poll.return_value = None + with self.assertRaisesRegex(RuntimeError, r"moved \['9', 'usb7'\], expected ifindex 7"): + net.check_device('ABC123') + self.assertEqual(len(self.calls), 1) + self.proc.send_signal.assert_called_once_with(net.signal.SIGTERM) + + def test_interrupt_terminates_client(self): + net.command.side_effect = KeyboardInterrupt() + self.proc.poll.return_value = None + with self.assertRaises(KeyboardInterrupt): + net.check_device('ABC123') + self.proc.send_signal.assert_called_once_with(net.signal.SIGTERM) + self.proc.communicate.assert_called_once_with(timeout=5) + + def test_setup_failure_releases_namespace_owner(self): + net.command.side_effect = RuntimeError('move failed') + self.proc.poll.return_value = None + with self.assertRaisesRegex(RuntimeError, 'move failed'): + net.check_device('ABC123') + self.proc.send_signal.assert_called_once_with(net.signal.SIGTERM) + + def test_stalled_client_escalates_after_cleanup_grace(self): + self.proc.communicate.side_effect = [net.subprocess.TimeoutExpired('client', 5), ('', '')] + net.stop_client(self.proc) + self.assertEqual([c.args for c in self.proc.send_signal.call_args_list], + [(net.signal.SIGTERM,), (net.signal.SIGKILL,)]) + + def test_http_failure_is_reported(self): + self.proc.returncode = 1 + self.proc.communicate.return_value = ('', 'corrupt data') + with self.assertRaisesRegex(RuntimeError, 'corrupt data'): + net.check_device('ABC123') + + def test_readiness_timeout_stops_the_client(self): + net.select.select.return_value = ([], [], []) + self.proc.poll.return_value = None + with self.assertRaisesRegex(RuntimeError, 'did not become ready'): + net.check_device('ABC123') + net.command.assert_not_called() + self.proc.send_signal.assert_called_once_with(net.signal.SIGTERM) + self.proc.communicate.assert_called_once_with(timeout=5) + + +class HttpVerification(unittest.TestCase): + def setUp(self): + self.payload = b'\x00\xffbinary\r\nasset' + self.response = Mock(status=200) + self.response.read.return_value = self.payload + self.conn = Mock() + self.conn.getresponse.return_value = self.response + for p in [patch.object(net, 'ASSETS', [('/fixture', len(self.payload), + hashlib.sha256(self.payload).hexdigest())]), + patch.object(net.http.client, 'HTTPConnection', return_value=self.conn), + patch('sys.stdout', new_callable=io.StringIO)]: + p.start() + self.addCleanup(p.stop) + + def test_repeated_complete_binary_responses(self): + net.check_http() + self.assertEqual(self.conn.request.call_count, 3) + self.assertEqual(self.conn.close.call_count, 3) + + def test_every_asset_is_requested_each_round(self): + # Each asset is its own fixture, one of them binary: a loop over ASSETS[:1] + # would skip the image on hardware while a single-asset test still passed. + payloads = {'/page.html': b'ok', '/img/bin.gif': b'GIF89a\x00\xff\r\n\x01'} + requested = [] + + def request(method, path, headers=None): + requested.append(path) + self.response.read.return_value = payloads[path] + + self.conn.request.side_effect = request + with patch.object(net, 'ASSETS', [(path, len(data), hashlib.sha256(data).hexdigest()) + for path, data in payloads.items()]): + net.check_http() + self.assertEqual({path: requested.count(path) for path in payloads}, + {path: 3 for path in payloads}) + self.assertEqual(len(requested), 6) + + def test_corruption_is_not_retried(self): + self.response.read.return_value = b'x' * len(self.payload) + with self.assertRaises(AssertionError): + net.check_http() + self.assertEqual(self.conn.request.call_count, 1) + self.conn.close.assert_called_once() + + def test_truncated_and_excess_data_fail(self): + for payload in (self.payload[:-1], self.payload + b'x'): + with self.subTest(payload=payload): + self.response.read.return_value = payload + with self.assertRaises(AssertionError): + net.check_http() + + def test_http_error_with_correct_body_fails(self): + self.response.status = 404 + with self.assertRaises(AssertionError): + net.check_http() + + def test_initial_connect_can_wait_for_link(self): + self.conn.connect.side_effect = [ConnectionRefusedError(), None, None, None] + with patch.object(net.time, 'sleep'): + net.check_http() + self.assertEqual(self.conn.request.call_count, 3) + + def test_later_connection_failure_is_not_hidden(self): + self.conn.connect.side_effect = [None, ConnectionRefusedError()] + with self.assertRaises(ConnectionRefusedError): + net.check_http() + self.assertEqual(self.conn.request.call_count, 1) + + @unittest.skipUnless(os.name == 'posix', 'process-group timeout is Linux HIL behavior') + def test_command_reaps_a_stalled_client(self): + with self.assertRaisesRegex(RuntimeError, 'rc=124'): + net.command([sys.executable, '-c', 'import time; time.sleep(60)'], timeout=0.1) + + +class IperfVerification(unittest.TestCase): + # Observed iperf 2.2.1 output against lwiperf: a 3 s host-to-device run, and a tradeoff + # run whose reverse leg failed yet exited 0. + GOOD = '20261110074851,192.168.7.2,38608,192.168.7.1,5001,1,0.0-3.3,1441856,3538660\n' + BROKEN = ('[ 1] tcp write failed\n[ 1] shutdown failed: Transport endpoint is not connected\n' + '20261110074852,192.168.7.2,38616,192.168.7.1,5001,1,0.0-0.0,75920,47917948\n' + 'ERROR: expected reverse connect did not occur\n') + + def setUp(self): + self.run = Mock(return_value=subprocess.CompletedProcess([], 0, self.GOOD, '')) + self.connect = Mock() + for p in [patch.object(net.subprocess, 'run', self.run), + patch.object(net.socket, 'create_connection', self.connect), + patch('sys.stdout', new_callable=io.StringIO)]: + p.start() + self.addCleanup(p.stop) + + def test_rate_from_one_whole_report(self): + net.check_iperf() + self.assertIn('iperf: 3.5 Mbit/s host to device', sys.stdout.getvalue()) + argv = self.run.call_args.args[0] + self.assertEqual(argv[:3], ['iperf', '-c', net.DEVICE_IP]) + self.assertIn('-y', argv) + + def test_refused_port_is_reported_not_failed(self): + self.connect.side_effect = ConnectionRefusedError() + net.check_iperf() + self.assertIn('iperf: port 5001 refused, no server', sys.stdout.getvalue()) + self.run.assert_not_called() + + def test_broken_or_refused_run_exiting_zero_fails(self): + refused = '[ 1] tcp connect to 192.168.7.1 port 5001 failed (Connection refused)\n' + for out, err in ((self.BROKEN, ''), (refused, ''), ('', ''), (self.GOOD, 'warning\n')): + with self.subTest(out=out, err=err): + self.run.return_value = subprocess.CompletedProcess([], 0, out, err) + with self.assertRaisesRegex(RuntimeError, 'iperf failed'): + net.check_iperf() + + def test_short_or_empty_run_fails(self): + for old, new in (('0.0-3.3', '0.0-0.4'), (',1441856,', ',0,')): + with self.subTest(new=new): + self.run.return_value = subprocess.CompletedProcess([], 0, self.GOOD.replace(old, new), '') + with self.assertRaisesRegex(RuntimeError, 'ended early'): + net.check_iperf() + + def test_hung_run_fails_with_its_bound(self): + self.run.side_effect = subprocess.TimeoutExpired(['iperf'], 18) + with self.assertRaisesRegex(RuntimeError, 'iperf timed out after 18s'): + net.check_iperf() + + def test_unreachable_port_is_not_read_as_missing_server(self): + self.connect.side_effect = TimeoutError() + with self.assertRaises(TimeoutError): + net.check_iperf() + self.run.assert_not_called() + + +class ReportCell(unittest.TestCase): + def setUp(self): + import usbtest_harness # noqa: F401 - stubs serial before hil_test is imported + import hil_test + self.hil_test = hil_test + + def cell(self, stdout): + ret = subprocess.CompletedProcess([], 0, stdout, '') + with patch.object(self.hil_test.hil_util, 'run_cmd', return_value=ret), \ + patch('sys.stdout', new_callable=io.StringIO): + return self.hil_test.test_device_net_lwip_webserver({'uid': 'ABC123'}) + + def test_iperf_rate_is_the_cell(self): + self.assertEqual(self.cell('HTTP: 6 responses verified, 7425 bytes\niperf: 3.5 Mbit/s host to device\n'), + '\u2705 3.5 Mbit/s') + + def test_no_iperf_server_keeps_the_plain_cell(self): + self.assertIsNone(self.cell('HTTP: 6 responses verified, 7425 bytes\niperf: port 5001 refused, no server\n')) + + +if __name__ == '__main__': + unittest.main()