init testing

tests/test_op25_controller.py

@@ -0,0 +1,221 @@
+import pytest
+from fastapi.testclient import TestClient
+from unittest.mock import patch, MagicMock, mock_open, ANY
+import json
+import os
+
+# The router is included in the main app, so we test through it.
+# We need to adjust the python path for imports to work correctly
+import sys
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
+
+from app.node_main import app
+
+# Use a client to make requests to the app
+client = TestClient(app)
+
+# Define a sample P25 config payload for testing
+SAMPLE_P25_CONFIG = {
+    "type": "P25",
+    "systemName": "TestSystem",
+    "channels": ["851.12345", "852.67890"],
+    "tags": "101,Group A\n102,Group B",
+    "whitelist": "101",
+    "icecastConfig": {
+        "icecast_host": "localhost",
+        "icecast_port": 8000,
+        "icecast_mountpoint": "test",
+        "icecast_password": "hackme"
+    }
+}
+
+@pytest.fixture(autouse=True)
+def reset_and_mock_globals(monkeypatch):
+    """
+    Fixture to reset the global op25_process state and mock dependencies
+    before each test, ensuring test isolation.
+    """
+    # Reset the global process variable in the controller module
+    monkeypatch.setattr("app.routers.op25_controller.op25_process", None)
+
+    # Mock asyncio.sleep to prevent tests from actually waiting
+    mock_sleep = MagicMock()
+    monkeypatch.setattr("asyncio.sleep", mock_sleep)
+
+    # Mock os functions related to process groups
+    monkeypatch.setattr("os.killpg", MagicMock())
+    monkeypatch.setattr("os.getpgid", MagicMock(return_value=12345))
+
+
+@patch("app.routers.op25_controller.subprocess.Popen")
+def test_start_op25_success(mock_popen):
+    """Test the /start endpoint successfully starts the process."""
+    mock_process = MagicMock()
+    mock_process.pid = 12345
+    mock_popen.return_value = mock_process
+
+    response = client.post("/op25/start")
+    assert response.status_code == 200
+    assert response.json() == {"status": "OP25 started"}
+    mock_popen.assert_called_once()
+
+
+@patch("app.routers.op25_controller.subprocess.Popen", side_effect=Exception("Popen failed"))
+def test_start_op25_failure(mock_popen):
+    """Test the /start endpoint when Popen raises an exception."""
+    response = client.post("/op25/start")
+    assert response.status_code == 500
+    assert "Failed to start OP25" in response.json()["detail"]
+
+
+def test_stop_op25_not_running():
+    """Test the /stop endpoint when the process is not running."""
+    response = client.post("/op25/stop")
+    assert response.status_code == 200
+    assert response.json() == {"status": "OP25 was not running"}
+
+
+@patch("app.routers.op25_controller.subprocess.Popen")
+def test_stop_op25_success(mock_popen, monkeypatch):
+    """Test the /stop endpoint successfully stops a running process."""
+    mock_process = MagicMock()
+    mock_process.pid = 12345
+    mock_process.poll.return_value = None  # Indicates it's running
+    monkeypatch.setattr("app.routers.op25_controller.op25_process", mock_process)
+
+    response = client.post("/op25/stop")
+    assert response.status_code == 200
+    assert response.json() == {"status": "OP25 stopped"}
+    os.killpg.assert_called_with(os.getpgid(mock_process.pid), ANY)
+
+
+def test_get_status_not_running():
+    """Test the /status endpoint when the process is not running."""
+    response = client.get("/op25/status")
+    assert response.status_code == 200
+    data = response.json()
+    assert data["is_running"] is False
+    assert data["pid"] is None
+    assert data["active_system"] is None
+
+
+@patch("app.routers.op25_controller.get_current_system_from_config", return_value="TestSystem")
+@patch("app.routers.op25_controller.subprocess.Popen")
+def test_get_status_running(mock_popen, mock_get_system, monkeypatch):
+    """Test the /status endpoint when the process is running."""
+    mock_process = MagicMock()
+    mock_process.pid = 12345
+    mock_process.poll.return_value = None # Running
+    monkeypatch.setattr("app.routers.op25_controller.op25_process", mock_process)
+
+    response = client.get("/op25/status")
+    assert response.status_code == 200
+    data = response.json()
+    assert data["is_running"] is True
+    assert data["pid"] == 12345
+    assert data["active_system"] == "TestSystem"
+    mock_get_system.assert_called_once()
+
+
+@patch("builtins.open", new_callable=mock_open)
+@patch("app.routers.op25_controller.json.dump")
+@patch("app.routers.op25_controller.save_talkgroup_tags")
+@patch("app.routers.op25_controller.save_whitelist")
+@patch("app.routers.op25_controller.generate_liquid_script")
+@patch("app.routers.op25_controller.subprocess.Popen")
+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."""
+    response = client.post("/op25/set_active_config?restart=false", json=SAMPLE_P25_CONFIG)
+
+    assert response.status_code == 200
+    assert response.json() == {"message": "Active configuration updated", "radio_restarted": False}
+
+    # Verify config files were written
+    mock_file.assert_called_with('/configs/active.cfg.json', 'w')
+    mock_dump.assert_called_once()
+    mock_tags.assert_called_with(SAMPLE_P25_CONFIG["tags"])
+    mock_white.assert_called_with(SAMPLE_P25_CONFIG["whitelist"])
+    mock_liquid.assert_called_with(SAMPLE_P25_CONFIG["icecastConfig"])
+
+    # Verify radio was NOT started/stopped
+    mock_popen.assert_not_called()
+    os.killpg.assert_not_called()
+
+
+@patch("app.routers.op25_controller.activate_config_from_library", return_value=True)
+@patch("app.routers.op25_controller.save_config_to_library")
+@patch("app.routers.op25_controller.save_library_sidecars")
+@patch("app.routers.op25_controller.subprocess.Popen")
+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."""
+    library_name = "MyNewSystem"
+    response = client.post(
+        f"/op25/set_active_config?restart=true&save_to_library_name={library_name}",
+        json=SAMPLE_P25_CONFIG
+    )
+
+    assert response.status_code == 200
+    assert response.json()["radio_restarted"] is True
+
+    # Verify it was saved and then activated from the library
+    mock_save_lib.assert_called_with(library_name, ANY)
+    mock_save_sidecars.assert_called_with(library_name, ANY)
+    mock_activate.assert_called_with(library_name)
+
+    # Verify radio was restarted
+    assert mock_popen.call_count == 1
+
+
+@patch("app.routers.op25_controller.activate_config_from_library", return_value=True)
+@patch("app.routers.op25_controller.subprocess.Popen")
+def test_load_from_library_success(mock_popen, mock_activate):
+    """Test loading a configuration from the library."""
+    system_name = "ExistingSystem"
+    response = client.post(f"/op25/load_from_library?system_name={system_name}")
+
+    assert response.status_code == 200
+    assert response.json() == {"status": f"Loaded and started library config: {system_name}"}
+
+    # Verify activation and restart
+    mock_activate.assert_called_with(system_name)
+    assert mock_popen.call_count == 1
+
+
+@patch("app.routers.op25_controller.activate_config_from_library", return_value=False)
+def test_load_from_library_not_found(mock_activate):
+    """Test loading a non-existent configuration from the library."""
+    system_name = "NotFoundSystem"
+    response = client.post(f"/op25/load_from_library?system_name={system_name}")
+
+    assert response.status_code == 404
+    assert "not found in library" in response.json()["detail"]
+
+
+@patch("app.routers.op25_controller.save_config_to_library", return_value=True)
+@patch("app.routers.op25_controller.save_library_sidecars")
+def test_save_to_library(mock_save_sidecars, mock_save_lib):
+    """Test saving a configuration directly to the library."""
+    system_name = "NewLibSystem"
+    response = client.post(f"/op25/save_to_library?system_name={system_name}", json=SAMPLE_P25_CONFIG)
+
+    assert response.status_code == 200
+    assert response.json() == {"status": f"Config saved as {system_name}"}
+    mock_save_lib.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"])
+def test_get_library(mock_scan):
+    """Test the /library endpoint."""
+    response = client.get("/op25/library")
+    assert response.status_code == 200
+    assert response.json() == ["System1.json", "System2.json"]
+    mock_scan.assert_called_once()
+
+
+@patch("app.routers.op25_controller.build_op25_config", side_effect=Exception("Build failed"))
+def test_set_active_config_build_failure(mock_build):
+    """Test error handling when config building fails."""
+    response = client.post("/op25/set_active_config", json=SAMPLE_P25_CONFIG)
+    assert response.status_code == 500
+    assert "Configuration error: Build failed" in response.json()["detail"]