diff --git a/test/test_hil_esp32_self_serve_ota.py b/test/test_hil_esp32_self_serve_ota.py index acdf8da2..784f519a 100644 --- a/test/test_hil_esp32_self_serve_ota.py +++ b/test/test_hil_esp32_self_serve_ota.py @@ -25,7 +25,9 @@ def load_module(name, path): return module -SOAK = load_module("self_serve_test_soak", HIL_DIR / "s3_memory_soak.py") +SESSION = load_module("self_serve_test_serial_session", HIL_DIR / "serial_session.py") +with mock.patch.dict(sys.modules, {"serial_session": SESSION}): + SOAK = load_module("self_serve_test_soak", HIL_DIR / "s3_memory_soak.py") with mock.patch.dict(sys.modules, {"s3_memory_soak": SOAK}): HIL = load_module("self_serve_test_harness", HIL_DIR / "esp32_self_serve_ota.py") diff --git a/test/test_hil_esp32_usb_host_reboot.py b/test/test_hil_esp32_usb_host_reboot.py index c9aeceb4..b659c0b6 100644 --- a/test/test_hil_esp32_usb_host_reboot.py +++ b/test/test_hil_esp32_usb_host_reboot.py @@ -25,7 +25,9 @@ def load_module(name, path): return module -SOAK = load_module("usb_reboot_test_soak", HIL_DIR / "s3_memory_soak.py") +SESSION = load_module("usb_reboot_test_serial_session", HIL_DIR / "serial_session.py") +with mock.patch.dict(sys.modules, {"serial_session": SESSION}): + SOAK = load_module("usb_reboot_test_soak", HIL_DIR / "s3_memory_soak.py") with mock.patch.dict(sys.modules, {"s3_memory_soak": SOAK}): OTA = load_module("usb_reboot_test_ota", HIL_DIR / "esp32_self_serve_ota.py") with mock.patch.dict(sys.modules, {"esp32_self_serve_ota": OTA}): diff --git a/tools/hil/contact_cache_serial_stress.py b/tools/hil/contact_cache_serial_stress.py index 66de5c49..0ba17da7 100644 --- a/tools/hil/contact_cache_serial_stress.py +++ b/tools/hil/contact_cache_serial_stress.py @@ -19,6 +19,8 @@ import struct import time import serial +from serial_session import open_configured_session +from profile_switch import configure_session from esp32_companion_serial_stress import ( DeviceFrameReader, @@ -43,12 +45,10 @@ def secret_counters(reply): def run(args): port = serial.Serial(None, 115200, timeout=0.05, write_timeout=10) - port.dtr = False - port.rts = False port.port = args.port - port.open() + configure_session(port) + open_configured_session(port) try: - port.dtr = True time.sleep(4) reader = DeviceFrameReader(TransportCounters()) diff --git a/tools/hil/esp32_companion_serial_stress.py b/tools/hil/esp32_companion_serial_stress.py index bbf71950..af315a1a 100644 --- a/tools/hil/esp32_companion_serial_stress.py +++ b/tools/hil/esp32_companion_serial_stress.py @@ -30,6 +30,7 @@ import struct import sys import time from typing import Any, Callable, Dict, Optional, Sequence, Tuple +from serial_session import open_configured_session HOST_FRAME_MARKER = ord("<") @@ -466,9 +467,9 @@ def _make_pyserial_factory() -> Callable[[StressConfig], Any]: ) from exc def factory(config: StressConfig) -> Any: - # Native USB/JTAG devices may require DTR for the CDC session. Set both - # control lines while closed and keep RTS deasserted, including when - # opting into DTR, to avoid deliberately driving the ESP reset line. + # Native USB/JTAG devices may require DTR for the CDC session. POSIX + # pySerial opens bridges by applying DTR before RTS; defer DTR there + # so a kernel's initial asserted lines cannot pulse the ESP reset line. port = serial.Serial() port.port = config.port port.baudrate = config.baudrate @@ -476,7 +477,7 @@ def _make_pyserial_factory() -> Callable[[StressConfig], Any]: port.write_timeout = config.write_timeout port.dtr = config.dtr port.rts = False - port.open() + open_configured_session(port) return port return factory diff --git a/tools/hil/fem_noise_floor_test.py b/tools/hil/fem_noise_floor_test.py index 2be75a1a..2ddc0afe 100755 --- a/tools/hil/fem_noise_floor_test.py +++ b/tools/hil/fem_noise_floor_test.py @@ -42,6 +42,7 @@ import time from typing import Any, Callable, Dict, List, Optional, Sequence import esp32_companion_serial_stress as companion +from serial_session import open_configured_session CMD_GET_FEM_RX_GAIN = 0x42 @@ -141,16 +142,17 @@ def _radio_stats_spec() -> companion.RequestSpec: ) -def _resolve_dtr(config: FemTestConfig) -> bool: +def _resolve_dtr(config: FemTestConfig, port: Any) -> bool: if config.dtr == "on": return True if config.dtr == "off": return False - # Text consoles require an asserted host session on both nRF52 USB CDC and - # ESP32-S3 USB-Serial/JTAG. Companion framing on V4 remains usable without - # DTR and avoids changing the device's host-session state during a binary - # protocol test. - return config.protocol == "cli" or "HT-n5262" in config.port + # AUTO follows the actual USB transport, not the CLI/framing protocol. + # A bridge's DTR may hold GPIO0; native CDC uses it as a host session. + # Import only after open() has requested the optional serial dependency. + from profile_switch import configure_session + configure_session(port) + return bool(port.dtr) class CompanionFemTransport: @@ -176,9 +178,9 @@ class CompanionFemTransport: self.config.read_poll_timeout, self.config.response_timeout ) port.write_timeout = self.config.write_timeout - port.dtr = _resolve_dtr(self.config) + port.dtr = _resolve_dtr(self.config, port) port.rts = False - port.open() + open_configured_session(port) self.counters.ports_opened += 1 self.port = port try: @@ -292,9 +294,9 @@ class CliFemTransport: self.config.read_poll_timeout, self.config.response_timeout ) port.write_timeout = self.config.write_timeout - port.dtr = _resolve_dtr(self.config) + port.dtr = _resolve_dtr(self.config, port) port.rts = False - port.open() + open_configured_session(port) self.port = port try: if self.config.open_delay: diff --git a/tools/hil/profile_four_tx_deploy.py b/tools/hil/profile_four_tx_deploy.py index 7977089b..a161c59b 100644 --- a/tools/hil/profile_four_tx_deploy.py +++ b/tools/hil/profile_four_tx_deploy.py @@ -7,7 +7,7 @@ import subprocess import time import zipfile from profile_four_tx_fixture import TXS,RX,resolve -from profile_switch import exchange,configure_session +from profile_switch import exchange,open_session import serial from serial.tools import list_ports @@ -52,7 +52,7 @@ def ready(t): try: name=resolve(t) p=serial.Serial();p.port=name;p.baudrate=115200;p.timeout=.2;p.write_timeout=2 - configure_session(p);p.open() + open_session(p) try: info=exchange(p,'info','board',3) if info.get('bench')!='production-profile-switch-v8' or not info.get('ready'): diff --git a/tools/hil/profile_four_tx_package.py b/tools/hil/profile_four_tx_package.py index 3e243a13..dd4cabe7 100644 --- a/tools/hil/profile_four_tx_package.py +++ b/tools/hil/profile_four_tx_package.py @@ -41,7 +41,7 @@ for name in ('firmware.bin','partitions.bin','bootloader.bin'): add('rx-'+name,(ROOT/'.pio/build'/RX['env']/name).read_bytes()) add('rx-boot_app0.bin',(Path.home()/'.platformio/packages/framework-arduinoespressif32/tools/partitions/boot_app0.bin').read_bytes()) for name in ('profile_four_tx_fixture.py','profile_four_tx_deploy.py','profile_four_tx.py', - 'profile_switch.py','profile_switch_channels.py','profile_switch_packets.py', + 'profile_switch.py','serial_session.py','profile_switch_channels.py','profile_switch_packets.py', 'profile_switch_sweep.py'): add(name,(ROOT/'tools/hil'/name).read_bytes()) for name in ('profile_fixed_tx.cpp','profile_fixed_tx.ini','profile_switch.cpp','profile_switch_channels.h', diff --git a/tools/hil/profile_four_tx_repeat_run.py b/tools/hil/profile_four_tx_repeat_run.py index d1a2f2a7..bb80b6af 100644 --- a/tools/hil/profile_four_tx_repeat_run.py +++ b/tools/hil/profile_four_tx_repeat_run.py @@ -5,6 +5,7 @@ is the reviewed collector; dependencies and image provenance are hash-checked against the previous bench bundle. No flashing or hub reset is performed. """ import base64 +import ast import hashlib import json from pathlib import Path @@ -24,6 +25,11 @@ if subprocess.run(['systemctl','is-active','--quiet','meshcore-memory-soak.servi raise RuntimeError('V4 soak active; do not interrupt another test') dependencies=('profile_four_tx_fixture.py','profile_switch.py','profile_switch_channels.py', 'profile_switch_packets.py','profile_switch_sweep.py') +# Copy the exact dependency closure of the immutable previous snapshot. +if any((isinstance(node,ast.ImportFrom) and node.module=='serial_session') + or (isinstance(node,ast.Import) and any(alias.name=='serial_session' for alias in node.names)) + for node in ast.walk(ast.parse((previous/'profile_switch.py').read_bytes()))): + dependencies+=('serial_session.py',) files={name:(previous/name).read_bytes() for name in dependencies} for name,data in files.items(): if hashlib.sha256(data).hexdigest()!=manifest['files'][name]['sha256']: diff --git a/tools/hil/profile_pair.py b/tools/hil/profile_pair.py index d900bc1e..5c135f39 100644 --- a/tools/hil/profile_pair.py +++ b/tools/hil/profile_pair.py @@ -14,7 +14,7 @@ import time import serial from serial.tools import list_ports from profile_four_tx_fixture import TXS,RX,resolve -from profile_switch import configure_session,exchange,read_packet_response +from profile_switch import open_session,exchange,read_packet_response TARGETS=(TXS[0],TXS[3]) PLAN=({'sf':7,'bw_khz':62.5,'freq_khz':909500}, @@ -28,7 +28,7 @@ def require(row,expected,label): def open_radio(target,pair=True): port=serial.Serial();port.port=resolve(target);port.baudrate=115200 - port.timeout=.2;port.write_timeout=2;configure_session(port);port.open() + port.timeout=.2;port.write_timeout=2;open_session(port) try: port.reset_input_buffer() info=exchange(port,'info','board',5) diff --git a/tools/hil/profile_pair_4p6_manage.py b/tools/hil/profile_pair_4p6_manage.py index e4907697..22951fcb 100644 --- a/tools/hil/profile_pair_4p6_manage.py +++ b/tools/hil/profile_pair_4p6_manage.py @@ -37,7 +37,7 @@ if args.action=='prepare': manifest['files'][name]={'bytes':len(raw),'sha256':hashlib.sha256(raw).hexdigest()} for name in ('firmware.bin','partitions.bin'):add('rx-'+name,root/'.pio/build/profile_four_tx_v4_rx'/name) for name in ('profile_pair.py','profile_pair_4p6_collect.py','profile_pair_run.py','profile_pair_4p6_deploy.py', - 'profile_switch.py','profile_four_tx_fixture.py','ProfilePairPlan.h','ProfileSwitchUsb.h', + 'profile_switch.py','serial_session.py','profile_four_tx_fixture.py','ProfilePairPlan.h','ProfileSwitchUsb.h', 'profile_switch.cpp','profile_switch.ini','profile_switch_channels.h','profile_switch_experiments.h', 'profile_stationary_baseline.h','ProfileChannelVisitClock.h','ProfileChannelTrace.h','ProfileFrequencyOffset.h'): add(name,root/'tools/hil'/name) diff --git a/tools/hil/profile_pair_launch.py b/tools/hil/profile_pair_launch.py index 7210bc3a..0faaa72b 100644 --- a/tools/hil/profile_pair_launch.py +++ b/tools/hil/profile_pair_launch.py @@ -1,5 +1,6 @@ """Launch the bounded pair collector only once; inspect status separately.""" import hashlib +import ast import json from pathlib import Path import subprocess @@ -7,7 +8,14 @@ root=Path.home()/'hwtest/runs/pair-sf8-32-20260914' if (root/'launch.json').exists() or (root/'results.json').exists():raise RuntimeError('Already launched') if not json.loads((root/'deployment.json').read_text()).get('complete'):raise RuntimeError('Deployment incomplete') manifest=json.loads((root/'manifest.json').read_text()) -for name in ('profile_pair.py','profile_pair_run.py','profile_switch.py','profile_four_tx_fixture.py'): +dependencies=('profile_pair.py','profile_pair_run.py','profile_switch.py','profile_four_tx_fixture.py') +# Historical bundles remain immutable. Require the helper only when their +# own shipped profile_switch imports it, then verify its recorded hash too. +if any((isinstance(node,ast.ImportFrom) and node.module=='serial_session') + or (isinstance(node,ast.Import) and any(alias.name=='serial_session' for alias in node.names)) + for node in ast.walk(ast.parse((root/'profile_switch.py').read_bytes()))): + dependencies+=('serial_session.py',) +for name in dependencies: if hashlib.sha256((root/name).read_bytes()).hexdigest()!=manifest['files'][name]['sha256']: raise RuntimeError('Collector/dependency changed') with (root/'collector.log').open('x') as log: diff --git a/tools/hil/profile_pair_package.py b/tools/hil/profile_pair_package.py index f17e005e..27748ddf 100644 --- a/tools/hil/profile_pair_package.py +++ b/tools/hil/profile_pair_package.py @@ -30,7 +30,7 @@ for target in (TXS[0],TXS[3]): add(target['env']+'.zip',data) manifest['files'][target['env']+'.zip'].update(app_base=min(addresses),image_end=max(addresses)+1) for name in ('firmware.bin','partitions.bin'):add('rx-'+name,(root/'.pio/build'/RX['env']/name).read_bytes()) -for name in ('profile_pair.py','profile_pair_deploy.py','profile_pair_run.py','profile_switch.py','profile_four_tx_fixture.py', +for name in ('profile_pair.py','profile_pair_deploy.py','profile_pair_run.py','profile_switch.py','serial_session.py','profile_four_tx_fixture.py', 'ProfilePairPlan.h','ProfileSwitchUsb.h','profile_fixed_tx.cpp','profile_fixed_tx.ini', 'profile_switch.cpp','profile_switch.ini','profile_switch_channels.h','profile_switch_experiments.h', 'profile_stationary_baseline.h','ProfileChannelVisitClock.h','ProfileChannelTrace.h','ProfileFrequencyOffset.h'): diff --git a/tools/hil/profile_switch.py b/tools/hil/profile_switch.py index 166af68d..4b66a691 100644 --- a/tools/hil/profile_switch.py +++ b/tools/hil/profile_switch.py @@ -10,6 +10,7 @@ import time import serial from serial.tools import list_ports +from serial_session import open_configured_session _run_sequences = itertools.count(secrets.randbelow(0x3fffffff) + 1) @@ -24,6 +25,12 @@ def configure_session(port): for info in list_ports.comports()) +def open_session(port): + """Open a configured native session or a bridge without pulsing reset.""" + configure_session(port) + open_configured_session(port) + + def read_response(port, key, timeout, expected=None): deadline = time.monotonic() + timeout pending = bytearray() @@ -113,9 +120,8 @@ def main(): port.baudrate = 115200 port.timeout = 0.25 port.write_timeout = 2 - configure_session(port) try: - port.open() + open_session(port) port.reset_input_buffer() info = exchange(port, "info", "bench", 5) if info.get("bench") != "production-profile-switch-v8" or not info.get("ready"): diff --git a/tools/hil/profile_switch_dwell.py b/tools/hil/profile_switch_dwell.py index fc8aabfa..f9a5b224 100644 --- a/tools/hil/profile_switch_dwell.py +++ b/tools/hil/profile_switch_dwell.py @@ -11,7 +11,7 @@ import sys import time import serial -from profile_switch import configure_session +from profile_switch import open_session DWELLS = tuple(value / 10 for value in range(41, 82, 5)) @@ -49,9 +49,8 @@ def reboot_boards(names): for name in names: port = serial.Serial() port.port, port.baudrate, port.timeout, port.write_timeout = name, 115200, .25, 2 - configure_session(port) try: - port.open() + open_session(port) port.write(b"reboot\n") port.flush() finally: diff --git a/tools/hil/profile_switch_packets.py b/tools/hil/profile_switch_packets.py index f0565cf9..a1a904e0 100644 --- a/tools/hil/profile_switch_packets.py +++ b/tools/hil/profile_switch_packets.py @@ -6,7 +6,7 @@ import secrets from pathlib import Path import serial -from profile_switch import exchange, exchange_ready, read_packet_response, configure_session +from profile_switch import exchange, exchange_ready, read_packet_response, open_session from profile_switch_sweep import CASES, BULK_CASES, PRODUCTION_CASES @@ -14,8 +14,7 @@ def open_port(name): port = serial.Serial() port.port, port.baudrate, port.timeout = name, 115200, 0.25 port.write_timeout = 2 - configure_session(port) - port.open() + open_session(port) port.reset_input_buffer() info = exchange(port, "info", "bench", 5) if info.get("bench") != "production-profile-switch-v8" or not info.get("ready"): diff --git a/tools/hil/profile_switch_sweep.py b/tools/hil/profile_switch_sweep.py index b6a08fcc..88f2b471 100644 --- a/tools/hil/profile_switch_sweep.py +++ b/tools/hil/profile_switch_sweep.py @@ -3,7 +3,7 @@ import argparse import json from pathlib import Path import serial -from profile_switch import exchange, exchange_ready, configure_session +from profile_switch import exchange, exchange_ready, open_session CASES = [ ("production_batch", 0, 2), ("manual_full_control", 128, 2), @@ -45,9 +45,8 @@ def main(): port = serial.Serial() port.port, port.baudrate, port.timeout = args.port, 115200, 0.25 port.write_timeout = 2 - configure_session(port) try: - port.open() + open_session(port) port.reset_input_buffer() info = exchange(port, "info", "bench", 5) if info.get("bench") != "production-profile-switch-v8" or not info.get("ready"): diff --git a/tools/hil/s3_memory_soak.py b/tools/hil/s3_memory_soak.py index 56fe5653..894b527a 100644 --- a/tools/hil/s3_memory_soak.py +++ b/tools/hil/s3_memory_soak.py @@ -16,6 +16,7 @@ import socketserver import subprocess import threading import time +from serial_session import open_configured_session class Device: @@ -68,7 +69,7 @@ class Device: # Keep RTS low; a held handle avoids repeated reset transitions. stream.dtr = os.name == 'nt' stream.rts = False - stream.open() + open_configured_session(stream) self.stream = stream self.read(time.monotonic()+20) self.write(b'\r') diff --git a/tools/hil/serial_session.py b/tools/hil/serial_session.py new file mode 100644 index 00000000..828f7ad7 --- /dev/null +++ b/tools/hil/serial_session.py @@ -0,0 +1,28 @@ +"""Open an already configured serial session without a bridge reset pulse.""" +import os + + +def open_configured_session(port): + # POSIX pySerial normally applies DTR before RTS when opening. Some + # bridge drivers initially assert both lines; clearing DTR first pulses + # ESP32 EN. Defer DTR until RTS is released. Windows configures both in + # one DCB operation and uses dsrdtr for real handshaking, so leave it alone. + # The caller selects DTR and supplies the already configured port; no USB + # inventory or pySerial import is needed here. + defer_dtr = os.name == "posix" and not port.dtr + previous_dsrdtr = port.dsrdtr + try: + if defer_dtr: + port.dsrdtr = True + port.open() + if defer_dtr: + port.rts = False + port.dtr = False + port.dsrdtr = previous_dsrdtr + except BaseException: + try: + port.close() + finally: + if defer_dtr: + port.dsrdtr = previous_dsrdtr + raise diff --git a/tools/hil/test_contact_cache_serial_stress.py b/tools/hil/test_contact_cache_serial_stress.py new file mode 100644 index 00000000..2e4500b8 --- /dev/null +++ b/tools/hil/test_contact_cache_serial_stress.py @@ -0,0 +1,63 @@ +"""Offline checks for the contact cache collector's serial opening boundary.""" +import os +from types import SimpleNamespace +import unittest +from unittest import mock + +import contact_cache_serial_stress as hil + + +class ContactCacheOpenTests(unittest.TestCase): + @unittest.skipUnless(os.name == "posix", "Requires the POSIX pySerial backend") + def test_actual_posix_open_avoids_reset_and_keeps_requested_session(self): + from test_profile_switch_host import PosixModemSerial + + class BeforeProtocol(Exception): + pass + + master, slave = os.openpty() + try: + for vid, expected in ((0x10C4, False), (0x303A, True)): + with self.subTest(vid=vid): + ports = [] + def factory(*args, **kwargs): + port = PosixModemSerial(*args, **kwargs) + ports.append(port) + return port + + with mock.patch.object(hil.serial, "Serial", side_effect=factory), mock.patch.object( + hil.time, "sleep", side_effect=BeforeProtocol), mock.patch( + "profile_switch.list_ports.comports", return_value=[ + SimpleNamespace(device=os.ttyname(slave), vid=vid)]): + # Stop at the existing initial settle, before any radio + # or protocol request. Exercise the actual collector open. + with self.assertRaises(BeforeProtocol): + hil.run(SimpleNamespace(port=os.ttyname(slave))) + self.assertEqual(len(ports), 1) + port = ports[0] + self.assertFalse(port.has_reset_transition(), port.modem_edges) + self.assertEqual(port.modem_lines, {"dtr": expected, "rts": False}) + self.assertFalse(port.dsrdtr) + self.assertFalse(port.is_open) + finally: + os.close(master) + os.close(slave) + + def test_collector_open_failure_closes_and_restores_flow(self): + from test_profile_switch_host import FailingSessionPort + + for phase in ("open", "rts", "dtr", "restore"): + with self.subTest(phase=phase): + port = FailingSessionPort(phase) + with mock.patch.object(hil.serial, "Serial", return_value=port), mock.patch( + "serial_session.os.name", "posix"), mock.patch( + "profile_switch.list_ports.comports", return_value=[]): + with self.assertRaisesRegex(RuntimeError, phase + " failed"): + hil.run(SimpleNamespace(port="test")) + self.assertFalse(port.is_open) + self.assertFalse(port.dsrdtr) + self.assertEqual(port.close_count, 1) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/hil/test_esp32_companion_serial_stress.py b/tools/hil/test_esp32_companion_serial_stress.py index b0857c65..9236055f 100644 --- a/tools/hil/test_esp32_companion_serial_stress.py +++ b/tools/hil/test_esp32_companion_serial_stress.py @@ -6,6 +6,7 @@ from __future__ import annotations from contextlib import redirect_stdout import io import json +import os import struct import sys import types @@ -214,18 +215,42 @@ class PayloadValidationTest(unittest.TestCase): class StressModesTest(unittest.TestCase): + @unittest.skipUnless(os.name == "posix", "Requires the POSIX pySerial backend") + def test_actual_posix_open_avoids_companion_reset(self) -> None: + try: + import serial + from test_profile_switch_host import PosixModemSerial + except ImportError: + self.skipTest("pySerial is required for the real open-order regression") + master, slave = os.openpty() + try: + with mock.patch.object(serial, "Serial", PosixModemSerial): + port = hil._make_pyserial_factory()(fast_config(port=os.ttyname(slave))) + try: + self.assertFalse(port.has_reset_transition(), port.modem_edges) + self.assertEqual(port.modem_lines, {"dtr": False, "rts": False}) + self.assertFalse(port.dsrdtr) + finally: + port.close() + finally: + os.close(master) + os.close(slave) + def test_selected_control_lines_are_set_while_port_is_closed(self) -> None: class Serial: def __init__(self): self.is_open = False + self.dsrdtr = False + self.events = [] def __setattr__(self, name, value): if name in ("dtr", "rts"): - assert not self.is_open + self.events.append((name, value)) object.__setattr__(self, name, value) def open(self): assert self.rts is False + self.events.append(("open", self.dtr)) self.is_open = True with mock.patch.dict(sys.modules, {"serial": types.SimpleNamespace(Serial=Serial)}): @@ -235,6 +260,43 @@ class StressModesTest(unittest.TestCase): port = factory(fast_config(dtr=dtr)) self.assertIs(port.dtr, dtr) self.assertTrue(port.is_open) + opened = port.events.index(("open", dtr)) + self.assertLess(port.events.index(("dtr", dtr)), opened) + self.assertLess(port.events.index(("rts", False)), opened) + self.assertFalse(port.dsrdtr) + + def test_other_backend_keeps_manual_companion_lines_without_dsr_handoff(self) -> None: + from test_profile_switch_host import SessionPort + + class OtherBackendPort(SessionPort): + def __init__(self): + super().__init__("test") + + def __setattr__(self, name, value): + if name == "dsrdtr" and value: + raise AssertionError("Other backend entered DSR handshaking") + super().__setattr__(name, value) + + with mock.patch.dict(sys.modules, {"serial": types.SimpleNamespace(Serial=OtherBackendPort)}), mock.patch( + "serial_session.os.name", "nt"): + port = hil._make_pyserial_factory()(fast_config()) + self.assertTrue(port.is_open) + self.assertEqual(port.opened_control_lines, (False, False)) + self.assertFalse(port.dsrdtr) + + def test_factory_open_or_release_failure_closes_and_restores_flow(self) -> None: + from test_profile_switch_host import FailingSessionPort + + for phase in ("open", "rts", "dtr", "restore"): + with self.subTest(phase=phase): + port = FailingSessionPort(phase) + with mock.patch.dict(sys.modules, {"serial": types.SimpleNamespace(Serial=lambda: port)}), mock.patch( + "serial_session.os.name", "posix"): + with self.assertRaisesRegex(RuntimeError, phase + " failed"): + hil._make_pyserial_factory()(fast_config()) + self.assertFalse(port.is_open) + self.assertFalse(port.dsrdtr) + self.assertEqual(port.close_count, 1) def test_dtr_cli_opt_in_preserves_default_and_rejects_non_boolean_config(self) -> None: parser = hil.build_argument_parser() diff --git a/tools/hil/test_fem_noise_floor_test.py b/tools/hil/test_fem_noise_floor_test.py index 6f9139ea..169aef20 100644 --- a/tools/hil/test_fem_noise_floor_test.py +++ b/tools/hil/test_fem_noise_floor_test.py @@ -4,8 +4,11 @@ from __future__ import annotations import struct +import os import time import unittest +from unittest import mock +from types import SimpleNamespace import esp32_companion_serial_stress as companion import fem_noise_floor_test as hil @@ -62,17 +65,71 @@ def fast_config(**overrides): class PayloadValidationTest(unittest.TestCase): + @unittest.skipUnless(os.name == "posix", "Requires the POSIX pySerial backend") + def test_actual_posix_open_preserves_selected_fem_dtr_without_reset(self): + try: + import serial + from test_profile_switch_host import PosixModemSerial + except ImportError: + self.skipTest("pySerial is required for the real open-order regression") + master, slave = os.openpty() + try: + for transport_type in (hil.CompanionFemTransport, hil.CliFemTransport): + for selected, vid, expected in (("off", 0x303A, False), ("on", 0x10C4, True), + ("auto", 0x10C4, False), ("auto", 0x1A86, False), + ("auto", 0x303A, True), ("auto", 0x2886, True), + ("auto", 0x239A, True)): + with self.subTest(transport=transport_type.__name__, dtr=selected, vid=vid): + transport = transport_type(fast_config( + port=os.ttyname(slave), dtr=selected, open_delay=0)) + try: + with mock.patch.object(serial, "Serial", PosixModemSerial), mock.patch( + "profile_switch.list_ports.comports", return_value=[ + SimpleNamespace(device=os.ttyname(slave), vid=vid)]): + transport.open() + port = transport.port + self.assertFalse(port.has_reset_transition(), port.modem_edges) + self.assertEqual(port.modem_lines, {"dtr": expected, "rts": False}) + self.assertFalse(port.dsrdtr) + finally: + transport.close() + finally: + os.close(master) + os.close(slave) + + def test_fem_open_or_release_failure_closes_unpublished_port(self): + from test_profile_switch_host import FailingSessionPort + import serial + + for transport_type in (hil.CompanionFemTransport, hil.CliFemTransport): + for phase in ("open", "rts", "dtr", "restore"): + with self.subTest(transport=transport_type.__name__, phase=phase): + port = FailingSessionPort(phase) + transport = transport_type(fast_config(dtr="off", open_delay=0)) + with mock.patch.object(serial, "Serial", return_value=port), mock.patch( + "serial_session.os.name", "posix"): + with self.assertRaisesRegex(RuntimeError, phase + " failed"): + transport.open() + self.assertIsNone(transport.port) + self.assertFalse(port.is_open) + self.assertFalse(port.dsrdtr) + self.assertEqual(port.close_count, 1) + def test_automatic_dtr_matches_transport_session_requirements(self): - self.assertTrue(hil._resolve_dtr(fast_config(protocol="cli"))) - self.assertTrue(hil._resolve_dtr(fast_config( - port="/dev/serial/by-id/usb-Heltec_HT-n5262G_test-if00", - ))) - self.assertFalse(hil._resolve_dtr(fast_config( - port="/dev/serial/by-id/usb-Espressif_USB_JTAG_test-if00", - ))) - self.assertFalse(hil._resolve_dtr(fast_config( - protocol="cli", dtr="off", - ))) + from test_profile_switch_host import SessionPort + + for protocol in ("cli", "companion"): + for vid, expected in ((0x303A, True), (0x2886, True), (0x239A, True), + (0x10C4, False), (0x1A86, False), (None, False)): + with self.subTest(protocol=protocol, vid=vid): + port = SessionPort("test") + with mock.patch("profile_switch.list_ports.comports", return_value=[ + SimpleNamespace(device="test", vid=vid)]): + self.assertEqual(hil._resolve_dtr(fast_config(protocol=protocol), port), expected) + with mock.patch("profile_switch.list_ports.comports") as inventory: + for selected, expected in (("on", True), ("off", False)): + self.assertEqual(hil._resolve_dtr(fast_config(dtr=selected), SessionPort("test")), expected) + inventory.assert_not_called() def test_fem_get_and_set_payloads(self): self.assertTrue(hil.validate_fem_get_response(b"\x00\x01")["enabled"]) diff --git a/tools/hil/test_profile_four_tx.py b/tools/hil/test_profile_four_tx.py index 7940c200..b0acc6a9 100644 --- a/tools/hil/test_profile_four_tx.py +++ b/tools/hil/test_profile_four_tx.py @@ -1,6 +1,7 @@ import ast import copy import json +import importlib.util from pathlib import Path import re import subprocess @@ -15,6 +16,44 @@ from profile_four_tx import check_scan,check_sent,parse_args from profile_four_tx_fixture import TXS,RX,resolve class FourTransmitterTests(unittest.TestCase): + def test_current_collector_opens_with_legacy_and_current_native_dependencies(self): + from test_profile_switch_host import SessionPort + + class NativePort(SessionPort): + def __init__(self): + super().__init__('test') + + def reset_input_buffer(self): + pass + + def legacy_configure(port): + port.dtr=True + port.rts=False + + # The repeat harness injects today's collector into an immutable old + # bundle. That native-only snapshot exports configure_session but not + # open_session. Exercise actual module import and its opening function. + legacy=SimpleNamespace(configure_session=legacy_configure,exchange=lambda *args:None, + exchange_ready=lambda *args:None,read_packet_response=lambda *args:None) + current=sys.modules['profile_switch'] + path=Path(__file__).with_name('profile_four_tx.py') + for dependency in (legacy,current): + with self.subTest(snapshot='legacy' if dependency is legacy else 'current'): + port=NativePort() + with patch.dict(sys.modules,{'profile_switch':dependency}), patch.object( + dependency,'exchange',return_value={'bench':'production-profile-switch-v8','ready':True}), patch( + 'serial.Serial',return_value=port), patch('serial.tools.list_ports.comports',return_value=[ + SimpleNamespace(device='/dev/native_fixture',vid=0x303A)]): + spec=importlib.util.spec_from_file_location('four_tx_repeat_import_test',path) + collector=importlib.util.module_from_spec(spec) + spec.loader.exec_module(collector) + with patch.object(collector,'resolve',return_value='/dev/native_fixture'): + opened,info=collector.open_radio(RX) + self.assertIs(opened,port) + self.assertTrue(info['ready']) + self.assertEqual(port.opened_control_lines,(True,False)) + port.close() + def test_dwell_override_changes_only_requested_visit_time(self): default=parse_args([]) self.assertEqual((default.dwell_symbols,default.dwell_us),(5.1,41780)) diff --git a/tools/hil/test_profile_switch_host.py b/tools/hil/test_profile_switch_host.py index 1a3b4925..3451769c 100644 --- a/tools/hil/test_profile_switch_host.py +++ b/tools/hil/test_profile_switch_host.py @@ -1,10 +1,44 @@ """Regression tests for the serial HIL result reader.""" import unittest +import os from pathlib import Path from tempfile import TemporaryDirectory from unittest.mock import patch from types import SimpleNamespace -from profile_switch import read_response, exchange_ready, configure_session, exchange +from profile_switch import read_response, exchange_ready, configure_session, exchange, open_session + +if os.name == "posix": + import serial.serialposix + + class PosixModemSerial(serial.serialposix.Serial): + """Actual pySerial on a PTY, with only modem ioctls simulated. + + The CP2102 kernel-open state observed in HIL asserts both lines. The + board's reset circuit is driven when DTR is cleared before RTS. + """ + def __init__(self, *args, **kwargs): + self.modem_lines = {"dtr": True, "rts": True} + self.modem_edges = [] + self.flow_at_reconfigure = [] + super().__init__(*args, **kwargs) + + def _reconfigure_port(self, force_update=False): + self.flow_at_reconfigure.append(self.dsrdtr) + super()._reconfigure_port(force_update) + + def _set_modem_line(self, name, value): + self.modem_lines[name] = value + self.modem_edges.append((name, value, self.modem_lines.copy())) + + def _update_dtr_state(self): + self._set_modem_line("dtr", self._dtr_state) + + def _update_rts_state(self): + self._set_modem_line("rts", self._rts_state) + + def has_reset_transition(self): + return any(not lines["dtr"] and lines["rts"] + for _, _, lines in self.modem_edges) class Port: @@ -27,14 +61,106 @@ class SessionPort: self.is_open = False self.dtr = True self.rts = True + self.dsrdtr = False self.opened_control_lines = None + self.close_count = 0 def open(self): self.opened_control_lines = (self.dtr, self.rts) self.is_open = True + def close(self): + self.close_count += 1 + self.is_open = False + + +class FailingSessionPort(SessionPort): + def __init__(self, phase): + super().__init__("test") + self.failure_phase = phase + + def __setattr__(self, name, value): + if getattr(self, "is_open", False): + phase = getattr(self, "failure_phase", None) + if name == phase or (phase == "restore" and name == "dsrdtr" and not value): + raise RuntimeError(phase + " failed") + object.__setattr__(self, name, value) + + def open(self): + if self.failure_phase == "open": + raise RuntimeError("open failed") + super().open() + class ResultReaderTests(unittest.TestCase): + @unittest.skipUnless(os.name == "posix", "Requires the POSIX pySerial backend") + def test_actual_posix_open_avoids_reset_and_preserves_native_dtr(self): + master, slave = os.openpty() + try: + device = os.ttyname(slave) + for vid, expected in ((0x10C4, False), (0x1A86, False), + (0x303A, True), (0x2886, True), (0x239A, True)): + with self.subTest(vid=vid): + port = PosixModemSerial() + port.port = device + try: + with patch("profile_switch.list_ports.comports", return_value=[ + SimpleNamespace(device=device, vid=vid)]): + open_session(port) + self.assertFalse(port.has_reset_transition(), port.modem_edges) + self.assertEqual(port.modem_lines, {"dtr": expected, "rts": False}) + self.assertFalse(port.dsrdtr) + if expected: + self.assertNotIn(True, port.flow_at_reconfigure, + "Native USB entered the bridge handoff") + finally: + port.close() + finally: + os.close(master) + os.close(slave) + + @unittest.skipUnless(os.name == "posix", "Requires the POSIX pySerial backend") + def test_actual_posix_old_open_order_is_a_reset_negative_control(self): + master, slave = os.openpty() + port = PosixModemSerial() + port.port = os.ttyname(slave) + try: + with patch("profile_switch.list_ports.comports", return_value=[]): + configure_session(port) + port.open() # Former caller sequence, using actual pySerial open. + self.assertTrue(port.has_reset_transition(), port.modem_edges) + finally: + port.close() + os.close(master) + os.close(slave) + + def test_other_backend_does_not_use_dsr_handshaking(self): + class OtherBackendPort(SessionPort): + def __setattr__(self, name, value): + if name == "dsrdtr" and value: + raise AssertionError("Other backend entered DSR handshaking") + super().__setattr__(name, value) + + port = OtherBackendPort("test") + with patch("profile_switch.os.name", "nt"), patch( + "profile_switch.list_ports.comports", return_value=[]): + open_session(port) + self.assertTrue(port.is_open) + self.assertEqual(port.opened_control_lines, (False, False)) + self.assertFalse(port.dsrdtr) + + def test_failed_open_or_line_release_closes_and_restores_flow(self): + for phase in ("open", "rts", "dtr", "restore"): + with self.subTest(phase=phase): + port = FailingSessionPort(phase) + with patch("profile_switch.os.name", "posix"), patch( + "profile_switch.list_ports.comports", return_value=[]): + with self.assertRaisesRegex(RuntimeError, phase + " failed"): + open_session(port) + self.assertFalse(port.is_open) + self.assertFalse(port.dsrdtr) + self.assertEqual(port.close_count, 1) + def test_timing_runs_get_correlated_cached_replies(self): port = Port([]) with patch("profile_switch._run_sequences", iter([99])), patch( diff --git a/tools/hil/test_s3_memory_soak.py b/tools/hil/test_s3_memory_soak.py index b28a72f3..bab50dc3 100644 --- a/tools/hil/test_s3_memory_soak.py +++ b/tools/hil/test_s3_memory_soak.py @@ -12,6 +12,28 @@ soak=importlib.util.module_from_spec(spec) spec.loader.exec_module(soak) class TelemetryTests(unittest.TestCase): + @unittest.skipUnless(os.name == 'posix', 'Requires the POSIX pySerial backend') + def test_actual_posix_soak_open_avoids_board_reset(self): + import serial + from test_profile_switch_host import PosixModemSerial + + master,slave=os.openpty() + device=soak.Device('test',os.ttyname(slave),'test') + try: + with patch.object(serial,'Serial',PosixModemSerial), patch.object( + device,'read',return_value=''), patch.object( + device,'_command',return_value='test'): + device.connect() + port=device.stream + self.assertFalse(port.has_reset_transition(),port.modem_edges) + self.assertEqual(port.modem_lines,{'dtr':False,'rts':False}) + self.assertFalse(port.dsrdtr) + self.assertEqual(device.connections,1) + finally: + device.close() + os.close(master) + os.close(slave) + def test_diagnostics_without_key_override_need_no_private_header(self): # Preprocess a clean copy: the developer's ignored key header must not # satisfy an accidental unconditional include during this check. diff --git a/tools/hil/test_serial_session.py b/tools/hil/test_serial_session.py new file mode 100644 index 00000000..1ad034a3 --- /dev/null +++ b/tools/hil/test_serial_session.py @@ -0,0 +1,62 @@ +"""Verify the shared serial opener keeps optional SDK dependencies optional.""" +from pathlib import Path +import ast +import hashlib +import subprocess +import sys +from tempfile import TemporaryDirectory +import unittest + + +class SerialSessionImportTests(unittest.TestCase): + def test_immutable_collectors_require_only_their_shipped_dependencies(self): + base = Path(__file__).parent + fixture_names = ("profile_pair.py", "profile_pair_run.py", "profile_four_tx_fixture.py", + "profile_switch.py", "profile_switch_channels.py", + "profile_switch_packets.py", "profile_switch_sweep.py") + for script, end in (("profile_pair_launch.py", "with (root/'collector.log')"), + ("profile_four_tx_repeat_run.py", "code=base64.b64decode")): + source = (base / script).read_text() + # Execute the actual dependency-selection and hash-validation + # block, without launch, subprocess, firmware or output operations. + block = source[source.index("dependencies="):source.index(end)] + for current, helper in ((False, "absent"), (True, "valid"), + (True, "absent"), (True, "tampered")): + with self.subTest(script=script, current=current, helper=helper), TemporaryDirectory() as tmp: + root = Path(tmp) + for name in fixture_names: + (root / name).write_text("") + (root / "profile_switch.py").write_text( + "from serial_session import open_configured_session\n" if current else "# Legacy snapshot\n") + if helper != "absent": + (root / "serial_session.py").write_text("# Shipped helper\n") + manifest = {"files": {file.name: {"sha256": hashlib.sha256(file.read_bytes()).hexdigest()} + for file in root.iterdir()}} + if helper == "tampered": + (root / "serial_session.py").write_text("# Changed after packaging\n") + namespace = dict(root=root, previous=root, manifest=manifest, + ast=ast, hashlib=hashlib) + if current and helper == "absent": + with self.assertRaises(FileNotFoundError): + exec(block, namespace) + elif helper == "tampered": + with self.assertRaises(RuntimeError): + exec(block, namespace) + else: + exec(block, namespace) + self.assertEqual("serial_session.py" in namespace["dependencies"], current) + + def test_helpers_import_with_only_the_standard_library(self): + # -S removes installed site packages. The companion/FEM/soak modules + # must remain importable until their caller actually requests serial. + program = ("import sys; sys.path.insert(0, " + repr(str(Path(__file__).parent)) + "); " + "import serial_session, esp32_companion_serial_stress, " + "fem_noise_floor_test, s3_memory_soak; " + "assert 'serial' not in sys.modules") + result = subprocess.run([sys.executable, "-I", "-S", "-c", program], + text=True, capture_output=True, timeout=10) + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + + +if __name__ == "__main__": + unittest.main()