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
import types
from typing import List, Optional
from pydantic import BaseModel

# 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')))
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))

# --- MOCK MODELS ---
# The actual models.models file has a NameError (IcecastConfig used before definition).
# Since we cannot edit the source code, we mock the module here to allow tests to run.
mock_models = types.ModuleType("models.models")

class MockTerminalConfig(BaseModel):
    pass

class MockTalkgroupTag(BaseModel):
    tagDec: int
    tagName: str

class MockDecodeMode:
    P25 = "P25"
    ANALOG = "ANALOG"

class MockIcecastConfig(BaseModel):
    icecast_host: str
    icecast_port: int
    icecast_mountpoint: str
    icecast_password: str

class MockAnalogConfig(BaseModel):
    systemName: str
    frequency: str
    nbfmSquelch: int

class MockConfigGenerator(BaseModel):
    type: str
    systemName: str
    channels: Optional[List[str]] = None
    tags: Optional[List[MockTalkgroupTag]] = None
    whitelist: Optional[str] = None
    icecastConfig: Optional[MockIcecastConfig] = None
    config: Optional[MockAnalogConfig] = None

class MockChannelConfig(BaseModel):
    name: Optional[str] = None
    trunking_sysname: Optional[str] = None
    enable_analog: Optional[str] = None
    demod_type: Optional[str] = None
    cqpsk_tracking: Optional[bool] = None
    filter_type: Optional[str] = None
    meta_stream_name: Optional[str] = None
    channelName: Optional[str] = None
    enableAnalog: Optional[str] = None
    demodType: Optional[str] = None
    frequency: Optional[str] = None
    filterType: Optional[str] = None
    nbfmSquelch: Optional[int] = None

class MockDeviceConfig(BaseModel):
    gain: Optional[str] = None

class MockTrunkingChannelConfig(BaseModel):
    sysname: str
    control_channel_list: str
    tagsFile: str
    whitelist: str

class MockTrunkingConfig(BaseModel):
    module: str
    chans: List[MockTrunkingChannelConfig]

class MockMetadataStreamConfig(BaseModel):
    stream_name: str
    icecastServerAddress: str
    icecastMountpoint: str
    icecastPass: str

class MockMetadataConfig(BaseModel):
    streams: List[MockMetadataStreamConfig]


mock_models.ConfigGenerator = MockConfigGenerator
mock_models.DecodeMode = MockDecodeMode
mock_models.ChannelConfig = MockChannelConfig
mock_models.DeviceConfig = MockDeviceConfig
mock_models.TrunkingConfig = MockTrunkingConfig
mock_models.TrunkingChannelConfig = MockTrunkingChannelConfig
mock_models.TerminalConfig = MockTerminalConfig
mock_models.MetadataConfig = MockMetadataConfig
mock_models.MetadataStreamConfig = MockMetadataStreamConfig
mock_models.IcecastConfig = MockIcecastConfig
mock_models.TalkgroupTag = MockTalkgroupTag

sys.modules["models.models"] = mock_models
sys.modules["models"] = types.ModuleType("models")
sys.modules["models"].models = mock_models
# -------------------

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": [{"tagDec": 101, "tagName": "Group A"}, {"tagDec": 102, "tagName": "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("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("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("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("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("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("routers.op25_controller.get_current_system_from_config", return_value="TestSystem")
@patch("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("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("routers.op25_controller.json.dump")
@patch("routers.op25_controller.save_talkgroup_tags")
@patch("routers.op25_controller.save_whitelist")
@patch("routers.op25_controller.generate_liquid_script")
@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):
    """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([MockTalkgroupTag(**t) for t in SAMPLE_P25_CONFIG["tags"]])
    mock_white.assert_called_with(SAMPLE_P25_CONFIG["whitelist"])
    mock_liquid.assert_called_with(MockIcecastConfig(**SAMPLE_P25_CONFIG["icecastConfig"]))

    # Verify radio was NOT started/stopped
    mock_popen.assert_not_called()
    os.killpg.assert_not_called()


@patch("routers.op25_controller.activate_config_from_library", return_value=True)
@patch("routers.op25_controller.save_config_to_library")
@patch("routers.op25_controller.save_library_sidecars")
@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):
    """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("routers.op25_controller.activate_config_from_library", return_value=True)
@patch("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("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("routers.op25_controller.save_config_to_library", return_value=True)
@patch("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("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("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"]