99 lines
3.1 KiB
Python
99 lines
3.1 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")
|
|
|
|
|
|
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]
|