Open Linux serial consoles without pulsing ESP32 BOOT

test/test_ethernet_hil_serial_open.py

@@ -6,6 +6,7 @@ import io
 import os
 from pathlib import Path
 import subprocess
+import struct
 import sys
 from types import ModuleType, SimpleNamespace
 import unittest
@@ -22,6 +23,7 @@ import serial_session as session
 class FakeSerial:
     def __init__(self, **kwargs):
         self.dsrdtr = False
+        self.rtscts = False
         self._dtr = True
         self._rts = True
         self.driver_rts = True
@@ -40,7 +42,7 @@ class FakeSerial:
     def dtr(self, value):
         self._dtr = value
         if self.is_open:
-            self._apply_dtr(value)
+            self._update_dtr_state()
 
     @property
     def rts(self):
@@ -50,8 +52,14 @@ class FakeSerial:
     def rts(self, value):
         self._rts = value
         if self.is_open:
-            self.driver_rts = value
-            self.trace.append(("rts", value))
+            self._update_rts_state()
+
+    def _update_dtr_state(self):
+        self._apply_dtr(self._dtr)
+
+    def _update_rts_state(self):
+        self.driver_rts = self._rts
+        self.trace.append(("rts", self._rts))
 
     def _apply_dtr(self, value):
         if not value and self.driver_rts:
@@ -62,10 +70,12 @@ class FakeSerial:
     def open(self):
         self.opens += 1
         self.is_open = True
+        FakeSerial.live = self
         # POSIX pySerial's real open order: DTR before RTS unless deferred.
         if not self.dsrdtr:
-            self._apply_dtr(self._dtr)
-        self.rts = self._rts
+            self._update_dtr_state()
+        if not self.rtscts:
+            self._update_rts_state()
 
     def close(self):
         self.is_open = False
@@ -97,6 +107,22 @@ def serial_modules(port):
 @unittest.skipUnless(os.name == "posix", "POSIX CH340 pin-order behavior")
 class EthernetHilSerialTests(unittest.TestCase):
     def setUp(self):
+        import fcntl
+        import termios
+
+        def clear_together(fd, request, mask):
+            self.assertEqual(fd, 77)
+            self.assertEqual(request, termios.TIOCMBIC)
+            self.assertEqual(struct.unpack("I", mask)[0], termios.TIOCM_DTR | termios.TIOCM_RTS)
+            port = FakeSerial.live
+            port.driver_dtr = port.driver_rts = False
+            port.trace.append(("modem", False, False))
+
+        for inspector in (patch.object(fcntl, "ioctl", side_effect=clear_together),
+                          patch.object(termios, "tcgetattr", return_value=[0, 0, termios.HUPCL, 0, 0, 0, []]),
+                          patch.object(termios, "tcsetattr")):
+            inspector.start()
+            self.addCleanup(inspector.stop)
         # Exercise the real Linux ownership guard without consulting host
         # fuser or a nonexistent synthetic port. A quiet rc1 is the exact
         # successful no-owner contract; diagnostic output must still refuse.
@@ -112,7 +138,7 @@ class EthernetHilSerialTests(unittest.TestCase):
     def assert_safe(self, port):
         self.assertEqual(port.opens, 1)
         self.assertEqual(port.resets, 0)
-        self.assertEqual(port.trace[0], ("rts", False))
+        self.assertEqual(port.trace, [("modem", False, False)])
         self.assertFalse(port.dsrdtr)
         self.assertFalse(port.driver_rts)
         self.assertFalse(port.driver_dtr)

tools/hil/serial_session.py

@@ -81,7 +81,48 @@ def assert_port_idle(path):
         raise RuntimeError("Cannot safely inspect serial-port ownership (fuser error)")
 
 
+def _open_linux_inactive_modem_session(port):
+    # pySerial updates DTR and RTS separately during open. Releasing either
+    # first can drive EN or BOOT on an ESP32 auto-reset circuit. Keep both
+    # kernel-open levels together until one joint modem-line update can clear
+    # them. Do not enable flow control to suppress the individual updates.
+    import fcntl
+    import struct
+    import termios
+
+    missing = object()
+    saved = {}
+    for name in ("_update_dtr_state", "_update_rts_state"):
+        if not callable(getattr(port, name, None)):
+            raise RuntimeError("Linux serial backend cannot defer modem-line updates")
+        saved[name] = port.__dict__.get(name, missing)
+    try:
+        for name in saved:
+            setattr(port, name, lambda: None)
+        port.open()
+        descriptor = port.fileno()
+        attributes = termios.tcgetattr(descriptor)
+        attributes[2] &= ~termios.HUPCL
+        termios.tcsetattr(descriptor, termios.TCSANOW, attributes)
+        fcntl.ioctl(descriptor, termios.TIOCMBIC,
+                    struct.pack("I", termios.TIOCM_DTR | termios.TIOCM_RTS))
+    except BaseException:
+        port.close()
+        raise
+    finally:
+        for name, previous in saved.items():
+            if previous is missing:
+                delattr(port, name)
+            else:
+                setattr(port, name, previous)
+
+
 def open_configured_session(port):
+    if (sys.platform.startswith("linux") and os.name == "posix"
+            and not port.dtr and not port.rts
+            and not port.dsrdtr and not port.rtscts):
+        _open_linux_inactive_modem_session(port)
+        return
     # 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

tools/hil/test_serial_session.py

@@ -3,6 +3,8 @@ from pathlib import Path
 import ast
 import hashlib
 import importlib.util
+import os
+import struct
 import subprocess
 import sys
 from tempfile import TemporaryDirectory
@@ -47,6 +49,153 @@ class SerialPortOwnershipTests(unittest.TestCase):
             self.inspect(subprocess.CompletedProcess([], 0, stdout="1234\n", stderr="/dev/ttyACM7:"))
 
 
+@unittest.skipUnless(os.name == "posix", "POSIX modem-line control")
+class AtomicModemOpenTests(unittest.TestCase):
+    class Port:
+        dtr = rts = dsrdtr = rtscts = False
+
+        def __init__(self):
+            self.is_open = False
+            self.events = []
+            self.driver_dtr = self.driver_rts = True
+
+        def _update_dtr_state(self):
+            self.events.append("individual_dtr")
+            self.driver_dtr = self.dtr
+
+        def _update_rts_state(self):
+            self.events.append("individual_rts")
+            self.driver_rts = self.rts
+
+        def open(self):
+            self.events.append("open")
+            self.is_open = True
+            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 close(self):
+            self.events.append("close")
+            self.is_open = False
+
+        def fileno(self):
+            return 77
+
+    def setUp(self):
+        spec = importlib.util.spec_from_file_location(
+            "atomic_serial_session", Path(__file__).with_name("serial_session.py"))
+        self.session = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(self.session)
+
+    def exercise(self, port, fail=None):
+        import fcntl
+        import termios
+
+        def get_attributes(fd):
+            self.assertEqual(fd, 77)
+            port.events.append("attributes")
+            if fail == "attributes":
+                raise OSError("attributes failed")
+            return [1, 2, termios.HUPCL | 0x4000, 4, 5, 6, [7]]
+
+        def set_attributes(fd, when, attributes):
+            self.assertEqual((fd, when), (77, termios.TCSANOW))
+            self.assertEqual(attributes, [1, 2, 0x4000, 4, 5, 6, [7]])
+            port.events.append("clear_hupcl")
+            if fail == "set_attributes":
+                raise OSError("set_attributes failed")
+
+        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.events.append("joint_clear")
+            if fail == "ioctl":
+                raise OSError("ioctl failed")
+            port.driver_dtr = port.driver_rts = False
+
+        with patch.object(self.session.sys, "platform", "linux"), \
+                patch.object(termios, "tcgetattr", side_effect=get_attributes), \
+                patch.object(termios, "tcsetattr", side_effect=set_attributes), \
+                patch.object(fcntl, "ioctl", side_effect=clear_lines) as modem:
+            self.session.open_configured_session(port)
+        return modem
+
+    def test_joint_clear_has_no_intermediate_reset_or_boot_levels(self):
+        port = self.Port()
+        modem = self.exercise(port)
+        self.assertEqual(port.events, ["open", ("opened_levels", True, True),
+                                      "attributes", "clear_hupcl", "joint_clear"])
+        modem.assert_called_once()
+        self.assertFalse(port.driver_dtr or port.driver_rts)
+        self.assertFalse(port.dsrdtr or port.rtscts)
+        self.assertNotIn("_update_dtr_state", port.__dict__)
+        self.assertNotIn("_update_rts_state", port.__dict__)
+        port._update_dtr_state()
+        port._update_rts_state()
+        self.assertEqual(port.events[-2:], ["individual_dtr", "individual_rts"])
+
+    def test_prior_dtr_only_deferral_has_an_intermediate_boot_level(self):
+        port = self.Port()
+        port.dsrdtr = True
+        port.open()
+        self.assertEqual(port.events[-1], ("opened_levels", True, False))
+
+    def test_all_failure_stages_close_and_restore_instance_overrides(self):
+        for fail in ("open", "attributes", "set_attributes", "ioctl"):
+            with self.subTest(stage=fail):
+                port = self.Port()
+                original_dtr = lambda: port.events.append("original_dtr")
+                original_rts = lambda: port.events.append("original_rts")
+                port._update_dtr_state = original_dtr
+                port._update_rts_state = original_rts
+                if fail == "open":
+                    def failed_open():
+                        port.is_open = True
+                        raise OSError("open failed")
+                    port.open = failed_open
+                with self.assertRaisesRegex(OSError, fail + " failed"):
+                    self.exercise(port, fail)
+                self.assertFalse(port.is_open)
+                self.assertEqual(port.events[-1], "close")
+                self.assertIs(port._update_dtr_state, original_dtr)
+                self.assertIs(port._update_rts_state, original_rts)
+
+    def test_true_dtr_and_other_platforms_retain_their_existing_opener(self):
+        for platform, dtr in (("linux", True), ("win32", False), ("darwin", False)):
+            with self.subTest(platform=platform, dtr=dtr):
+                port = self.Port()
+                port.dtr = dtr
+                with patch.object(self.session.sys, "platform", platform), \
+                        patch.object(self.session, "_open_linux_inactive_modem_session") as atomic:
+                    self.session.open_configured_session(port)
+                atomic.assert_not_called()
+                self.assertTrue(port.is_open)
+                self.assertEqual(port.dtr, dtr)
+
+    def test_installed_pyserial_open_uses_the_deferred_methods(self):
+        try:
+            import serial.serialposix as backend
+        except ImportError:
+            self.skipTest("installed pySerial backend required")
+        port = self.Port()
+        port._port = port.portstr = "/dev/test-only"
+        port._dsrdtr = port._rtscts = False
+        port._reconfigure_port = lambda **kwargs: port.events.append("configure")
+        port._reset_input_buffer = lambda: port.events.append("reset_input")
+        port.open = lambda: backend.Serial.open(port)
+        with patch.object(backend.os, "open", return_value=77), \
+                patch.object(backend.os, "pipe", side_effect=[(80, 81), (82, 83)]), \
+                patch.object(backend.fcntl, "fcntl"):
+            self.exercise(port)
+        self.assertNotIn("individual_dtr", port.events)
+        self.assertNotIn("individual_rts", port.events)
+        self.assertTrue(port.is_open)
+        self.assertEqual(port.events, ["configure", "reset_input", "attributes",
+                                      "clear_hupcl", "joint_clear"])
+
+
 class SerialSessionImportTests(unittest.TestCase):
     def test_immutable_collectors_require_only_their_shipped_dependencies(self):
         base = Path(__file__).parent