KrzyszurandClaude Opus 5.5
2b6d43eb45
On macOS/Linux the OS asserts DTR and RTS when the port is opened and pyserial then writes DTR before RTS, passing through RTS=1/DTR=0 - the USB-Serial-JTAG reset state. Every Connect restarted the box. Drop RTS before open and DTR right after it on posix; Windows is unchanged. Verified on macOS 27 / arm64 with box A4:CB:8F:C9:47:3C (fw 1.0.16, build 3ae273ec): connect keeps the box running, Reset & capture still restarts it. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
139 lines
4.6 KiB
Python
139 lines
4.6 KiB
Python
"""SerialManager end-to-end over pyserial's loop:// (TX is looped back to RX)."""
|
|
|
|
import threading
|
|
import time
|
|
|
|
from kota_tech.core.logline import LineKind
|
|
from kota_tech.serialio import manager as mgr_mod
|
|
from kota_tech.serialio.manager import SerialManager, State
|
|
from kota_tech.serialio.ports import PortInfo
|
|
|
|
LOOP = PortInfo("loop://", "loopback", None, None, None)
|
|
|
|
|
|
def _wait(pred, timeout=3.0):
|
|
end = time.monotonic() + timeout
|
|
while time.monotonic() < end:
|
|
if pred():
|
|
return True
|
|
time.sleep(0.02)
|
|
return False
|
|
|
|
|
|
def test_loopback_lines_reset_and_disconnect(monkeypatch):
|
|
monkeypatch.setattr(mgr_mod, "find_port", lambda sn, dev: LOOP)
|
|
lines, states = [], []
|
|
lock = threading.Lock()
|
|
|
|
def on_line(ln):
|
|
with lock:
|
|
lines.append(ln)
|
|
|
|
m = SerialManager(on_line, lambda s, d: states.append(s))
|
|
m.connect(LOOP)
|
|
assert _wait(lambda: m.state == State.CONNECTED)
|
|
|
|
m.write(b"stale tailESP-ROM:esp32s3-20210327\r\nE (5) INFLUX: token: s3cr3t\r\npartial")
|
|
assert _wait(lambda: any(ln.text == "partial" for ln in lines)) # idle flush
|
|
assert not any("stale" in ln.text for ln in lines) # first fragment after open dropped
|
|
m.reset_board()
|
|
assert _wait(lambda: any("reset issued" in ln.text for ln in lines))
|
|
m.disconnect()
|
|
|
|
kinds = [ln.kind for ln in lines]
|
|
assert LineKind.BOOT in kinds
|
|
err = next(ln for ln in lines if ln.level == "E")
|
|
assert "s3cr3t" not in err.text
|
|
assert states[0] == State.CONNECTED and states[-1] == State.DISCONNECTED
|
|
assert lines[-1].text == "--- disconnected ---"
|
|
|
|
|
|
class _RecordingSerial:
|
|
"""Records control-line writes in order, like usbser.sys would see them."""
|
|
|
|
def __init__(self):
|
|
self.calls = []
|
|
self._dtr = self._rts = False
|
|
|
|
@property
|
|
def dtr(self):
|
|
return self._dtr
|
|
|
|
@dtr.setter
|
|
def dtr(self, v):
|
|
self._dtr = v
|
|
self.calls.append(("dtr", v, self._rts)) # DTR write carries both lines
|
|
|
|
@property
|
|
def rts(self):
|
|
return self._rts
|
|
|
|
@rts.setter
|
|
def rts(self, v):
|
|
self._rts = v
|
|
self.calls.append(("rts", v))
|
|
|
|
|
|
def test_hard_reset_pushes_rts_via_dtr_write(monkeypatch):
|
|
monkeypatch.setattr(mgr_mod.time, "sleep", lambda s: None)
|
|
ser = _RecordingSerial()
|
|
SerialManager._hard_reset(ser)
|
|
# Every RTS change must be followed by a DTR write carrying the new RTS,
|
|
# with DTR (GPIO0) low so the chip boots the app, not download mode.
|
|
pushed = [(c[1], c[2]) for c in ser.calls if c[0] == "dtr"]
|
|
assert (False, True) in pushed # reset asserted (RTS=1, DTR=0)
|
|
assert pushed[-1] == (False, False) # released
|
|
assert all(c[1] is False for c in ser.calls if c[0] == "dtr")
|
|
|
|
|
|
class _PosixOpenSerial:
|
|
"""pyserial on macOS/Linux: the OS asserts both lines on open, then
|
|
pyserial writes the requested DTR, then RTS. Records every line state."""
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
self.port = args[0] if args else "fake"
|
|
self.dtr = self.rts = True # pyserial defaults
|
|
self.states = []
|
|
|
|
def __setattr__(self, name, value):
|
|
object.__setattr__(self, name, value)
|
|
if name in ("dtr", "rts") and getattr(self, "is_open", False):
|
|
self.states.append((self.dtr, self.rts))
|
|
|
|
def open(self):
|
|
want_dtr, want_rts = self.dtr, self.rts
|
|
object.__setattr__(self, "dtr", True)
|
|
object.__setattr__(self, "rts", True)
|
|
self.is_open = True
|
|
self.states.append((True, True))
|
|
self.dtr = want_dtr
|
|
self.rts = want_rts
|
|
|
|
|
|
def test_open_on_posix_never_passes_through_reset(monkeypatch):
|
|
monkeypatch.setattr(mgr_mod.sys, "platform", "darwin")
|
|
monkeypatch.setattr(mgr_mod, "find_port", lambda sn, dev: LOOP)
|
|
opened = []
|
|
monkeypatch.setattr(
|
|
mgr_mod.serial, "serial_for_url",
|
|
lambda *a, **k: opened.append(_PosixOpenSerial(*a, **k)) or opened[-1],
|
|
)
|
|
m = SerialManager(lambda ln: None, lambda s, d: None)
|
|
m._target = mgr_mod._Target(LOOP.device, None)
|
|
ser = m._open()
|
|
# (DTR, RTS) = (False, True) resets the chip on connect.
|
|
assert (False, True) not in ser.states
|
|
assert ser.states[-1] == (False, False)
|
|
|
|
|
|
def test_waits_when_port_missing(monkeypatch):
|
|
monkeypatch.setattr(mgr_mod, "find_port", lambda sn, dev: None)
|
|
states, lines = [], []
|
|
m = SerialManager(lines.append, lambda s, d: states.append(s))
|
|
m.connect(LOOP)
|
|
assert _wait(lambda: m.state == State.WAITING)
|
|
m.reset_board()
|
|
assert _wait(lambda: any("ignored" in ln.text for ln in lines))
|
|
m.disconnect()
|
|
assert states == [State.WAITING, State.DISCONNECTED]
|