drb-edge-node/tests/test_op25_controller.py

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__), '..', 'app')))

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"]