swaits
e9c4bb3e93
The 1.0 Python client, mirroring the Rust client 1.0 API and the DongLoRa Protocol v2 wire format. * Module split: monolithic protocol.py / client.py replaced with dongle.py (high-level async Dongle), session.py (reader task + tag routing), frame.py (streaming FrameDecoder + encode_frame), commands.py / events.py / errors.py / modulation.py / info.py / crc.py / chip_id (re-export) matching the donglora-protocol Rust crate. * Async API throughout. Serial I/O via asyncio-serial; mux connections via asyncio streams. The blocking client surface is gone. * thiserror-equivalent error hierarchy (DongloraError base with Timeout, TransportClosed, ReaderExited, BadFrame, ErrClientError variants mirroring the Rust ClientError). * Tag-aware dispatch with concurrent in-flight commands: TagAllocator hands out 16-bit device tags, Session routes each response back to the originating coroutine. * Auto-recovery from ERR(ENOTCONFIGURED): the Dongle caches the last SET_CONFIG and silently re-applies on timeout. * 500 ms background keepalive task per PROTOCOL.md §3.4. * Tests rewritten: test_dongle, test_session, test_frame, test_events, test_modulation, test_crc, test_ergonomics; old test_client / test_protocol removed. * Sticky mux reconnect (from the prior fix commit) is preserved: once a mux socket is in use, reconnects commit to it instead of falling back to USB.
338 lines
9.9 KiB
Python
338 lines
9.9 KiB
Python
"""Session-level tests: tag correlation, async events, auto-recovery.
|
|
|
|
Uses a `FakeTransport` — a thread-safe in-memory duplex pipe with a
|
|
simple script of pre-canned firmware responses. Every test exercises
|
|
the real frame codec; only the transport is faked.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
from collections.abc import Callable
|
|
|
|
import pytest
|
|
|
|
from donglora.commands import (
|
|
TYPE_GET_INFO,
|
|
TYPE_PING,
|
|
TYPE_RX_START,
|
|
TYPE_SET_CONFIG,
|
|
TYPE_TX,
|
|
)
|
|
from donglora.errors import (
|
|
BusyError,
|
|
Cancelled,
|
|
ChannelBusy,
|
|
NotConfiguredError,
|
|
)
|
|
from donglora.events import (
|
|
TYPE_ERR,
|
|
TYPE_OK,
|
|
TYPE_RX,
|
|
TYPE_TX_DONE,
|
|
Owner,
|
|
RxEvent,
|
|
RxOrigin,
|
|
SetConfigResult,
|
|
SetConfigResultCode,
|
|
TxDone,
|
|
TxResult,
|
|
)
|
|
from donglora.frame import decode_frame, encode_frame
|
|
from donglora.info import Info, RadioChipId
|
|
from donglora.modulation import LoRaConfig
|
|
from donglora.session import Session
|
|
|
|
# ── A tiny in-memory transport that simulates a device ─────────
|
|
|
|
|
|
class FakeTransport:
|
|
"""Thread-safe in-memory duplex pipe with a scripted device."""
|
|
|
|
def __init__(self, *, script: Callable[[int, int, bytes], list[bytes]] | None = None):
|
|
self._read_buf = bytearray()
|
|
self._read_cond = threading.Condition()
|
|
self._write_buf = bytearray()
|
|
self._write_cond = threading.Condition()
|
|
self._closed = False
|
|
self._script = script or _default_script
|
|
self._writer_thread = threading.Thread(
|
|
target=self._writer_loop,
|
|
name="fake-transport-writer",
|
|
daemon=True,
|
|
)
|
|
self._writer_thread.start()
|
|
self.log: list[bytes] = []
|
|
|
|
def write(self, data: bytes) -> int:
|
|
with self._write_cond:
|
|
self._write_buf.extend(data)
|
|
self._write_cond.notify_all()
|
|
return len(data)
|
|
|
|
def flush(self) -> None:
|
|
pass
|
|
|
|
def read(self, n: int = 1) -> bytes:
|
|
with self._read_cond:
|
|
while not self._read_buf and not self._closed:
|
|
self._read_cond.wait(timeout=0.1)
|
|
if self._closed:
|
|
return b""
|
|
chunk = bytes(self._read_buf[:n])
|
|
del self._read_buf[:n]
|
|
return chunk
|
|
|
|
def close(self) -> None:
|
|
with self._read_cond:
|
|
self._closed = True
|
|
self._read_cond.notify_all()
|
|
with self._write_cond:
|
|
self._write_cond.notify_all()
|
|
|
|
def _writer_loop(self) -> None:
|
|
"""Read framed commands off _write_buf, feed to the script,
|
|
push responses onto _read_buf.
|
|
"""
|
|
while not self._closed:
|
|
with self._write_cond:
|
|
while not self._write_buf and not self._closed:
|
|
self._write_cond.wait(timeout=0.1)
|
|
if self._closed:
|
|
return
|
|
# Collect complete frames (ending in 0x00) from write_buf.
|
|
frames_raw: list[bytes] = []
|
|
while True:
|
|
try:
|
|
idx = self._write_buf.index(0)
|
|
except ValueError:
|
|
break
|
|
frames_raw.append(bytes(self._write_buf[:idx]))
|
|
del self._write_buf[: idx + 1]
|
|
for raw in frames_raw:
|
|
try:
|
|
frame = decode_frame(raw)
|
|
except Exception:
|
|
continue
|
|
self.log.append(raw)
|
|
responses = self._script(frame.type_id, frame.tag, frame.payload)
|
|
if responses:
|
|
with self._read_cond:
|
|
for r in responses:
|
|
self._read_buf.extend(r)
|
|
self._read_cond.notify_all()
|
|
|
|
# Kick an async event (e.g. RX) into the read stream.
|
|
def push_async(self, type_id: int, tag: int, payload: bytes) -> None:
|
|
wire = encode_frame(type_id, tag, payload)
|
|
with self._read_cond:
|
|
self._read_buf.extend(wire)
|
|
self._read_cond.notify_all()
|
|
|
|
|
|
def _default_script(type_id: int, tag: int, payload: bytes) -> list[bytes]: # noqa: ARG001
|
|
"""Reasonable default: ACK every command with a shaped OK."""
|
|
if type_id == TYPE_PING:
|
|
return [encode_frame(TYPE_OK, tag, b"")]
|
|
if type_id == TYPE_GET_INFO:
|
|
info = _make_info()
|
|
return [encode_frame(TYPE_OK, tag, info.encode())]
|
|
if type_id == TYPE_SET_CONFIG:
|
|
result = SetConfigResult(
|
|
result=SetConfigResultCode.APPLIED,
|
|
owner=Owner.MINE,
|
|
current=LoRaConfig.default(),
|
|
)
|
|
return [encode_frame(TYPE_OK, tag, result.encode())]
|
|
if type_id == TYPE_RX_START:
|
|
return [encode_frame(TYPE_OK, tag, b"")]
|
|
if type_id == TYPE_TX:
|
|
td = TxDone(result=TxResult.TRANSMITTED, airtime_us=30_976)
|
|
return [
|
|
encode_frame(TYPE_OK, tag, b""),
|
|
encode_frame(TYPE_TX_DONE, tag, td.encode()),
|
|
]
|
|
return [encode_frame(TYPE_OK, tag, b"")]
|
|
|
|
|
|
def _make_info() -> Info:
|
|
return Info(
|
|
proto_major=1,
|
|
proto_minor=0,
|
|
fw_major=0,
|
|
fw_minor=1,
|
|
fw_patch=0,
|
|
radio_chip_id=RadioChipId.SX1262,
|
|
capability_bitmap=1 | (1 << 16), # LoRa + CAD_BEFORE_TX
|
|
supported_sf_bitmap=0x1FE0,
|
|
supported_bw_bitmap=0x03FF,
|
|
max_payload_bytes=255,
|
|
rx_queue_capacity=32,
|
|
tx_queue_capacity=1,
|
|
freq_min_hz=150_000_000,
|
|
freq_max_hz=960_000_000,
|
|
tx_power_min_dbm=-9,
|
|
tx_power_max_dbm=22,
|
|
mcu_uid=b"\xde\xad\xbe\xef\x01\x23",
|
|
radio_uid=b"",
|
|
)
|
|
|
|
|
|
# ── Session tests ──────────────────────────────────────────────
|
|
|
|
|
|
@pytest.fixture
|
|
def transport():
|
|
t = FakeTransport()
|
|
yield t
|
|
t.close()
|
|
|
|
|
|
def test_session_ping(transport):
|
|
s = Session(transport)
|
|
try:
|
|
s.ping(timeout=2.0)
|
|
finally:
|
|
s.close()
|
|
|
|
|
|
def test_session_get_info_returns_info_struct(transport):
|
|
s = Session(transport)
|
|
try:
|
|
info = s.get_info(timeout=2.0)
|
|
assert isinstance(info, Info)
|
|
assert info.proto_major == 1
|
|
finally:
|
|
s.close()
|
|
|
|
|
|
def test_session_set_config_returns_set_config_result(transport):
|
|
s = Session(transport)
|
|
try:
|
|
result = s.set_config(LoRaConfig.default())
|
|
assert isinstance(result, SetConfigResult)
|
|
assert result.result == SetConfigResultCode.APPLIED
|
|
finally:
|
|
s.close()
|
|
|
|
|
|
def test_session_transmit_returns_tx_done(transport):
|
|
s = Session(transport)
|
|
try:
|
|
td = s.transmit(b"Hello")
|
|
assert td.result == TxResult.TRANSMITTED
|
|
assert td.airtime_us == 30_976
|
|
finally:
|
|
s.close()
|
|
|
|
|
|
def test_session_transmit_channel_busy(transport):
|
|
def script(type_id: int, tag: int, payload: bytes) -> list[bytes]:
|
|
if type_id == TYPE_TX:
|
|
td = TxDone(result=TxResult.CHANNEL_BUSY, airtime_us=0)
|
|
return [
|
|
encode_frame(TYPE_OK, tag, b""),
|
|
encode_frame(TYPE_TX_DONE, tag, td.encode()),
|
|
]
|
|
return _default_script(type_id, tag, payload)
|
|
|
|
transport._script = script
|
|
s = Session(transport)
|
|
try:
|
|
with pytest.raises(ChannelBusy):
|
|
s.transmit(b"x")
|
|
finally:
|
|
s.close()
|
|
|
|
|
|
def test_session_transmit_cancelled(transport):
|
|
def script(type_id: int, tag: int, payload: bytes) -> list[bytes]:
|
|
if type_id == TYPE_TX:
|
|
td = TxDone(result=TxResult.CANCELLED, airtime_us=0)
|
|
return [
|
|
encode_frame(TYPE_OK, tag, b""),
|
|
encode_frame(TYPE_TX_DONE, tag, td.encode()),
|
|
]
|
|
return _default_script(type_id, tag, payload)
|
|
|
|
transport._script = script
|
|
s = Session(transport)
|
|
try:
|
|
with pytest.raises(Cancelled):
|
|
s.transmit(b"x")
|
|
finally:
|
|
s.close()
|
|
|
|
|
|
def test_session_err_not_configured_raises(transport):
|
|
def script(type_id: int, tag: int, payload: bytes) -> list[bytes]:
|
|
if type_id == TYPE_TX:
|
|
return [encode_frame(TYPE_ERR, tag, b"\x03\x00")]
|
|
return _default_script(type_id, tag, payload)
|
|
|
|
transport._script = script
|
|
s = Session(transport)
|
|
try:
|
|
with pytest.raises(NotConfiguredError):
|
|
s.transmit(b"x")
|
|
finally:
|
|
s.close()
|
|
|
|
|
|
def test_session_err_ebusy_raises(transport):
|
|
def script(type_id: int, tag: int, payload: bytes) -> list[bytes]:
|
|
if type_id == TYPE_TX:
|
|
return [encode_frame(TYPE_ERR, tag, b"\x06\x00")]
|
|
return _default_script(type_id, tag, payload)
|
|
|
|
transport._script = script
|
|
s = Session(transport)
|
|
try:
|
|
with pytest.raises(BusyError):
|
|
s.transmit(b"x")
|
|
finally:
|
|
s.close()
|
|
|
|
|
|
def test_session_async_rx_event_arrives_on_queue(transport):
|
|
s = Session(transport)
|
|
try:
|
|
rx = RxEvent(
|
|
rssi_dbm=-73.5,
|
|
snr_db=9.5,
|
|
freq_err_hz=0,
|
|
timestamp_us=100,
|
|
crc_valid=True,
|
|
packets_dropped=0,
|
|
origin=RxOrigin.OTA,
|
|
data=b"\x01\x02\x03",
|
|
)
|
|
transport.push_async(TYPE_RX, 0, rx.encode())
|
|
delivered = s.next_rx(timeout=2.0)
|
|
assert delivered is not None
|
|
assert delivered.data == b"\x01\x02\x03"
|
|
finally:
|
|
s.close()
|
|
|
|
|
|
def test_session_tag_correlation_across_concurrent_sends(transport):
|
|
# Two threads each send a PING. Both must resolve against their own tag.
|
|
s = Session(transport)
|
|
try:
|
|
errors: list[Exception] = []
|
|
|
|
def do_ping() -> None:
|
|
try:
|
|
s.ping(timeout=2.0)
|
|
except Exception as exc:
|
|
errors.append(exc)
|
|
|
|
threads = [threading.Thread(target=do_ping) for _ in range(10)]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join(timeout=3.0)
|
|
assert errors == []
|
|
finally:
|
|
s.close()
|