mirrors/dmrlink3
1
0 Fork 0
mirror of https://github.com/n0mjs710/dmrlink3 synced 2026-11-08 04:19:24 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
main
dmrlink3/tests/test_protocol.py
T
Cort BuffingtonandClaude Opus 4.8 ad4e0ed13a reporting: mirror dashboard normalization to dmrlink3 (Step 3) 
Ports hblink3's dashboard-reporting overhaul to dmrlink3's IPSC path so the
two converge (plan: ~/development/dashboard-reporting-normalization.md):

- Granular peer events: a centralized reconciliation sweep in both the master
  and peer maintenance loops (report_maintenance) diffs currently-connected
  peers vs the last-reported set and emits send_peer connected/disconnected
  deltas only on change -- the dashboard reflects membership immediately
  instead of waiting for the periodic full push. (The single master link-state,
  in PEER mode, continues to ride the snapshot.)

- Demote the full push: reporting_loop now calls periodic_push, which re-pushes
  the full config+bridge snapshot only every REPORT_RESYNC_SECONDS (60s) as a
  reconciliation safety net and sends a tiny {'type':'ping'} heartbeat on the
  other ticks. Idle feed stops re-serializing full state every REPORT_INTERVAL.

- Ping-loss quality metric (PING_LOSS): detects a peer/master that STAYS
  CONNECTED but loses keepalives (lossy link -> choppy audio). One uniform
  gap-inference path serves both roles -- MASTER mode observes peers'
  MASTER_ALIVE_REQs, PEER mode observes the master's/peers' *_ALIVE_REPLYs to
  our requests. IPSC's cadence is known (ALIVE_TIMER) so no self-calibration is
  needed (unlike hblink3). A gap beyond the keepalive-timeout horizon is treated
  as a session lapse (reconnect), not loss. Sliding window pruned to
  PING_LOSS_WINDOW minutes; dashboard flags the row gold at >= PING_LOSS_WARN%.

Config: PING_LOSS_WINDOW / PING_LOSS_WARN added to GLOBAL (optional, defaults).
Dashboard: backend applies peer deltas + ignores the ping heartbeat; frontend
renders "N% ping loss" and golds the radio id. Tests: TestPeerEvents +
TestPingLoss (73 pass). Verified live over real sockets (peer delta fired,
heartbeat/resync split confirmed, PING_LOSS climbed to 30% under a lossy peer)
and deployed to the live dmrlink3 services clean.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-12 10:00:33 -05:00

412 lines
17 KiB
Python
Raw Permalink Blame History

"""Unit tests for the IPSC protocol state machine (dmrlink.IPSC)."""
import unittest
from time import time
from const import (
MASTER_REG_REPLY, PEER_LIST_REPLY, MASTER_ALIVE_REPLY,
DE_REG_REPLY, OPCODE_0xF0, SYSTEM_MAP_REQ,
VOICE_HEAD, VOICE_TERM, SLOT1_VOICE, SLOT2_VOICE,
)
from dmrlink import IPSC
from dmr_utils3.utils import bytes_4
from helpers import (
MockDatagramTransport, make_config, make_system, make_peer_entry,
make_reg_req, make_alive_req, make_dereg_req, make_peer_list_req,
make_gv_packet, MODE_DIGITAL_TS12, FLAGS_PEER,
)
MASTER_ID = 3112000
PEER_A_ID = 3112001
PEER_B_ID = 3112002
EXT_ID = 3112010 # an external repeater sending voice
MASTER_ID_B = bytes_4(MASTER_ID)
PEER_A_ID_B = bytes_4(PEER_A_ID)
PEER_B_ID_B = bytes_4(PEER_B_ID)
EXT_ID_B = bytes_4(EXT_ID)
SRC_SUB = b'\x00\x00\x01'
DST_GRP = b'\x00\x00\x09'
class _CapturingIPSC(IPSC):
"""Records group_voice() calls for timeslot-detection assertions."""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.gv_calls = []
def group_voice(self, src, dst, ts, end, peerid, data):
self.gv_calls.append({'src': src, 'dst': dst, 'ts': ts, 'end': end})
class IPSCMasterBase(unittest.TestCase):
"""Base: master-mode IPSC with a mock transport, no real socket."""
SYSTEM_NAME = 'TEST-MASTER'
def setUp(self):
cfg = make_config({
self.SYSTEM_NAME: make_system(MASTER_ID, '127.0.0.1', 50100, master_peer=True),
})
self.proto = IPSC(self.SYSTEM_NAME, cfg, None)
self.transport = MockDatagramTransport()
self.proto.transport = self.transport
def inject(self, data, host='127.0.0.1', port=50200):
self.proto.datagram_received(data, (host, port))
def pre_register(self, peer_id_bytes, host='127.0.0.1', port=50200):
self.proto._peers[peer_id_bytes] = make_peer_entry(host, port)
def pkts_to(self, host, port):
return [p for p, addr in self.transport.sent if addr == (host, port)]
def opcodes_to(self, host, port):
return [p[0:1] for p in self.pkts_to(host, port)]
# ---------------------------------------------------------------------------
# Registration
# ---------------------------------------------------------------------------
class TestRegistration(IPSCMasterBase):
def test_reg_adds_peer(self):
self.inject(make_reg_req(PEER_A_ID_B), '127.0.0.1', 50201)
self.assertIn(PEER_A_ID_B, self.proto._peers)
def test_reg_num_peers_updated(self):
self.inject(make_reg_req(PEER_A_ID_B), '127.0.0.1', 50201)
self.assertEqual(self.proto._local['NUM_PEERS'], 1)
def test_reg_sends_reg_reply(self):
self.inject(make_reg_req(PEER_A_ID_B), '127.0.0.1', 50201)
self.assertIn(MASTER_REG_REPLY, self.opcodes_to('127.0.0.1', 50201))
def test_reg_sends_peer_list(self):
self.inject(make_reg_req(PEER_A_ID_B), '127.0.0.1', 50201)
self.assertIn(PEER_LIST_REPLY, self.opcodes_to('127.0.0.1', 50201))
def test_rereg_no_duplicate(self):
pkt = make_reg_req(PEER_A_ID_B)
self.inject(pkt, '127.0.0.1', 50201)
self.inject(pkt, '127.0.0.1', 50201)
self.assertEqual(len(self.proto._peers), 1)
def test_second_peer_broadcast_to_first(self):
self.inject(make_reg_req(PEER_A_ID_B), '127.0.0.1', 50201)
self.transport.sent.clear()
self.inject(make_reg_req(PEER_B_ID_B), '127.0.0.1', 50202)
self.assertIn(PEER_LIST_REPLY, self.opcodes_to('127.0.0.1', 50201),
'Existing peer A should receive a fresh peer list when peer B registers')
def test_peer_list_contains_registered_peers(self):
self.inject(make_reg_req(PEER_A_ID_B), '127.0.0.1', 50201)
peer_list_pkts = [p for p in self.pkts_to('127.0.0.1', 50201)
if p[0:1] == PEER_LIST_REPLY]
self.assertTrue(peer_list_pkts)
# The peer list payload (after opcode + radio_id) should include PEER_A's radio ID
self.assertIn(PEER_A_ID_B, peer_list_pkts[-1])
# ---------------------------------------------------------------------------
# De-registration
# ---------------------------------------------------------------------------
class TestDeregistration(IPSCMasterBase):
def test_dereg_removes_peer(self):
self.pre_register(PEER_A_ID_B, '127.0.0.1', 50201)
self.inject(make_dereg_req(PEER_A_ID_B), '127.0.0.1', 50201)
self.assertNotIn(PEER_A_ID_B, self.proto._peers)
def test_dereg_sends_reply(self):
self.pre_register(PEER_A_ID_B, '127.0.0.1', 50201)
self.inject(make_dereg_req(PEER_A_ID_B), '127.0.0.1', 50201)
self.assertIn(DE_REG_REPLY, self.opcodes_to('127.0.0.1', 50201))
# ---------------------------------------------------------------------------
# Keepalive
# ---------------------------------------------------------------------------
class TestKeepalive(IPSCMasterBase):
def test_master_alive_req_gets_reply(self):
self.pre_register(PEER_A_ID_B, '127.0.0.1', 50201)
self.inject(make_alive_req(PEER_A_ID_B), '127.0.0.1', 50201)
self.assertIn(MASTER_ALIVE_REPLY, self.opcodes_to('127.0.0.1', 50201))
def test_master_alive_req_increments_ka_received(self):
self.pre_register(PEER_A_ID_B, '127.0.0.1', 50201)
self.inject(make_alive_req(PEER_A_ID_B), '127.0.0.1', 50201)
self.assertEqual(self.proto._peers[PEER_A_ID_B]['STATUS']['KEEP_ALIVES_RECEIVED'], 1)
def test_any_gv_packet_refreshes_watchdog(self):
self.pre_register(PEER_A_ID_B, '127.0.0.1', 50201)
old_ka_time = self.proto._peers[PEER_A_ID_B]['STATUS']['KEEP_ALIVE_RX_TIME']
import time
time.sleep(0.01)
pkt = make_gv_packet(PEER_A_ID_B, SRC_SUB, DST_GRP, VOICE_HEAD, 1)
self.inject(pkt, '127.0.0.1', 50201)
new_ka_time = self.proto._peers[PEER_A_ID_B]['STATUS']['KEEP_ALIVE_RX_TIME']
self.assertGreaterEqual(new_ka_time, old_ka_time,
'KA timestamp should refresh on any received packet')
# ---------------------------------------------------------------------------
# Peer list request — must unicast to requester only
# ---------------------------------------------------------------------------
class TestPeerListRequest(IPSCMasterBase):
def test_peer_list_req_unicasts_to_requester(self):
self.inject(make_reg_req(PEER_A_ID_B), '127.0.0.1', 50201)
self.inject(make_reg_req(PEER_B_ID_B), '127.0.0.1', 50202)
self.transport.sent.clear()
self.inject(make_peer_list_req(PEER_A_ID_B), '127.0.0.1', 50201)
# Peer A gets a reply
self.assertIn(PEER_LIST_REPLY, self.opcodes_to('127.0.0.1', 50201))
# Peer B does NOT get a reply
self.assertNotIn(PEER_LIST_REPLY, self.opcodes_to('127.0.0.1', 50202))
# ---------------------------------------------------------------------------
# Timeslot detection
# ---------------------------------------------------------------------------
class TestTimeslotDetection(unittest.TestCase):
"""
For VOICE_HEAD/VOICE_TERM: _ts comes from call_info bit 5.
For SLOT1/SLOT2_VOICE: _ts comes from burst_type bit 7.
This tests the critical bug fix where `(_call_info & TS_CALL_MSK) + 1`
would produce _ts=33 for TS2 instead of _ts=2.
"""
def setUp(self):
cfg = make_config({
'TEST-MASTER': make_system(MASTER_ID, '127.0.0.1', 50100, master_peer=True),
})
self.proto = _CapturingIPSC('TEST-MASTER', cfg, None)
self.proto.transport = MockDatagramTransport()
self.proto._peers[EXT_ID_B] = make_peer_entry()
def _inject(self, burst_type, timeslot):
pkt = make_gv_packet(EXT_ID_B, SRC_SUB, DST_GRP, burst_type, timeslot)
self.proto.datagram_received(pkt, ('127.0.0.1', 50200))
return self.proto.gv_calls[-1]['ts']
def test_voice_head_ts1(self):
self.assertEqual(self._inject(VOICE_HEAD, 1), 1)
def test_voice_head_ts2(self):
self.assertEqual(self._inject(VOICE_HEAD, 2), 2)
def test_voice_term_ts1(self):
self.assertEqual(self._inject(VOICE_TERM, 1), 1)
def test_voice_term_ts2(self):
self.assertEqual(self._inject(VOICE_TERM, 2), 2)
def test_slot1_voice_ts1(self):
# SLOT1_VOICE = 0x0A, bit 7 = 0 → TS1 regardless of call_info
self.assertEqual(self._inject(SLOT1_VOICE, 1), 1)
def test_slot2_voice_ts2(self):
# SLOT2_VOICE = 0x8A, bit 7 = 1 → TS2 regardless of call_info
self.assertEqual(self._inject(SLOT2_VOICE, 2), 2)
def test_slot2_voice_overrides_call_info_ts1(self):
# Packet has SLOT2_VOICE burst type (says TS2) but call_info TS bit = 0 (says TS1).
# The correct TS is 2 — burst_type wins for SLOT1/SLOT2.
pkt = make_gv_packet(EXT_ID_B, SRC_SUB, DST_GRP, SLOT2_VOICE, 1) # call_info=TS1
self.proto.datagram_received(pkt, ('127.0.0.1', 50200))
ts = self.proto.gv_calls[-1]['ts']
self.assertEqual(ts, 2, 'SLOT2_VOICE (bit7=1) must yield TS2 even if call_info says TS1')
# ---------------------------------------------------------------------------
# Unknown / new opcodes — no crash, no response
# ---------------------------------------------------------------------------
class TestUnknownOpcodes(IPSCMasterBase):
def test_opcode_0xf0_no_exception(self):
pkt = OPCODE_0xF0 + PEER_A_ID_B + b'\x00' * 10
self.inject(pkt) # must not raise
def test_opcode_0xf0_no_reply(self):
pkt = OPCODE_0xF0 + PEER_A_ID_B + b'\x00' * 10
self.inject(pkt)
self.assertFalse(self.transport.sent)
def test_system_map_req_no_exception(self):
pkt = SYSTEM_MAP_REQ + PEER_A_ID_B + b'\x00' * 10
self.inject(pkt)
def test_truly_unknown_opcode_no_exception(self):
pkt = b'\xFF' + PEER_A_ID_B + b'\x00' * 10
self.inject(pkt)
# ---------------------------------------------------------------------------
# Granular peer connect/disconnect reporting (dashboard delta events)
# ---------------------------------------------------------------------------
class _FakeReport:
"""Captures send_peer() deltas emitted by the reconciliation sweep."""
def __init__(self):
self.peer_events = []
def send_peer(self, system, radio_id, action, info=None):
self.peer_events.append({'system': system, 'radio_id': radio_id,
'action': action, 'info': info})
def send_config(self): pass
def send_bridge(self): pass
def send_bridge_event(self, e): pass
def send_rcm(self, d): pass
class TestPeerEvents(unittest.TestCase):
"""report_maintenance() emits a granular delta only when peer membership
actually changes -- the reconciliation-sweep contract mirrored from hblink3."""
SYSTEM_NAME = 'TEST-MASTER'
def setUp(self):
cfg = make_config({
self.SYSTEM_NAME: make_system(MASTER_ID, '127.0.0.1', 50100, master_peer=True),
})
self.report = _FakeReport()
self.proto = IPSC(self.SYSTEM_NAME, cfg, self.report)
self.proto.transport = MockDatagramTransport()
def register(self, peer_id_bytes, host='127.0.0.1', port=50200):
self.proto.datagram_received(make_reg_req(peer_id_bytes), (host, port))
def test_connect_emits_single_connected_delta(self):
self.register(PEER_A_ID_B)
self.proto.report_maintenance()
evts = self.report.peer_events
self.assertEqual(len(evts), 1)
self.assertEqual(evts[0]['action'], 'connected')
self.assertEqual(evts[0]['radio_id'], PEER_A_ID)
self.assertEqual(evts[0]['system'], self.SYSTEM_NAME)
self.assertIsNotNone(evts[0]['info']) # connected carries a snapshot
self.assertEqual(evts[0]['info']['RADIO_ID'], PEER_A_ID)
def test_no_change_emits_nothing(self):
self.register(PEER_A_ID_B)
self.proto.report_maintenance()
self.report.peer_events.clear()
self.proto.report_maintenance() # nothing changed
self.assertEqual(self.report.peer_events, [])
def test_disconnect_emits_disconnected_delta_without_info(self):
self.register(PEER_A_ID_B)
self.proto.report_maintenance()
self.report.peer_events.clear()
# A de-registration (peer leaves) removes it from self._peers.
self.proto.datagram_received(make_dereg_req(PEER_A_ID_B), ('127.0.0.1', 50200))
self.proto.report_maintenance()
evts = self.report.peer_events
self.assertEqual(len(evts), 1)
self.assertEqual(evts[0]['action'], 'disconnected')
self.assertEqual(evts[0]['radio_id'], PEER_A_ID)
self.assertIsNone(evts[0]['info']) # disconnected omits the snapshot
def test_two_peers_emit_two_connected_deltas(self):
self.register(PEER_A_ID_B)
self.register(PEER_B_ID_B, port=50201)
self.proto.report_maintenance()
actions = sorted((e['radio_id'], e['action']) for e in self.report.peer_events)
self.assertEqual(actions, [(PEER_A_ID, 'connected'), (PEER_B_ID, 'connected')])
def test_no_report_server_is_safe(self):
# Same sweep with reporting disabled (report=None) must not raise.
cfg = make_config({
self.SYSTEM_NAME: make_system(MASTER_ID, '127.0.0.1', 50100, master_peer=True),
})
proto = IPSC(self.SYSTEM_NAME, cfg, None)
proto.transport = MockDatagramTransport()
proto.datagram_received(make_reg_req(PEER_A_ID_B), ('127.0.0.1', 50200))
proto.report_maintenance() # must not raise
# ---------------------------------------------------------------------------
# Ping-loss quality metric (keepalive loss over a sliding window)
# ---------------------------------------------------------------------------
class TestPingLoss(unittest.TestCase):
"""_note_keepalive/_ping_loss_pct infer keepalive loss from the gap between
observed keepalives vs the known ALIVE_TIMER cadence (5s in make_system)."""
def setUp(self):
cfg = make_config({
'S': make_system(MASTER_ID, '127.0.0.1', 50100, master_peer=True),
})
self.proto = IPSC('S', cfg, None)
self.proto.transport = MockDatagramTransport()
self.rid = PEER_A_ID_B
def feed(self, *times):
for t in times:
self.proto._note_keepalive(self.rid, _now=t)
def test_clean_cadence_is_zero_loss(self):
self.feed(0, 5, 10, 15, 20) # every 5s == ALIVE_TIMER
self.assertEqual(self.proto._ping_loss_pct(self.rid, 20), 0)
def test_first_observation_no_false_positive(self):
self.feed(1000) # no prior -> gap 0 -> 0 missed
self.assertEqual(self.proto._ping_loss_pct(self.rid, 1000), 0)
def test_jitter_under_threshold_not_counted(self):
self.feed(0, 6) # gap 6 < base*1.5 (7.5) -> not lossy
self.assertEqual(self.proto._ping_loss_pct(self.rid, 6), 0)
def test_gap_infers_missed_keepalives(self):
# 0,5 clean; then a 15s gap == two keepalives lost before the one at 20.
self.feed(0, 5, 20)
st = self.proto._ping_q[self.rid]
self.assertEqual([m for _, m in st['ev']], [0, 0, 2])
# received=3 events, missed=2 -> 2/(3+2) = 40%
self.assertEqual(self.proto._ping_loss_pct(self.rid, 20), 40)
def test_window_prunes_old_loss(self):
# PING_LOSS_WINDOW defaults to 5 min (300s). Loss at t~20 ages out by t~400.
self.feed(0, 5, 20)
self.assertGreater(self.proto._ping_loss_pct(self.rid, 20), 0)
self.feed(400, 405, 410) # fresh clean pings well past the window
self.assertEqual(self.proto._ping_loss_pct(self.rid, 410), 0)
def test_report_maintenance_sets_ping_loss_on_connected_peer(self):
cfg = make_config({'S': make_system(MASTER_ID, '127.0.0.1', 50100, master_peer=True)})
proto = IPSC('S', cfg, None)
proto.transport = MockDatagramTransport()
proto.datagram_received(make_reg_req(PEER_A_ID_B), ('127.0.0.1', 50200))
# Inject a lossy keepalive history near wall-clock now, so it stays inside
# the window when report_maintenance() refreshes against real time().
now = time()
proto._note_keepalive(PEER_A_ID_B, _now=now - 15) # first: no prior gap
proto._note_keepalive(PEER_A_ID_B, _now=now - 10) # +5s clean
proto._note_keepalive(PEER_A_ID_B, _now=now) # +10s gap -> 1 missed
proto.report_maintenance()
self.assertGreater(proto._peers[PEER_A_ID_B]['STATUS']['PING_LOSS'], 0)
if __name__ == '__main__':
unittest.main()
Reference in new issue View Git Blame Copy Permalink