This commit is contained in:
Logan Cusano
@@ -1,233 +0,0 @@
|
||||
# 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"}
|
||||
Reference in New Issue
Block a user