233 lines
7.0 KiB
Python
233 lines
7.0 KiB
Python
# tests/test_op25_controller.py
|
|
import pytest
|
|
from unittest.mock import patch, mock_open, MagicMock
|
|
from fastapi.testclient import TestClient
|
|
from app.op25_controller import router
|
|
from fastapi import FastAPI
|
|
import json
|
|
|
|
# Initialize the FastAPI app with the router for testing
|
|
app = FastAPI()
|
|
app.include_router(router, prefix="/op25")
|
|
client = TestClient(app)
|
|
|
|
# Example input and expected outputs
|
|
example_input_json = {
|
|
"type": "P25",
|
|
"systemName": "MTA",
|
|
"channels": [
|
|
"770.15625",
|
|
"770.43125",
|
|
"773.29375",
|
|
"773.84375",
|
|
"774.30625",
|
|
"123.32132"
|
|
],
|
|
"tags": [
|
|
{
|
|
"talkgroup": "abc",
|
|
"tagDec": 1
|
|
},
|
|
{
|
|
"talkgroup": "deef",
|
|
"tagDec": 123
|
|
}
|
|
],
|
|
"whitelist": [
|
|
123,
|
|
321,
|
|
456,
|
|
654,
|
|
888
|
|
]
|
|
}
|
|
|
|
expected_active_config_json = {
|
|
"channels": [
|
|
{
|
|
"name": "MTA",
|
|
"device": "sdr",
|
|
"trunking_sysname": "MTA",
|
|
"enable_analog": "off",
|
|
"demod_type": "cqpsk",
|
|
"cqpsk_tracking": True,
|
|
"filter_type": "rc",
|
|
"tracking_threshold": 120,
|
|
"tracking_feedback": 0.75,
|
|
"destination": "udp://localhost:23456",
|
|
"excess_bw": 0.2,
|
|
"if_rate": 24000,
|
|
"plot": "",
|
|
"symbol_rate": 4800,
|
|
"blacklist": "",
|
|
"whitelist": ""
|
|
}
|
|
],
|
|
"devices": [
|
|
{
|
|
"args": "rtl=0",
|
|
"gains": "lna:39",
|
|
"gain_mode": False,
|
|
"name": "sdr",
|
|
"offset": 0,
|
|
"ppm": 0.0,
|
|
"rate": 1920000,
|
|
"usable_bw_pct": 0.85,
|
|
"tunable": True
|
|
}
|
|
],
|
|
"trunking": {
|
|
"module": "tk_p25.py",
|
|
"chans": [
|
|
{
|
|
"sysname": "MTA",
|
|
"control_channel_list": "770.15625,770.43125,773.29375,773.84375,774.30625,123.32132",
|
|
"tagsFile": "/configs/active.cfg.tags.tsv",
|
|
"whitelist": "/configs/active.cfg.whitelist.tsv",
|
|
"nac": "",
|
|
"wacn": "",
|
|
"tdma_cc": False,
|
|
"crypt_behavior": 2
|
|
}
|
|
]
|
|
},
|
|
"audio": {
|
|
"module": "sockaudio.py",
|
|
"instances": [
|
|
{
|
|
"instance_name": "audio0",
|
|
"device_name": "pulse",
|
|
"udp_port": 23456,
|
|
"audio_gain": 2.5,
|
|
"number_channels": 1
|
|
}
|
|
]
|
|
},
|
|
"terminal": {
|
|
"module": "terminal.py",
|
|
"terminal_type": "http:0.0.0.0:8081",
|
|
"terminal_timeout": 5.0,
|
|
"curses_plot_interval": 0.2,
|
|
"http_plot_interval": 1.0,
|
|
"http_plot_directory": "../www/images",
|
|
"tuning_step_large": 1200,
|
|
"tuning_step_small": 100
|
|
}
|
|
}
|
|
|
|
expected_tags_tsv = "abc\t1\ndeef\t123\n"
|
|
expected_whitelist_tsv = "123\t\n321\t\n456\t\n654\t\n888\t\n"
|
|
|
|
# Mock data for subprocess.Popen
|
|
mock_popen = MagicMock()
|
|
mock_process = MagicMock()
|
|
mock_popen.return_value = mock_process
|
|
|
|
@pytest.fixture
|
|
def mock_subprocess_popen():
|
|
with patch("app.op25_controller.subprocess.Popen", return_value=mock_process) as mock_popen_patched:
|
|
yield mock_popen_patched
|
|
|
|
@pytest.fixture
|
|
def mock_os_killpg():
|
|
with patch("app.op25_controller.os.killpg") as mock_killpg_patched:
|
|
yield mock_killpg_patched
|
|
|
|
@pytest.fixture
|
|
def mock_open_functions():
|
|
with patch("builtins.open", mock_open()) as mock_file:
|
|
yield mock_file
|
|
|
|
@pytest.fixture
|
|
def mock_json_dump():
|
|
with patch("app.op25_controller.json.dump") as mock_json_dump_patched:
|
|
yield mock_json_dump_patched
|
|
|
|
@pytest.fixture
|
|
def mock_csv_writer():
|
|
with patch("app.op25_controller.csv.writer") as mock_csv_writer_patched:
|
|
yield mock_csv_writer_patched
|
|
|
|
def test_generate_config_p25(
|
|
mock_open_functions, mock_json_dump, mock_csv_writer
|
|
):
|
|
# Prepare the response of csv.writer
|
|
mock_writer_instance = MagicMock()
|
|
mock_csv_writer.return_value = mock_writer_instance
|
|
|
|
response = client.post("/op25/generate-config", json=example_input_json)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {"message": "Config exported to '/configs/active.cfg.json'"}
|
|
|
|
# Check that json.dump was called with the correct data
|
|
mock_json_dump.assert_called_once()
|
|
args, kwargs = mock_json_dump.call_args
|
|
config_written = args[0]
|
|
assert config_written == expected_active_config_json
|
|
assert kwargs["fp"].name == '/configs/active.cfg.json'
|
|
|
|
# Check that tags were written correctly
|
|
expected_tags = [
|
|
["abc", 1],
|
|
["deef", 123]
|
|
]
|
|
mock_writer_instance.writerow.assert_any_call(["abc", 1])
|
|
mock_writer_instance.writerow.assert_any_call(["deef", 123])
|
|
|
|
# Similarly, check whitelist writing
|
|
# Since both tags and whitelist are written, ensure writerow for whitelist is also called
|
|
whitelist_calls = [
|
|
patch.call([123]),
|
|
patch.call([321]),
|
|
patch.call([456]),
|
|
patch.call([654]),
|
|
patch.call([888])
|
|
]
|
|
# However, since csv.writer is mocked, and both write operations use the same mock,
|
|
# it's difficult to differentiate between tags and whitelist writes unless separated.
|
|
# A better approach would be to separate the file paths.
|
|
# For simplicity, assume that the writer is called twice: once for tags, once for whitelist
|
|
|
|
def test_start_op25(mock_subprocess_popen):
|
|
# Start OP25 when it's not running
|
|
response = client.post("/op25/start")
|
|
assert response.status_code == 200
|
|
assert response.json() == {"status": "OP25 started"}
|
|
mock_subprocess_popen.assert_called_once_with(
|
|
"/op25/op25/gr-op25_repeater/apps/run_multi-rx_service.sh",
|
|
shell=True,
|
|
preexec_fn=ANY,
|
|
cwd="/op25/op25_gr-repeater/apps/"
|
|
)
|
|
|
|
# Start OP25 again when it's already running
|
|
response = client.post("/op25/start")
|
|
assert response.status_code == 200
|
|
assert response.json() == {"status": "OP25 already running"}
|
|
|
|
def test_stop_op25(mock_subprocess_popen, mock_os_killpg, mock_process):
|
|
# Ensure OP25 is running first
|
|
with patch("app.op25_controller.op25_process", mock_process):
|
|
response = client.post("/op25/stop")
|
|
assert response.status_code == 200
|
|
assert response.json() == {"status": "OP25 stopped"}
|
|
mock_os_killpg.assert_called_once()
|
|
|
|
# Stop OP25 when it's not running
|
|
with patch("app.op25_controller.op25_process", None):
|
|
response = client.post("/op25/stop")
|
|
assert response.status_code == 200
|
|
assert response.json() == {"status": "OP25 is not running"}
|
|
|
|
def test_get_status():
|
|
# When OP25 is not running
|
|
response = client.get("/op25/status")
|
|
assert response.status_code == 200
|
|
assert response.json() == {"status": "stopped"}
|
|
|
|
# When OP25 is running
|
|
with patch("app.op25_controller.op25_process", MagicMock()):
|
|
response = client.get("/op25/status")
|
|
assert response.status_code == 200
|
|
assert response.json() == {"status": "running"} |