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.
309 lines
17 KiB
309 lines
17 KiB
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()
|