You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
133 lines
7.2 KiB
133 lines
7.2 KiB
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)
|