Alec Perkins
33bf287c30
All checks were successful
tests / pytest (push) Successful in 1m24s
Two related changes to make sender/receiver logs analyzable:
1. MeshCoreRadio.send no longer returns until the radio firmware
reports the packet was actually transmitted. send_chan_msg only
waits for EventType.OK (firmware accepted bytes) — no
tx-complete event is pushed for channel messages, so we poll
commands.get_stats_packets() and wait for nb_sent to strictly
increment past a pre-send baseline. Times out after 30s.
2. Each test packet now embeds a random 8-hex-char token as the
4th comma-separated field of the prefix, distinct from any
radio-assigned packet id. Receivers extract it from the text
and write it on every TSV row, so a sender's queued/sent rows
tie 1:1 to each receiver's record without depending on
whatever packet id the radio surfaces on each side.
TSV gains a new column (3rd, after packet_id):
queued: ts queued <id> <token> <freq,bw,sf,cr,pow>
sent: ts sent <id> <token> <duration_ms> <text>
received: ts received <id> <token> <text>
Missing token writes "-".
UDP simulator regex updated to accept both legacy 3-field test
payloads and the new 4-field format with token.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
51 lines
1.7 KiB
Python
51 lines
1.7 KiB
Python
from modjam import payload
|
|
|
|
|
|
def test_make_payload_pads_to_size():
|
|
text, token = payload.make_test_payload(size=80, seq=1, elapsed=0.5, ts=1234.5)
|
|
assert len(text) == 80
|
|
assert text.startswith("1234.5,0.5,1,")
|
|
assert f",{token}|" in text
|
|
|
|
|
|
def test_make_payload_short_when_prefix_exceeds_size():
|
|
text, token = payload.make_test_payload(size=5, seq=99, elapsed=12.34, ts=1234567890.0)
|
|
# prefix already > 5; payload returns prefix as-is
|
|
assert text.endswith("|")
|
|
assert token in text
|
|
|
|
|
|
def test_token_is_hex_and_correct_length():
|
|
_, token = payload.make_test_payload(size=80, seq=1, elapsed=0.0, ts=0.0)
|
|
assert len(token) == payload.TOKEN_LEN
|
|
assert all(c in "0123456789abcdef" for c in token)
|
|
|
|
|
|
def test_tokens_are_unique():
|
|
tokens = {payload.make_test_payload(80, 1, 0, 0)[1] for _ in range(50)}
|
|
assert len(tokens) == 50 # vanishingly unlikely to collide
|
|
|
|
|
|
def test_extract_token_roundtrip():
|
|
text, token = payload.make_test_payload(size=120, seq=42, elapsed=3.0, ts=999.0)
|
|
assert payload.extract_token(text) == token
|
|
|
|
|
|
def test_extract_token_returns_none_on_protocol_message():
|
|
assert payload.extract_token("1234567890|A|IDLE|-100") is None
|
|
assert payload.extract_token("START|name:foo") is None
|
|
assert payload.extract_token("STOP") is None
|
|
|
|
|
|
def test_extract_token_returns_none_on_legacy_format():
|
|
# 3-field legacy format (no token) — should not falsely extract
|
|
assert payload.extract_token("1234.5,1.0,42|abcdef") is None
|
|
|
|
|
|
def test_extract_token_rejects_non_hex():
|
|
assert payload.extract_token("1234.5,1.0,42,nothex!!|abc") is None
|
|
|
|
|
|
def test_extract_token_handles_empty():
|
|
assert payload.extract_token("") is None
|
|
assert payload.extract_token("nopipe") is None
|