"""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()