logan/drb-edge-node
1
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

WIP: init testing #2

Draft
logan wants to merge 7 commits from implement-unit-testing into main
pull from: implement-unit-testing
merge into: logan:main
Conversation 0 Commits 7 Files Changed 2 +30 -29
1 changed files with 30 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 98727615a3 - Show all commits

59
tests/test_op25_controller.py
View File

@@ -37,7 +37,7 @@ class MockConfigGenerator(BaseModel):
type: str type: str
systemName: str systemName: str
channels: Optional[List[str]] = None channels: Optional[List[str]] = None
tags: Optional[str] = None tags: Optional[List[MockTalkgroupTag]] = None
whitelist: Optional[str] = None whitelist: Optional[str] = None
icecastConfig: Optional[MockIcecastConfig] = None icecastConfig: Optional[MockIcecastConfig] = None
config: Optional[MockAnalogConfig] = None config: Optional[MockAnalogConfig] = None
@@ -83,7 +83,8 @@ class MockTerminalConfig(BaseModel):
pass pass
class MockTalkgroupTag(BaseModel): class MockTalkgroupTag(BaseModel):
pass tagDec: int
tagName: str
mock_models.ConfigGenerator = MockConfigGenerator mock_models.ConfigGenerator = MockConfigGenerator
mock_models.DecodeMode = MockDecodeMode mock_models.DecodeMode = MockDecodeMode
@@ -112,7 +113,7 @@ SAMPLE_P25_CONFIG = {
"type": "P25", "type": "P25",
"systemName": "TestSystem", "systemName": "TestSystem",
"channels": ["851.12345", "852.67890"], "channels": ["851.12345", "852.67890"],
"tags": "101,Group A\n102,Group B", "tags": [{"tagDec": 101, "tagName": "Group A"}, {"tagDec": 102, "tagName": "Group B"}],
"whitelist": "101", "whitelist": "101",
"icecastConfig": { "icecastConfig": {
"icecast_host": "localhost", "icecast_host": "localhost",
@@ -129,7 +130,7 @@ def reset_and_mock_globals(monkeypatch):
before each test, ensuring test isolation. before each test, ensuring test isolation.
""" """
# Reset the global process variable in the controller module # Reset the global process variable in the controller module
monkeypatch.setattr("app.routers.op25_controller.op25_process", None) monkeypatch.setattr("routers.op25_controller.op25_process", None)
# Mock asyncio.sleep to prevent tests from actually waiting # Mock asyncio.sleep to prevent tests from actually waiting
mock_sleep = MagicMock() mock_sleep = MagicMock()
@@ -140,7 +141,7 @@ def reset_and_mock_globals(monkeypatch):
monkeypatch.setattr("os.getpgid", MagicMock(return_value=12345)) monkeypatch.setattr("os.getpgid", MagicMock(return_value=12345))
@patch("app.routers.op25_controller.subprocess.Popen") @patch("routers.op25_controller.subprocess.Popen")
def test_start_op25_success(mock_popen): def test_start_op25_success(mock_popen):
"""Test the /start endpoint successfully starts the process.""" """Test the /start endpoint successfully starts the process."""
mock_process = MagicMock() mock_process = MagicMock()
@@ -153,7 +154,7 @@ def test_start_op25_success(mock_popen):
mock_popen.assert_called_once() mock_popen.assert_called_once()
@patch("app.routers.op25_controller.subprocess.Popen", side_effect=Exception("Popen failed")) @patch("routers.op25_controller.subprocess.Popen", side_effect=Exception("Popen failed"))
def test_start_op25_failure(mock_popen): def test_start_op25_failure(mock_popen):
"""Test the /start endpoint when Popen raises an exception.""" """Test the /start endpoint when Popen raises an exception."""
response = client.post("/op25/start") response = client.post("/op25/start")
@@ -168,13 +169,13 @@ def test_stop_op25_not_running():
assert response.json() == {"status": "OP25 was not running"} assert response.json() == {"status": "OP25 was not running"}
@patch("app.routers.op25_controller.subprocess.Popen") @patch("routers.op25_controller.subprocess.Popen")
def test_stop_op25_success(mock_popen, monkeypatch): def test_stop_op25_success(mock_popen, monkeypatch):
"""Test the /stop endpoint successfully stops a running process.""" """Test the /stop endpoint successfully stops a running process."""
mock_process = MagicMock() mock_process = MagicMock()
mock_process.pid = 12345 mock_process.pid = 12345
mock_process.poll.return_value = None # Indicates it's running mock_process.poll.return_value = None # Indicates it's running
monkeypatch.setattr("app.routers.op25_controller.op25_process", mock_process) monkeypatch.setattr("routers.op25_controller.op25_process", mock_process)
response = client.post("/op25/stop") response = client.post("/op25/stop")
assert response.status_code == 200 assert response.status_code == 200
@@ -192,14 +193,14 @@ def test_get_status_not_running():
assert data["active_system"] is None assert data["active_system"] is None
@patch("app.routers.op25_controller.get_current_system_from_config", return_value="TestSystem") @patch("routers.op25_controller.get_current_system_from_config", return_value="TestSystem")
@patch("app.routers.op25_controller.subprocess.Popen") @patch("routers.op25_controller.subprocess.Popen")
def test_get_status_running(mock_popen, mock_get_system, monkeypatch): def test_get_status_running(mock_popen, mock_get_system, monkeypatch):
"""Test the /status endpoint when the process is running.""" """Test the /status endpoint when the process is running."""
mock_process = MagicMock() mock_process = MagicMock()
mock_process.pid = 12345 mock_process.pid = 12345
mock_process.poll.return_value = None # Running mock_process.poll.return_value = None # Running
monkeypatch.setattr("app.routers.op25_controller.op25_process", mock_process) monkeypatch.setattr("routers.op25_controller.op25_process", mock_process)
response = client.get("/op25/status") response = client.get("/op25/status")
assert response.status_code == 200 assert response.status_code == 200
@@ -211,11 +212,11 @@ def test_get_status_running(mock_popen, mock_get_system, monkeypatch):
@patch("builtins.open", new_callable=mock_open) @patch("builtins.open", new_callable=mock_open)
@patch("app.routers.op25_controller.json.dump") @patch("routers.op25_controller.json.dump")
@patch("app.routers.op25_controller.save_talkgroup_tags") @patch("routers.op25_controller.save_talkgroup_tags")
@patch("app.routers.op25_controller.save_whitelist") @patch("routers.op25_controller.save_whitelist")
@patch("app.routers.op25_controller.generate_liquid_script") @patch("routers.op25_controller.generate_liquid_script")
@patch("app.routers.op25_controller.subprocess.Popen") @patch("routers.op25_controller.subprocess.Popen")
def test_set_active_config_no_restart(mock_popen, mock_liquid, mock_white, mock_tags, mock_dump, mock_file): def test_set_active_config_no_restart(mock_popen, mock_liquid, mock_white, mock_tags, mock_dump, mock_file):
"""Test setting active config without restarting the radio.""" """Test setting active config without restarting the radio."""
response = client.post("/op25/set_active_config?restart=false", json=SAMPLE_P25_CONFIG) response = client.post("/op25/set_active_config?restart=false", json=SAMPLE_P25_CONFIG)
@@ -226,19 +227,19 @@ def test_set_active_config_no_restart(mock_popen, mock_liquid, mock_white, mock_
# Verify config files were written # Verify config files were written
mock_file.assert_called_with('/configs/active.cfg.json', 'w') mock_file.assert_called_with('/configs/active.cfg.json', 'w')
mock_dump.assert_called_once() mock_dump.assert_called_once()
mock_tags.assert_called_with(SAMPLE_P25_CONFIG["tags"]) mock_tags.assert_called_with([MockTalkgroupTag(**t) for t in SAMPLE_P25_CONFIG["tags"]])
mock_white.assert_called_with(SAMPLE_P25_CONFIG["whitelist"]) mock_white.assert_called_with(SAMPLE_P25_CONFIG["whitelist"])
mock_liquid.assert_called_with(SAMPLE_P25_CONFIG["icecastConfig"]) mock_liquid.assert_called_with(MockIcecastConfig(**SAMPLE_P25_CONFIG["icecastConfig"]))
# Verify radio was NOT started/stopped # Verify radio was NOT started/stopped
mock_popen.assert_not_called() mock_popen.assert_not_called()
os.killpg.assert_not_called() os.killpg.assert_not_called()
@patch("app.routers.op25_controller.activate_config_from_library", return_value=True) @patch("routers.op25_controller.activate_config_from_library", return_value=True)
@patch("app.routers.op25_controller.save_config_to_library") @patch("routers.op25_controller.save_config_to_library")
@patch("app.routers.op25_controller.save_library_sidecars") @patch("routers.op25_controller.save_library_sidecars")
@patch("app.routers.op25_controller.subprocess.Popen") @patch("routers.op25_controller.subprocess.Popen")
def test_set_active_config_with_save_to_library(mock_popen, mock_save_sidecars, mock_save_lib, mock_activate): def test_set_active_config_with_save_to_library(mock_popen, mock_save_sidecars, mock_save_lib, mock_activate):
"""Test setting active config and saving it to the library.""" """Test setting active config and saving it to the library."""
library_name = "MyNewSystem" library_name = "MyNewSystem"
@@ -259,8 +260,8 @@ def test_set_active_config_with_save_to_library(mock_popen, mock_save_sidecars,
assert mock_popen.call_count == 1 assert mock_popen.call_count == 1
@patch("app.routers.op25_controller.activate_config_from_library", return_value=True) @patch("routers.op25_controller.activate_config_from_library", return_value=True)
@patch("app.routers.op25_controller.subprocess.Popen") @patch("routers.op25_controller.subprocess.Popen")
def test_load_from_library_success(mock_popen, mock_activate): def test_load_from_library_success(mock_popen, mock_activate):
"""Test loading a configuration from the library.""" """Test loading a configuration from the library."""
system_name = "ExistingSystem" system_name = "ExistingSystem"
@@ -274,7 +275,7 @@ def test_load_from_library_success(mock_popen, mock_activate):
assert mock_popen.call_count == 1 assert mock_popen.call_count == 1
@patch("app.routers.op25_controller.activate_config_from_library", return_value=False) @patch("routers.op25_controller.activate_config_from_library", return_value=False)
def test_load_from_library_not_found(mock_activate): def test_load_from_library_not_found(mock_activate):
"""Test loading a non-existent configuration from the library.""" """Test loading a non-existent configuration from the library."""
system_name = "NotFoundSystem" system_name = "NotFoundSystem"
@@ -284,8 +285,8 @@ def test_load_from_library_not_found(mock_activate):
assert "not found in library" in response.json()["detail"] assert "not found in library" in response.json()["detail"]
@patch("app.routers.op25_controller.save_config_to_library", return_value=True) @patch("routers.op25_controller.save_config_to_library", return_value=True)
@patch("app.routers.op25_controller.save_library_sidecars") @patch("routers.op25_controller.save_library_sidecars")
def test_save_to_library(mock_save_sidecars, mock_save_lib): def test_save_to_library(mock_save_sidecars, mock_save_lib):
"""Test saving a configuration directly to the library.""" """Test saving a configuration directly to the library."""
system_name = "NewLibSystem" system_name = "NewLibSystem"
@@ -297,7 +298,7 @@ def test_save_to_library(mock_save_sidecars, mock_save_lib):
mock_save_sidecars.assert_called_with(system_name, ANY) mock_save_sidecars.assert_called_with(system_name, ANY)
@patch("app.routers.op25_controller.scan_local_library", return_value=["System1.json", "System2.json"]) @patch("routers.op25_controller.scan_local_library", return_value=["System1.json", "System2.json"])
def test_get_library(mock_scan): def test_get_library(mock_scan):
"""Test the /library endpoint.""" """Test the /library endpoint."""
response = client.get("/op25/library") response = client.get("/op25/library")
@@ -306,7 +307,7 @@ def test_get_library(mock_scan):
mock_scan.assert_called_once() mock_scan.assert_called_once()
@patch("app.routers.op25_controller.build_op25_config", side_effect=Exception("Build failed")) @patch("routers.op25_controller.build_op25_config", side_effect=Exception("Build failed"))
def test_set_active_config_build_failure(mock_build): def test_set_active_config_build_failure(mock_build):
"""Test error handling when config building fails.""" """Test error handling when config building fails."""
response = client.post("/op25/set_active_config", json=SAMPLE_P25_CONFIG) response = client.post("/op25/set_active_config", json=SAMPLE_P25_CONFIG)