import copy import configparser import json import socket import tempfile import unittest from pathlib import Path from unittest.mock import patch from urllib.request import Request, urlopen from API import make_api_resource from tests.test_api import FakeRequest from tests.harness.deterministic import ( DeterministicScenario, PacketSpec, minimal_config, bytes_3, HBPF_DATA_SYNC, HBPF_SLT_VHEAD, HBPF_SLT_VTERM, HBPF_VOICE, ) from tests.harness.udp_blackbox import ( DependencySandbox, FreeDmrProcess, HbpRepeater, free_udp_port, require_udp_integration_enabled, write_bridge_master_config, ) class DialAPITests(unittest.TestCase): def test_peer_key_control_keeps_rf_busy_and_slot_boundaries(self): with DeterministicScenario(config=self.config()) as scenario: scenario.register_peer('MASTER-A') scenario.config['SYSTEMS']['MASTER-A']['_opt_key'] = 'peer-secret' scenario.inject_hbp('MASTER-A', PacketSpec(slot=1, peer_id=1001, dst_id=235)) resource = make_api_resource(scenario.config, scenario.bridge_state, scenario.bm.control_dial) before = copy.deepcopy(scenario.bridge_state) request = FakeRequest('/api/v1/peer/dial', { 'dmrid': 1001, 'key': 'peer-secret', 'slot': 1, 'action': 'unlink', }) result = json.loads(resource.render_POST(request)) self.assertEqual((request.code, result['error']), (409, 'rf_busy')) self.assertEqual(scenario.bridge_state, before) request = FakeRequest('/api/v1/peer/dial', { 'dmrid': 1001, 'key': 'peer-secret', 'slot': 2, 'action': 'link', 'tg': 91, }) result = json.loads(resource.render_POST(request)) self.assertEqual((request.code, result['slot'], result['tg']), (200, 2, 91)) def config(self): config = minimal_config() config['GLOBAL']['SYSTEM_API_KEY'] = 'test-key' config['SYSTEMS']['MASTER-A']['DEFAULT_DIAL_TS1'] = 91 config['SYSTEMS']['MASTER-A']['DEFAULT_DIAL_TS2'] = 92 return config def request(self, scenario, action, slot, tg=None, peer=1001): resource = make_api_resource(scenario.config, scenario.bridge_state, scenario.bm.control_dial) payload = {'systemkey': 'test-key', 'dmrid': peer, 'slot': slot, 'action': action} if tg is not None: payload['tg'] = tg request = FakeRequest('/api/v1/dial', payload) result = json.loads(resource.render_POST(request)) return request.code, result def test_link_unlink_and_status_match_rf_dial_state(self): for slot in (1, 2): results = [] for via_api in (False, True): with DeterministicScenario(config=self.config()) as scenario: scenario.register_peer('MASTER-A') scenario.bm.make_default_reflectors() scenario.bm.make_static_tg(2350, slot, 10, 'MASTER-A') snapshots = [] for index, (action, tg) in enumerate((('link', 235), ('link', 235), ('link', 91), ('unlink', 4000), ('status', 5000))): scenario.clock.advance(2) if via_api: code, _ = self.request(scenario, action, slot, tg if action == 'link' else None) self.assertEqual(code, 200) else: scenario.inject_hbp('MASTER-A', PacketSpec( peer_id=1001, dst_id=tg, slot=slot, call_type='unit', frame_type=HBPF_DATA_SYNC, dtype_vseq=HBPF_SLT_VHEAD, stream_id=100 + index, )) snapshots.append(copy.deepcopy(scenario.bridge_state)) results.append(snapshots) with self.subTest(slot=slot): self.assertEqual(results[0], results[1]) def test_reset_restores_only_target_slot_default_and_status_is_read_only(self): for slot in (1, 2): with self.subTest(slot=slot), DeterministicScenario(config=self.config()) as scenario: scenario.register_peer('MASTER-A') scenario.bm.make_default_reflectors() scenario.bm.make_static_tg(2350, slot, 10, 'MASTER-A') config_before = copy.deepcopy(scenario.config) self.assertEqual(self.request(scenario, 'link', slot, 235)[0], 200) untouched = [(name, copy.deepcopy(entry)) for name, entries in scenario.bridge_state.items() for entry in entries if not name.startswith('#') or entry['SYSTEM'] != 'MASTER-A' or entry['TS'] != slot] code, result = self.request(scenario, 'reset', slot) self.assertEqual(code, 200) self.assertEqual(result['tg'], 90 + slot) self.assertEqual(scenario.config, config_before) for name, entry in untouched: self.assertIn(entry, scenario.bridge_state[name]) default = next(entry for entry in scenario.bridge_state['#{}'.format(90 + slot)] if entry['SYSTEM'] == 'MASTER-A' and entry['TS'] == slot) self.assertTrue(default['ACTIVE']) self.assertEqual(default['TO_TYPE'], 'OFF') before = copy.deepcopy(scenario.bridge_state) self.assertEqual(self.request(scenario, 'status', slot)[0], 200) self.assertEqual(before, scenario.bridge_state) callback, args, kwargs = scenario.reactor.thread_calls[-1] self.assertEqual(callback, scenario.bm.disconnectedVoice) self.assertEqual(args, ('MASTER-A', slot, 90 + slot)) def test_reset_without_default_disconnects_and_unlink_default_returns_on_timer(self): config = self.config() config['SYSTEMS']['MASTER-A']['DEFAULT_DIAL_TS2'] = 0 with DeterministicScenario(config=config) as scenario: scenario.register_peer('MASTER-A') scenario.bm.make_default_reflectors() self.assertEqual(self.request(scenario, 'unlink', 1)[1]['tg'], 0) scenario.clock.advance(61) scenario.bm.rule_timer_loop() self.assertEqual(scenario.bm.active_reflector_for_system_slot('MASTER-A', 1), '#91') self.assertEqual(self.request(scenario, 'link', 2, 235)[0], 200) self.assertEqual(self.request(scenario, 'reset', 2)[1]['tg'], 0) def test_rf_voice_blocks_control_but_network_voice_does_not(self): with DeterministicScenario(config=self.config()) as scenario: scenario.register_peer('MASTER-A') scenario.inject_hbp('MASTER-A', PacketSpec(slot=1, peer_id=1001, dst_id=235)) before = copy.deepcopy(scenario.bridge_state) for action in ('link', 'unlink', 'reset', 'status', 'reset_tgs'): code, result = self.request(scenario, action, 1, 91 if action == 'link' else None) self.assertEqual((code, result['error']), (409, 'rf_busy')) self.assertEqual(before, scenario.bridge_state) # RF on the other slot is not a veto for the requested slot. self.assertEqual(self.request(scenario, 'link', 2, 235)[0], 200) scenario.inject_hbp('MASTER-A', PacketSpec( slot=1, peer_id=1001, dst_id=235, seq=1, payload=b'\x01' * 33, frame_type=HBPF_DATA_SYNC, dtype_vseq=HBPF_SLT_VTERM, )) status = scenario.systems['MASTER-A'].STATUS[1] status['TX_TYPE'] = HBPF_SLT_VHEAD status['TX_TIME'] = scenario.clock.time() self.assertEqual(self.request(scenario, 'unlink', 1)[0], 200) self.assertEqual(self.request(scenario, 'reset_tgs', 1)[0], 200) scenario.inject_hbp('MASTER-A', PacketSpec( slot=1, peer_id=1001, dst_id=235, stream_id=0x01020306, )) scenario.clock.advance(scenario.bm.STREAM_TO + 1) self.assertEqual(self.request(scenario, 'reset', 1)[0], 200) def test_reset_tgs_applies_timer_expiry_only_to_selected_non_dial_entries(self): for slot in (1, 2): with self.subTest(slot=slot), DeterministicScenario(config=self.config()) as scenario: scenario.register_peer('MASTER-A') scenario.bm.make_default_reflectors() for ts in (1, 2): scenario.bm.make_static_tg(2350, ts, 10, 'MASTER-A') # Exercise both states of every timer type without changing the # existing constructors or relying on unrelated RF trigger logic. template = next(entry for entry in scenario.bridge_state['2350'] if entry['SYSTEM'] == 'MASTER-A' and entry['TS'] == slot) for index, (timer_type, active) in enumerate( (kind, active) for kind in ('ON', 'OFF', 'NONE', 'STAT') for active in (False, True) ): entry = copy.deepcopy(template) entry.update(TGID=bytes_3(2360 + index), TO_TYPE=timer_type, ACTIVE=active, TIMER=scenario.clock.time() + 600) scenario.bridge_state[str(2360 + index)] = [entry] # Conventional TG control must work even with dial-a-TG disabled. scenario.config['SYSTEMS']['MASTER-A']['DIAL_A_TG'] = False before = copy.deepcopy(scenario.bridge_state) config_before = copy.deepcopy(scenario.config) prompts_before = list(scenario.reactor.thread_calls) expected = copy.deepcopy(before) expired = 0 for name, entries in expected.items(): for entry in entries: if name.startswith('#') or entry['SYSTEM'] != 'MASTER-A' or entry['TS'] != slot: continue if ((entry['TO_TYPE'] == 'ON' and entry['ACTIVE']) or (entry['TO_TYPE'] == 'OFF' and not entry['ACTIVE'])): entry['ACTIVE'] = entry['TO_TYPE'] == 'OFF' entry['TIMER'] = scenario.clock.time() expired += 1 code, result = self.request(scenario, 'reset_tgs', slot) self.assertEqual(code, 200) self.assertEqual(result['expired'], expired) self.assertEqual(result['announcement'], 'none') self.assertEqual(scenario.bridge_state, expected) self.assertEqual(scenario.config, config_before) self.assertEqual(scenario.reactor.thread_calls, prompts_before) self.assertEqual(self.request(scenario, 'reset_tgs', slot)[1]['expired'], 0) self.assertEqual(scenario.bridge_state, expected) def test_every_voice_burst_blocks_api_until_a_real_terminator(self): for call_type in ('group', 'unit'): with self.subTest(call_type=call_type), DeterministicScenario(config=self.config()) as scenario: scenario.register_peer('MASTER-A') scenario.bm.words['en_GB'].update({str(n): b'' for n in range(10)}) for burst in range(6): scenario.inject_hbp('MASTER-A', PacketSpec( slot=1, peer_id=1001, dst_id=235, call_type=call_type, seq=burst, payload=bytes([burst]) * 33, frame_type=HBPF_VOICE, dtype_vseq=burst, )) code, result = self.request(scenario, 'status', 1) self.assertEqual((code, result.get('error')), (409, 'rf_busy')) scenario.inject_hbp('MASTER-A', PacketSpec( slot=1, peer_id=1001, dst_id=235, call_type=call_type, seq=6, payload=b'\x06' * 33, frame_type=HBPF_DATA_SYNC, dtype_vseq=HBPF_SLT_VTERM, )) self.assertEqual(self.request(scenario, 'status', 1)[0], 200) def test_data_does_not_block_control_and_reserved_targets_do_not_mutate(self): with DeterministicScenario(config=self.config()) as scenario: scenario.register_peer('MASTER-A') for call_type in ('unit', 'group'): scenario.inject_hbp('MASTER-A', PacketSpec( peer_id=1001, dst_id=900999, slot=1, call_type=call_type, frame_type=HBPF_DATA_SYNC, dtype_vseq=6, )) self.assertEqual(self.request(scenario, 'link', 1, 235)[0], 200) before = copy.deepcopy(scenario.bridge_state) for tg in (1, 8, 9, 4000, 4001, 4999, 5000, 9991, 9999): code, result = self.request(scenario, 'link', 1, tg) self.assertEqual((code, result['error']), (400, 'invalid_tg')) self.assertEqual(before, scenario.bridge_state) scenario.config['SYSTEMS']['MASTER-A']['DIAL_A_TG'] = False self.assertEqual(self.request(scenario, 'link', 1, 235)[1]['error'], 'dial_disabled') scenario.config['SYSTEMS']['MASTER-A']['DIAL_A_TG'] = True scenario.systems['MASTER-A'].STATUS[1]['_allStarMode'] = True self.assertEqual(self.request(scenario, 'reset', 1)[1]['error'], 'dial_disabled') self.assertEqual(self.request(scenario, 'reset_tgs', 1)[1]['error'], 'dial_disabled') self.assertEqual(before, scenario.bridge_state) def test_announcement_uses_actual_link_not_default(self): with DeterministicScenario(config=self.config()) as scenario: scenario.register_peer('MASTER-A') scenario.bm.words['en_GB'].update({ 'linkedto': b'linkedto', 'notlinked': b'notlinked', **{str(n): str(n).encode('ascii') for n in range(10)}, }) for slot in (1, 2): for action, tg in (('link', 235), ('unlink', None)): self.assertEqual(self.request(scenario, action, slot, tg)[0], 200) callback, args, kwargs = scenario.reactor.thread_calls[-1] with patch.object(scenario.bm, 'pkt_gen', side_effect=lambda *args: iter(())) as generate, patch.object( scenario.bm, 'sleep', lambda seconds: None ): callback(*args, **kwargs) generated = generate.call_args.args self.assertEqual(generated[3], slot - 1) self.assertIn(b'linkedto' if tg else b'notlinked', generated[4]) self.assertEqual([word for word in generated[4] if word.isdigit()], [b'2', b'3', b'5'] if tg else []) class DialAPIBlackBoxTest(unittest.TestCase): def test_http_dial_generates_voice_to_logged_in_hbp_peer_on_requested_slot(self): require_udp_integration_enabled() root = Path(__file__).resolve().parents[1] deps = DependencySandbox(root) self.addCleanup(deps.cleanup) python = deps.resolve_python() for slot in (1, 2): with self.subTest(slot=slot), tempfile.TemporaryDirectory() as directory: path = Path(directory) / 'freedmr.cfg' keys = Path(directory) / 'keys.json' keys.write_text(json.dumps({'SYSTEM_API_KEY': 'test-key'}), encoding='utf-8') udp_port = free_udp_port() with socket.socket() as listener: listener.bind(('127.0.0.1', 0)) api_port = listener.getsockname()[1] write_bridge_master_config(path, {'MASTER-A': udp_port}) config = configparser.ConfigParser(interpolation=None) config.read(path) config['GLOBAL']['ENABLE_API'] = 'True' config['GLOBAL']['API_PORT'] = str(api_port) config['ALIASES']['PATH'] = directory + '/' config['ALIASES']['KEYS_FILE'] = keys.name with path.open('w', encoding='utf-8') as output: config.write(output) with FreeDmrProcess(root, path, python) as process: process.wait_for_start() peer = HbpRepeater(udp_port, 1001) try: peer.login() body = json.dumps({'systemkey': 'test-key', 'dmrid': 1001, 'action': 'link', 'slot': slot, 'tg': 235}).encode('utf-8') request = Request('http://127.0.0.1:{}/api/v1/dial'.format(api_port), data=body, headers={'Content-Type': 'application/json'}) with urlopen(request, timeout=3) as response: result = json.load(response) self.assertEqual(result['tg'], 235) self.assertEqual(result['slot'], slot) prompt = peer.recv(timeout=4) self.assertEqual(prompt.fields['slot'], slot) self.assertEqual(prompt.fields['dst_id'], bytes_3(9)) self.assertEqual(prompt.fields['rf_src'], bytes_3(5000)) finally: peer.close() if __name__ == '__main__': unittest.main()