swaits
b15b661c2d
The 1.0 Python mux, mirroring the Rust donglora-mux 1.0 one-for-one and implementing the full §13 multi-client semantics on the DongLoRa Protocol v2 wire format.
* Structural rewrite. donglora_mux/{daemon,intercept,session,tagmap}.py replace the 601-line __init__.py monolith.
* Tag-aware dispatch. TagMapper with skip-zero allocation and per-client bulk drop on disconnect. Multiple clients can have commands in flight concurrently; the single-oneshot response slot is gone.
* Full §13.2 multi-client SET_CONFIG semantics. Tracks (client_id, modulation) lock owner and synthesizes OK(APPLIED/MINE), OK(ALREADY_MATCHED/OTHER), or OK(LOCKED_MISMATCH/OTHER).
* TX loopback fan-out (§13.4). On TX_DONE(TRANSMITTED) the mux synthesizes an RX frame with origin = LOCAL_LOOPBACK for every connected client except the transmitter.
* Async ERR(EFRAME) fan-out: tag-0 framing errors are broadcast to all clients.
* Per-client TX cache on ClientSession so the loopback fan-out does not need the sender to re-buffer.
* Socket and serial fcntl.flock to prevent multiple mux instances from fighting over the same dongle or socket path.
* 500 ms keepalive matching PROTOCOL.md §3.4 (2x margin on the 1 s inactivity timer), replacing the 5 s cadence from 0.x.
* Clean SIGINT / SIGTERM shutdown: per-client handler tasks are tracked and cancelled, so blocked reader.read() calls no longer pin the event loop open.
215 lines
7.1 KiB
Python
215 lines
7.1 KiB
Python
"""Integration tests for :class:`donglora_mux.daemon.MuxDaemon`.
|
|
|
|
Drives the daemon's dispatch logic directly without spinning up a real
|
|
serial port — we push synthetic dongle frames through
|
|
``_dispatch_device_frame`` and inspect what ends up in each client
|
|
session's queue.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
|
|
from donglora.commands import (
|
|
TYPE_PING,
|
|
TYPE_SET_CONFIG,
|
|
TYPE_TX,
|
|
encode_set_config_payload,
|
|
encode_tx_payload,
|
|
)
|
|
from donglora.events import (
|
|
TYPE_OK,
|
|
TYPE_RX,
|
|
TYPE_TX_DONE,
|
|
Owner,
|
|
RxEvent,
|
|
RxOrigin,
|
|
SetConfigResult,
|
|
SetConfigResultCode,
|
|
TxDone,
|
|
TxResult,
|
|
)
|
|
from donglora.frame import Frame, decode_frame, encode_frame
|
|
from donglora.modulation import LoRaBandwidth, LoRaCodingRate, LoRaConfig
|
|
from donglora_mux.daemon import MuxDaemon
|
|
from donglora_mux.session import ClientSession
|
|
from donglora_mux.tagmap import TagMapper
|
|
|
|
|
|
def _lora(freq: int = 910_525_000) -> LoRaConfig:
|
|
return LoRaConfig(
|
|
freq_hz=freq,
|
|
sf=7,
|
|
bw=LoRaBandwidth.KHZ_62_5,
|
|
cr=LoRaCodingRate.CR_4_5,
|
|
preamble_len=16,
|
|
sync_word=0x1424,
|
|
tx_power_dbm=20,
|
|
)
|
|
|
|
|
|
def _make_daemon() -> MuxDaemon:
|
|
# Construct without touching filesystem / serial — we're exercising
|
|
# the dispatch surface directly.
|
|
return MuxDaemon("/dev/null", "/tmp/test-mux.sock", tcp_addr=None)
|
|
|
|
|
|
def _drain_queue(session: ClientSession) -> list[Frame]:
|
|
"""Non-blocking: drain whatever is currently in the session's queue."""
|
|
out: list[Frame] = []
|
|
while True:
|
|
try:
|
|
raw = session._queue.get_nowait() # type: ignore[attr-defined]
|
|
except asyncio.QueueEmpty:
|
|
return out
|
|
# Strip trailing 0x00 sentinel before decode_frame.
|
|
assert raw.endswith(b"\x00")
|
|
out.append(decode_frame(raw[:-1]))
|
|
|
|
|
|
@pytest.fixture
|
|
def daemon():
|
|
d = _make_daemon()
|
|
# Two sessions, manually added.
|
|
a = ClientSession()
|
|
b = ClientSession()
|
|
d.sessions[a.id] = a
|
|
d.sessions[b.id] = b
|
|
yield d, a, b
|
|
|
|
|
|
async def test_dispatch_routes_tagged_response_to_owning_client(daemon):
|
|
d, a, b = daemon
|
|
# A issued a PING with upstream tag 0x0042.
|
|
device_tag = d.tagmap.alloc(a.id, 0x0042, TYPE_PING)
|
|
# Dongle replies with OK on the device_tag.
|
|
await d._dispatch_device_frame(Frame(type_id=TYPE_OK, tag=device_tag, payload=b""))
|
|
|
|
a_frames = _drain_queue(a)
|
|
b_frames = _drain_queue(b)
|
|
assert len(a_frames) == 1
|
|
assert a_frames[0].type_id == TYPE_OK
|
|
assert a_frames[0].tag == 0x0042
|
|
assert b_frames == [] # Only A gets the routed response.
|
|
|
|
|
|
async def test_dispatch_fans_out_async_rx_to_all_clients(daemon):
|
|
d, a, b = daemon
|
|
rx = RxEvent(
|
|
rssi_dbm=-80.0,
|
|
snr_db=7.5,
|
|
freq_err_hz=0,
|
|
timestamp_us=1_000_000,
|
|
crc_valid=True,
|
|
packets_dropped=0,
|
|
origin=RxOrigin.OTA,
|
|
data=b"hello",
|
|
)
|
|
await d._dispatch_device_frame(Frame(type_id=TYPE_RX, tag=0, payload=rx.encode()))
|
|
|
|
a_frames = _drain_queue(a)
|
|
b_frames = _drain_queue(b)
|
|
assert len(a_frames) == 1
|
|
assert len(b_frames) == 1
|
|
assert a_frames[0].type_id == TYPE_RX
|
|
assert a_frames[0].tag == 0
|
|
|
|
|
|
async def test_set_config_applied_installs_lock(daemon):
|
|
d, a, _b = daemon
|
|
cfg = _lora()
|
|
device_tag = d.tagmap.alloc(a.id, 0x0099, TYPE_SET_CONFIG)
|
|
result = SetConfigResult(result=SetConfigResultCode.APPLIED, owner=Owner.MINE, current=cfg)
|
|
await d._dispatch_device_frame(Frame(type_id=TYPE_OK, tag=device_tag, payload=result.encode()))
|
|
# Lock should now be installed with A as owner and the applied cfg.
|
|
assert d.state.locked is not None
|
|
owner_id, locked_cfg = d.state.locked
|
|
assert owner_id == a.id
|
|
assert locked_cfg.freq_hz == cfg.freq_hz
|
|
|
|
|
|
async def test_tx_done_transmitted_loopbacks_to_peers(daemon):
|
|
d, a, b = daemon
|
|
# A sent a TX of "hi" with upstream tag 0x00AA.
|
|
payload = encode_tx_payload(b"hi")
|
|
# Stash like the writer would.
|
|
a.tx_cache[0x00AA] = payload[1:] # strip flags byte
|
|
device_tag = d.tagmap.alloc(a.id, 0x00AA, TYPE_TX)
|
|
# Absorb the intermediate OK (mapping kept, tx_cache still populated).
|
|
await d._dispatch_device_frame(Frame(type_id=TYPE_OK, tag=device_tag, payload=b""))
|
|
# Terminal TX_DONE(TRANSMITTED).
|
|
td = TxDone(result=TxResult.TRANSMITTED, airtime_us=5_000)
|
|
await d._dispatch_device_frame(Frame(type_id=TYPE_TX_DONE, tag=device_tag, payload=td.encode()))
|
|
|
|
a_frames = _drain_queue(a)
|
|
b_frames = _drain_queue(b)
|
|
|
|
# A got OK (intermediate) + TX_DONE.
|
|
assert [f.type_id for f in a_frames] == [TYPE_OK, TYPE_TX_DONE]
|
|
assert all(f.tag == 0x00AA for f in a_frames)
|
|
# B got a loopback RX (tag=0, origin=LOCAL_LOOPBACK, data="hi").
|
|
assert len(b_frames) == 1
|
|
assert b_frames[0].type_id == TYPE_RX
|
|
assert b_frames[0].tag == 0
|
|
decoded = RxEvent # noqa: F841 — just a hint for readers
|
|
# Inline decode (RxEvent.decode if available, else pull fields from the wire payload).
|
|
buf = b_frames[0].payload
|
|
assert buf[19] == int(RxOrigin.LOCAL_LOOPBACK)
|
|
assert buf[20:] == b"hi"
|
|
|
|
|
|
async def test_disconnect_releases_lock(daemon):
|
|
d, a, _b = daemon
|
|
d.state.locked = (a.id, _lora())
|
|
# Simulate the post-writer-cancel cleanup step in _remove_client.
|
|
await d._remove_client(a)
|
|
assert d.state.locked is None
|
|
|
|
|
|
async def test_concurrent_tagged_commands_route_independently(daemon):
|
|
"""Two concurrent PINGs from different clients must correlate correctly."""
|
|
d, a, b = daemon
|
|
tag_a = d.tagmap.alloc(a.id, 0x0001, TYPE_PING)
|
|
tag_b = d.tagmap.alloc(b.id, 0x0001, TYPE_PING)
|
|
# Dongle replies for B first, then A.
|
|
await d._dispatch_device_frame(Frame(type_id=TYPE_OK, tag=tag_b, payload=b""))
|
|
await d._dispatch_device_frame(Frame(type_id=TYPE_OK, tag=tag_a, payload=b""))
|
|
a_frames = _drain_queue(a)
|
|
b_frames = _drain_queue(b)
|
|
assert len(a_frames) == 1
|
|
assert len(b_frames) == 1
|
|
# Each sees their own upstream tag, not the mux's internal device tag.
|
|
assert a_frames[0].tag == 0x0001
|
|
assert b_frames[0].tag == 0x0001
|
|
|
|
|
|
async def test_unsolicited_tagged_frame_is_dropped(daemon):
|
|
"""A frame with a tag that isn't in the pending map should be ignored,
|
|
not routed to a random client or crash the dispatcher."""
|
|
d, a, b = daemon
|
|
# Tag 9999 was never allocated.
|
|
await d._dispatch_device_frame(Frame(type_id=TYPE_OK, tag=9999, payload=b""))
|
|
assert _drain_queue(a) == []
|
|
assert _drain_queue(b) == []
|
|
|
|
|
|
def test_tag_mapper_isolation(daemon):
|
|
"""Regression check: tagmap should start empty per daemon instance."""
|
|
d, _a, _b = daemon
|
|
assert isinstance(d.tagmap, TagMapper)
|
|
assert len(d.tagmap) == 0
|
|
|
|
|
|
def test_encode_decode_roundtrip_sanity():
|
|
"""Sanity check: encode_frame + strip sentinel + decode_frame are inverses
|
|
for the shapes we push through the mux."""
|
|
payload = encode_set_config_payload(_lora())
|
|
wire = encode_frame(TYPE_SET_CONFIG, 42, payload)
|
|
assert wire.endswith(b"\x00")
|
|
frame = decode_frame(wire[:-1])
|
|
assert frame.type_id == TYPE_SET_CONFIG
|
|
assert frame.tag == 42
|
|
assert frame.payload == payload
|