Update serial fixtures for atomic Linux console opens

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()

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"))

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)

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)

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)