import json import unittest from bitarray import bitarray from tests.harness.deterministic import ( DeterministicScenario, PacketSpec, active_bridge, add_openbridge_system, minimal_config, bytes_4, HBPF_DATA_SYNC, HBPF_SLT_VTERM, HBPF_VOICE, ) from tests.test_deterministic_harness import embedded_lc_payloads class TalkerAliasReportingTests(unittest.TestCase): def test_rx_end_and_timeout_report_alias_on_hbp_and_openbridge(self): for ingress in ('MASTER-A', 'OBP-1'): for timeout in (False, True): with self.subTest(ingress=ingress, timeout=timeout): config = minimal_config(('MASTER-A', 'MASTER-B')) add_openbridge_system(config, 'OBP-1', network_id=3001) config['REPORTS']['REPORT'] = True slot = 1 if ingress == 'OBP-1' else 2 bridges = active_bridge('91', 91, ((ingress, slot), ('MASTER-B', 2))) with DeterministicScenario(config=config, bridges=bridges) as scenario: events = [] report = scenario.bm.bridgeReportFactory(config) report.send_clients = events.append scenario.systems[ingress]._report = report inject = scenario.inject_obp if ingress == 'OBP-1' else scenario.inject_hbp for index, payload in enumerate(embedded_lc_payloads( bytes.fromhex('04004c43414c4c3132')), start=1): inject(ingress, PacketSpec( dst_id=91, slot=slot, seq=index, frame_type=HBPF_VOICE, dtype_vseq=index, payload=payload, )) self.assertFalse(any(b'talker_alias' in event for event in events)) if timeout: scenario.clock.advance(6) scenario.bm.stream_trimmer_loop() else: inject(ingress, PacketSpec( dst_id=91, slot=slot, seq=5, frame_type=HBPF_DATA_SYNC, dtype_vseq=HBPF_SLT_VTERM, )) end = [event[1:].decode('utf-8') for event in events if event[1:].startswith(b'GROUP VOICE,END,RX,')] self.assertEqual(len(end), 1) self.assertEqual(json.loads(end[0].split(',', 12)[12]), {'talker_alias': 'CALL12'}) # The next over must not inherit either alias or LC fragments. events.clear() scenario.clock.advance(6) inject(ingress, PacketSpec(dst_id=91, slot=slot, stream_id=0x01020305, frame_type=HBPF_VOICE, dtype_vseq=0)) inject(ingress, PacketSpec(dst_id=91, slot=slot, stream_id=0x01020305, seq=1, frame_type=HBPF_DATA_SYNC, dtype_vseq=HBPF_SLT_VTERM)) self.assertTrue(any(event[1:].startswith(b'GROUP VOICE,END,RX,') for event in events)) self.assertFalse(any(b'talker_alias' in event for event in events)) def test_complete_alias_decoding_and_missing_blocks(self): with DeterministicScenario(config=minimal_config(('MASTER-A',))) as scenario: decode = scenario.bm._complete_talker_alias for fmt, encoding, text in ( (1, 'latin-1', 'CALL, caf\u00e9'), (2, 'utf-8', 'CALL \u6f22\u5b57'), (3, 'utf-16-be', 'CALL \u6f22\u5b57'), ): with self.subTest(format=fmt): payload = text.encode(encoding) state = {'FORMAT': fmt, 'LENGTH': len(text), 'HEADER': payload[:6], 'BLOCKS': {}} self.assertIsNone(decode(state)) for number, offset in enumerate(range(6, len(payload), 7), start=1): state['BLOCKS'][number] = payload[offset:offset + 7] self.assertEqual(decode(state), text) state['BLOCKS'].pop(1) self.assertIsNone(decode(state)) text = 'CALL123456789012345' bits = bitarray(''.join(format(ord(char), '07b') for char in text)) state = {'FORMAT': 0, 'LENGTH': len(text), 'BITS': bits[:49], 'BLOCK_BITS': {2: bits[105:], 1: bits[49:105]}} self.assertEqual(decode(state), text) state['BLOCK_BITS'].pop(1) self.assertIsNone(decode(state)) self.assertIsNone(decode({'FORMAT': 2, 'LENGTH': 2, 'HEADER': b'A\xff'})) self.assertIsNone(decode({})) self.assertEqual(decode({'FORMAT': 1, 'LENGTH': 4, 'HEADER': b'CALLxx'}), 'CALL') def test_report_tail_preserves_prefix_and_survives_mux(self): from report_mux import prefix_bridge_event with DeterministicScenario(config=minimal_config(('MASTER-A',))) as scenario: events = [] report = scenario.bm.bridgeReportFactory(scenario.config) report.send_clients = events.append base = b'GROUP VOICE,END,RX,SYSTEM,1,2,3,2,91,4.25' alias = 'G7RZU, "caf\u00e9"\n\u6f22\u5b57' report.send_bridgeEvent(base) self.assertEqual(events[-1][1:], base) for server, repeater in ((None, None), (2341, 1001)): report.send_bridgeEvent(base, bytes_4(server) if server else None, bytes_4(repeater) if repeater else None, alias) raw = events[-1][1:] fields = raw.decode('utf-8').split(',', 12) self.assertEqual(','.join(fields[:10]).encode(), base) self.assertEqual(fields[10:12], ['', ''] if server is None else ['2341', '1001']) self.assertEqual(json.loads(fields[12]), {'talker_alias': alias}) self.assertNotIn(b'\n', raw) muxed = prefix_bridge_event(2, raw).decode('utf-8').split(',', 12) self.assertEqual(muxed[3], '2:SYSTEM') self.assertEqual(muxed[12], fields[12]) def test_last_complete_alias_survives_incomplete_repeat(self): with DeterministicScenario(config=minimal_config(('MASTER-A',))) as scenario: state = {} def observe(lc): scenario.bm._log_talker_alias( 'MASTER-A', state, b'\x23\xbe\x10', b'\x00\x00\x5b', 2, bytes_4(1), lc, lc[0]) # UTF-8 character crosses the header/block boundary. text = 'CALL \u6f22\u5b57' payload = text.encode('utf-8') header = bytes((4, 0, 0x80 | (len(text) << 1))) + payload[:6] observe(header) self.assertNotIn('TALKER_ALIAS', state) observe(bytes((5, 0)) + payload[6:].ljust(7, b'\x00')) self.assertEqual(state['TALKER_ALIAS'], text) observe(header) self.assertEqual(state['TALKER_ALIAS'], text)