From de2401c5464341f2127c3f6ca976c3fcd795c353 Mon Sep 17 00:00:00 2001 From: mikecarper Date: Thu, 29 Oct 2026 02:33:12 -0700 Subject: [PATCH] Update serial fixtures for atomic Linux console opens --- test/test_companion_serial_hil.py | 64 ++++++++++++-- tools/hil/test_contact_cache_serial_stress.py | 1 + .../hil/test_esp32_companion_serial_stress.py | 30 ++++--- tools/hil/test_fem_noise_floor_test.py | 3 +- tools/hil/test_profile_switch_host.py | 88 +++++++++++++++++++ 5 files changed, 169 insertions(+), 17 deletions(-) diff --git a/test/test_companion_serial_hil.py b/test/test_companion_serial_hil.py index 4d286a1dd..b05ab50ec 100644 --- a/test/test_companion_serial_hil.py +++ b/test/test_companion_serial_hil.py @@ -1,6 +1,7 @@ """Exercise the shared local USB transport without opening a real device.""" from collections import deque +from contextlib import ExitStack from pathlib import Path import os import struct @@ -41,7 +42,8 @@ class FakeSerial: def __init__(self, **options): self.options = options self.events = [] - self._dtr, self._rts, self._dsrdtr = True, True, False + self._dtr, self._rts, self._dsrdtr, self._rtscts = True, True, False, False + self.driver_dtr, self.driver_rts = True, True self.port = None self.is_open = False self.chunks = deque() @@ -57,11 +59,22 @@ class FakeSerial: def set(self, value): setattr(self, "_" + name, value) self.events.append((name, value)) + if self.is_open and name in ("dtr", "rts"): + getattr(self, "_update_" + name + "_state")() return property(get, set) dtr = _line("dtr") rts = _line("rts") dsrdtr = _line("dsrdtr") + rtscts = _line("rtscts") + + def _update_dtr_state(self): + self.driver_dtr = self.dtr + self.events.append(("driver_dtr", self.driver_dtr)) + + def _update_rts_state(self): + self.driver_rts = self.rts + self.events.append(("driver_rts", self.driver_rts)) @property def in_waiting(self): @@ -72,6 +85,16 @@ class FakeSerial: if self.fail_open: raise OSError("fixture open failure") self.is_open = True + # Match pySerial's independent open-time modem hooks. The Linux + # atomic opener must suppress both until the joint ioctl below. + if not self.dsrdtr: + self._update_dtr_state() + if not self.rtscts: + self._update_rts_state() + self.events.append(("opened_levels", self.driver_dtr, self.driver_rts)) + + def fileno(self): + return 77 def close(self): self.events.append(("close",)) @@ -105,7 +128,30 @@ class CompanionSerialTests(unittest.TestCase): port = port or FakeSerial() module = ModuleType("serial") module.Serial = mock.Mock(return_value=port) - with mock.patch.dict(sys.modules, {"serial": module}): + with ExitStack() as stack: + stack.enter_context(mock.patch.dict(sys.modules, {"serial": module})) + if sys.platform.startswith("linux") and os.name == "posix": + import fcntl + import termios + + def attributes(fd): + self.assertEqual(fd, 77) + return [0, 0, termios.HUPCL | 0x4000, 0, 0, 0, []] + + def clear_hupcl(fd, when, values): + self.assertEqual((fd, when), (77, termios.TCSANOW)) + self.assertEqual(values, [0, 0, 0x4000, 0, 0, 0, []]) + port.events.append(("clear_hupcl",)) + + def clear_lines(fd, request, packed): + self.assertEqual((fd, request), (77, termios.TIOCMBIC)) + self.assertEqual(struct.unpack("I", packed)[0], termios.TIOCM_DTR | termios.TIOCM_RTS) + port.driver_dtr = port.driver_rts = False + port.events.append(("joint_clear",)) + + stack.enter_context(mock.patch.object(termios, "tcgetattr", side_effect=attributes)) + stack.enter_context(mock.patch.object(termios, "tcsetattr", side_effect=clear_hupcl)) + stack.enter_context(mock.patch.object(fcntl, "ioctl", side_effect=clear_lines)) node = HIL.Companion("/dev/fake-local") self.assertEqual(module.Serial.call_args.kwargs, {"port": None, "baudrate": 115200, "timeout": 0.05, @@ -127,9 +173,16 @@ class CompanionSerialTests(unittest.TestCase): self.assertEqual(port.events[:2], [("dtr", False), ("rts", False)]) opened = next(event for event in port.events if event[0] == "open") self.assertEqual(opened[1:3], (False, False)) - if os.name == "posix": + if sys.platform.startswith("linux") and os.name == "posix": + self.assertFalse(opened[3]) + self.assertIn(("opened_levels", True, True), port.events) + self.assertEqual(port.events[-2:], [("clear_hupcl",), ("joint_clear",)]) + self.assertFalse(any(event[0] in ("driver_dtr", "driver_rts") for event in port.events)) + self.assertNotIn("_update_dtr_state", port.__dict__) + self.assertNotIn("_update_rts_state", port.__dict__) + elif os.name == "posix": self.assertTrue(opened[3]) - self.assertEqual(port.events[-3:], [("rts", False), ("dtr", False), ("dsrdtr", False)]) + self.assertFalse(port.driver_dtr or port.driver_rts) self.assertFalse(port.dsrdtr) self.assertFalse(any(event in (("dtr", True), ("rts", True)) for event in port.events)) node.close() @@ -166,7 +219,8 @@ class CompanionSerialTests(unittest.TestCase): HIL.encode_host_frame(bytes((22, 14))), HIL.encode_host_frame(HIL.make_app_start_spec("Meshtext-Auto-HIL").payload)]) stop = port.events.index(("write", writes[0])) - self.assertEqual(port.events[stop - 2:stop], [("rts", False), ("dtr", True)]) + lines_before_stop = [event for event in port.events[:stop] if event[0] in ("dtr", "rts")] + self.assertEqual(lines_before_stop[-2:], [("rts", False), ("dtr", True)]) def test_start_rejects_logging_owner_before_any_framed_request(self): node, port = self.connect() diff --git a/tools/hil/test_contact_cache_serial_stress.py b/tools/hil/test_contact_cache_serial_stress.py index 2e4500b88..385a9b0a6 100644 --- a/tools/hil/test_contact_cache_serial_stress.py +++ b/tools/hil/test_contact_cache_serial_stress.py @@ -51,6 +51,7 @@ class ContactCacheOpenTests(unittest.TestCase): port = FailingSessionPort(phase) with mock.patch.object(hil.serial, "Serial", return_value=port), mock.patch( "serial_session.os.name", "posix"), mock.patch( + "serial_session.sys.platform", "darwin"), mock.patch( "profile_switch.list_ports.comports", return_value=[]): with self.assertRaisesRegex(RuntimeError, phase + " failed"): hil.run(SimpleNamespace(port="test")) diff --git a/tools/hil/test_esp32_companion_serial_stress.py b/tools/hil/test_esp32_companion_serial_stress.py index 9236055fd..a829f6264 100644 --- a/tools/hil/test_esp32_companion_serial_stress.py +++ b/tools/hil/test_esp32_companion_serial_stress.py @@ -237,11 +237,12 @@ class StressModesTest(unittest.TestCase): os.close(slave) def test_selected_control_lines_are_set_while_port_is_closed(self) -> None: - class Serial: + from test_profile_switch_host import SessionPort + + class Serial(SessionPort): def __init__(self): - self.is_open = False - self.dsrdtr = False self.events = [] + super().__init__('test') def __setattr__(self, name, value): if name in ("dtr", "rts"): @@ -251,19 +252,25 @@ class StressModesTest(unittest.TestCase): def open(self): assert self.rts is False self.events.append(("open", self.dtr)) - self.is_open = True + super().open() with mock.patch.dict(sys.modules, {"serial": types.SimpleNamespace(Serial=Serial)}): factory = hil._make_pyserial_factory() for dtr in (False, True): with self.subTest(dtr=dtr): 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) + try: + 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) + if sys.platform.startswith('linux') and not dtr: + self.assertEqual(port.modem_edges, [ + ('joint_clear', False, {'dtr': False, 'rts': False})]) + finally: + port.close() def test_other_backend_keeps_manual_companion_lines_without_dsr_handoff(self) -> None: from test_profile_switch_host import SessionPort @@ -291,7 +298,8 @@ class StressModesTest(unittest.TestCase): 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"): + "serial_session.os.name", "posix"), mock.patch( + "serial_session.sys.platform", "darwin"): with self.assertRaisesRegex(RuntimeError, phase + " failed"): hil._make_pyserial_factory()(fast_config()) self.assertFalse(port.is_open) diff --git a/tools/hil/test_fem_noise_floor_test.py b/tools/hil/test_fem_noise_floor_test.py index 169aef202..43814ce24 100644 --- a/tools/hil/test_fem_noise_floor_test.py +++ b/tools/hil/test_fem_noise_floor_test.py @@ -107,7 +107,8 @@ class PayloadValidationTest(unittest.TestCase): 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"): + "serial_session.os.name", "posix"), mock.patch( + "serial_session.sys.platform", "darwin"): with self.assertRaisesRegex(RuntimeError, phase + " failed"): transport.open() self.assertIsNone(transport.port) diff --git a/tools/hil/test_profile_switch_host.py b/tools/hil/test_profile_switch_host.py index 68138b6e9..8eb0c7d3d 100644 --- a/tools/hil/test_profile_switch_host.py +++ b/tools/hil/test_profile_switch_host.py @@ -1,12 +1,50 @@ """Regression tests for the serial HIL result reader.""" import unittest import os +import sys +from contextlib import ExitStack 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, open_session + +def simulated_native_modem(port, descriptor, *, synthetic_attributes=False): + """Model only unsupported PTY/synthetic-fd modem ioctls, not the opener.""" + import fcntl + import struct + import termios + + stack = ExitStack() + actual_ioctl = fcntl.ioctl + + def ioctl(fd, request, value, *args, **kwargs): + if fd != descriptor or request != termios.TIOCMBIC: + return actual_ioctl(fd, request, value, *args, **kwargs) + mask = struct.unpack('I', value)[0] + if mask != termios.TIOCM_DTR | termios.TIOCM_RTS: + raise AssertionError('modem lines were not cleared together') + port.modem_lines.update(dtr=False, rts=False) + port.modem_edges.append(('joint_clear', False, port.modem_lines.copy())) + return 0 + + stack.enter_context(patch.object(fcntl, 'ioctl', side_effect=ioctl)) + if synthetic_attributes: + def attributes(fd): + if fd != descriptor: + raise AssertionError('unexpected synthetic descriptor') + return [0, 0, termios.HUPCL | 0x4000, 0, 0, 0, []] + + def set_attributes(fd, when, values): + if fd != descriptor or when != termios.TCSANOW or values[2] & termios.HUPCL: + raise AssertionError('HUPCL was not cleared on the selected descriptor') + + stack.enter_context(patch.object(termios, 'tcgetattr', side_effect=attributes)) + stack.enter_context(patch.object(termios, 'tcsetattr', side_effect=set_attributes)) + return stack + + if os.name == "posix": import serial.serialposix @@ -20,8 +58,24 @@ if os.name == "posix": self.modem_lines = {"dtr": True, "rts": True} self.modem_edges = [] self.flow_at_reconfigure = [] + self._native_fixture = None super().__init__(*args, **kwargs) + def open(self): + super().open() + # Real pySerial/termios still configure a real PTY. Only the + # modem ioctl that PTYs cannot implement is simulated afterward. + self._native_fixture = simulated_native_modem(self, self.fileno()) + + def close(self): + try: + super().close() + finally: + fixture = getattr(self, '_native_fixture', None) + if fixture is not None: + fixture.close() + self._native_fixture = None + def _reconfigure_port(self, force_update=False): self.flow_at_reconfigure.append(self.dsrdtr) super()._reconfigure_port(force_update) @@ -62,16 +116,44 @@ class SessionPort: self.dtr = True self.rts = True self.dsrdtr = False + self.rtscts = False self.opened_control_lines = None self.close_count = 0 + self.modem_lines = {'dtr': True, 'rts': True} + self.modem_edges = [] + self._native_fixture = None + + def _update_dtr_state(self): + self.modem_lines['dtr'] = self.dtr + self.modem_edges.append(('dtr', self.dtr, self.modem_lines.copy())) + + def _update_rts_state(self): + self.modem_lines['rts'] = self.rts + self.modem_edges.append(('rts', self.rts, self.modem_lines.copy())) + + def fileno(self): + return 77 def open(self): + # Only the atomic opener requests native descriptor operations after + # open. Direct fake opens need no global native-function patch at all. + deferred = all(name in self.__dict__ for name in + ('_update_dtr_state', '_update_rts_state')) + if sys.platform.startswith('linux') and os.name == 'posix' and deferred: + self._native_fixture = simulated_native_modem(self, self.fileno(), synthetic_attributes=True) self.opened_control_lines = (self.dtr, self.rts) self.is_open = True + if not self.dsrdtr: + self._update_dtr_state() + if not self.rtscts: + self._update_rts_state() def close(self): self.close_count += 1 self.is_open = False + if self._native_fixture is not None: + self._native_fixture.close() + self._native_fixture = None class FailingSessionPort(SessionPort): @@ -111,6 +193,11 @@ class ResultReaderTests(unittest.TestCase): self.assertFalse(port.has_reset_transition(), port.modem_edges) self.assertEqual(port.modem_lines, {"dtr": expected, "rts": False}) self.assertFalse(port.dsrdtr) + if sys.platform.startswith('linux') and not expected: + self.assertEqual(port.modem_edges, [ + ('joint_clear', False, {'dtr': False, 'rts': False})]) + self.assertNotIn(True, port.flow_at_reconfigure, + 'Atomic Linux open enabled flow control') if expected: self.assertNotIn(True, port.flow_at_reconfigure, "Native USB entered the bridge handoff") @@ -155,6 +242,7 @@ class ResultReaderTests(unittest.TestCase): with self.subTest(phase=phase): port = FailingSessionPort(phase) with patch("profile_switch.os.name", "posix"), patch( + "serial_session.sys.platform", "darwin"), patch( "profile_switch.list_ports.comports", return_value=[]): with self.assertRaisesRegex(RuntimeError, phase + " failed"): open_session(port)