From cfc9d44bc94ff039007b78ac3df51f4a455e735b Mon Sep 17 00:00:00 2001 From: mikecarper Date: Thu, 29 Oct 2026 01:19:12 -0700 Subject: [PATCH] Open Linux serial consoles without pulsing ESP32 BOOT --- test/test_ethernet_hil_serial_open.py | 38 +++++-- tools/hil/serial_session.py | 41 +++++++ tools/hil/test_serial_session.py | 149 ++++++++++++++++++++++++++ 3 files changed, 222 insertions(+), 6 deletions(-) diff --git a/test/test_ethernet_hil_serial_open.py b/test/test_ethernet_hil_serial_open.py index 43f344894..ad422a125 100644 --- a/test/test_ethernet_hil_serial_open.py +++ b/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) diff --git a/tools/hil/serial_session.py b/tools/hil/serial_session.py index a1fc3196e..6ed0e032e 100644 --- a/tools/hil/serial_session.py +++ b/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 diff --git a/tools/hil/test_serial_session.py b/tools/hil/test_serial_session.py index 8f3084266..810661a26 100644 --- a/tools/hil/test_serial_session.py +++ b/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