249 lines
10 KiB
Python
249 lines
10 KiB
Python
"""
|
|
Local dashboard: the health checklist (/api/health) and choosing what to
|
|
listen to without C2 (/api/listen, /api/config/system, /api/config/revert).
|
|
|
|
Same standalone-app pattern as test_auth_endpoints.py — the real routers,
|
|
none of app.main's lifespan. OP25/MQTT are mocked at the client boundary.
|
|
"""
|
|
import base64
|
|
import time
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.config import settings
|
|
from app.internal import config_manager, credentials, health
|
|
from app.internal.op25_client import parse_terminal_messages
|
|
from app.routers import api, ui
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def isolated(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(credentials, "_CREDS_FILE", tmp_path / "credentials.json")
|
|
monkeypatch.setattr(credentials, "_api_key", None)
|
|
monkeypatch.setattr(credentials, "_auth_salt", None)
|
|
monkeypatch.setattr(credentials, "_session_secret", None)
|
|
monkeypatch.setattr(config_manager, "_CONFIG_FILE", tmp_path / "node_config.json")
|
|
monkeypatch.setattr(settings, "dashboard_username", "tester")
|
|
monkeypatch.setattr(settings, "dashboard_password", "s3cret-pass")
|
|
yield
|
|
|
|
|
|
@pytest.fixture
|
|
def client():
|
|
app = FastAPI()
|
|
app.include_router(api.router)
|
|
app.include_router(ui.router)
|
|
auth = base64.b64encode(b"tester:s3cret-pass").decode()
|
|
with TestClient(app, headers={"Authorization": f"Basic {auth}"}) as c:
|
|
yield c
|
|
|
|
|
|
@pytest.fixture
|
|
def op25():
|
|
"""Mock the OP25 side of an apply: generate, stop, start, checkin."""
|
|
with patch("app.main._generate_op25_config", AsyncMock(return_value=True)) as gen, \
|
|
patch("app.routers.api.op25_client.stop", AsyncMock(return_value=True)) as stop, \
|
|
patch("app.routers.api.op25_client.start_with_reason", AsyncMock(return_value=None)) as start, \
|
|
patch("app.routers.api.asyncio.sleep", AsyncMock()), \
|
|
patch("app.routers.api.mqtt_manager.publish_checkin", AsyncMock()) as checkin:
|
|
yield {"gen": gen, "stop": stop, "start": start, "checkin": checkin}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Health checklist — pure function of facts
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _facts(**over):
|
|
now = 1_000_000.0
|
|
f = {
|
|
"now": now, "node_id": "toughny", "default_password": False,
|
|
"c2_url": "https://api.example", "has_api_key": True, "mqtt_connected": True,
|
|
"mqtt_broker": "mqtt.example", "mqtt_port": 8883, "sdr_count": 1,
|
|
"source": {"name": "County", "type": "P25", "system_id": "s1", "local": False},
|
|
"op25_status": "running",
|
|
"console": {"last_poll_epoch": now - 1, "channels": [], "trunk": [{"last_tsbk": now - 1, "rxchan": 851012500}],
|
|
"last_heard": None},
|
|
"audio_capturing": True, "last_voice_epoch": now - 30, "icecast_live": True, "icecast_port": 8000,
|
|
}
|
|
f.update(over)
|
|
return f
|
|
|
|
|
|
def _by_id(checks):
|
|
return {c["id"]: c for c in checks}
|
|
|
|
|
|
def test_healthy_node_is_all_ok():
|
|
checks = health.build_checks(_facts())
|
|
assert all(c["state"] == "ok" for c in checks), [c for c in checks if c["state"] != "ok"]
|
|
assert "851.0125 MHz" in _by_id(checks)["decode"]["summary"]
|
|
|
|
|
|
def test_unapproved_node_explains_wait_and_mqtt_refusal():
|
|
"""Toughny's state: C2 set, no API key, broker refuses, nothing chosen."""
|
|
c = _by_id(health.build_checks(_facts(has_api_key=False, mqtt_connected=False, source=None,
|
|
op25_status="stopped")))
|
|
assert c["c2"]["state"] == "warn" and "approve" in c["c2"]["action"]
|
|
assert "toughny" in c["c2"]["action"]
|
|
assert c["mqtt"]["state"] == "warn"
|
|
assert c["source"]["state"] == "warn" and "Listen locally" in c["source"]["action"]
|
|
assert c["op25"]["state"] == "off"
|
|
assert c["decode"]["state"] == "off"
|
|
|
|
|
|
def test_key_on_disk_but_mqtt_down_says_restart():
|
|
c = _by_id(health.build_checks(_facts(mqtt_connected=False)))
|
|
assert c["mqtt"]["state"] == "error"
|
|
assert "restart the edge-node container" in c["mqtt"]["action"]
|
|
|
|
|
|
def test_standalone_node_is_not_an_error():
|
|
c = _by_id(health.build_checks(_facts(c2_url=None, has_api_key=False, mqtt_connected=False)))
|
|
assert c["c2"]["state"] == "off"
|
|
assert c["mqtt"]["state"] == "off"
|
|
|
|
|
|
def test_no_sdr_and_crashed_op25():
|
|
c = _by_id(health.build_checks(_facts(sdr_count=0, op25_status="crashed")))
|
|
assert c["sdr"]["state"] == "error"
|
|
assert c["op25"]["state"] == "error" and "log" in c["op25"]["action"]
|
|
|
|
|
|
def test_running_but_no_control_channel():
|
|
now = 1_000_000.0
|
|
c = _by_id(health.build_checks(_facts(console={"last_poll_epoch": now, "channels": [],
|
|
"trunk": [{"last_tsbk": 0}], "last_heard": None})))
|
|
assert c["decode"]["state"] == "error"
|
|
|
|
|
|
def test_analog_needs_no_control_channel():
|
|
now = 1_000_000.0
|
|
c = _by_id(health.build_checks(_facts(
|
|
source={"name": "Fire", "type": "NBFM", "system_id": "local-analog", "local": True},
|
|
console={"last_poll_epoch": now, "channels": [{"freq": 154430000}], "trunk": [], "last_heard": None},
|
|
)))
|
|
assert c["decode"]["state"] == "ok" and "154.4300 MHz" in c["decode"]["summary"]
|
|
assert "not recorded" in c["decode"]["action"]
|
|
assert "this dashboard" in c["source"]["summary"]
|
|
|
|
|
|
def test_op25_unreachable_and_icecast_down():
|
|
c = _by_id(health.build_checks(_facts(op25_status=None, icecast_live=None)))
|
|
assert c["op25"]["state"] == "error"
|
|
assert c["stream"]["state"] == "error"
|
|
|
|
|
|
def test_default_password_warns():
|
|
assert _by_id(health.build_checks(_facts(default_password=True)))["password"]["state"] == "warn"
|
|
|
|
|
|
def test_trunk_update_is_parsed():
|
|
update = parse_terminal_messages([
|
|
{"json_type": "trunk_update", "0": {"last_tsbk": 5.0, "rxchan": 851012500}, "nac": 0},
|
|
{"json_type": "channel_update", "channels": ["0"], "0": {"freq": 1, "tgid": None}},
|
|
])
|
|
assert update.trunk == [{"last_tsbk": 5.0, "rxchan": 851012500}]
|
|
assert len(update.channels) == 1
|
|
|
|
|
|
def test_health_route_returns_checks(client):
|
|
facts = _facts(now=time.time())
|
|
with patch("app.internal.health.gather_facts", AsyncMock(return_value=facts)):
|
|
r = client.get("/api/health")
|
|
assert r.status_code == 200
|
|
assert [c["id"] for c in r.json()["checks"]][:3] == ["password", "c2", "mqtt"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Choosing what to listen to
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_listen_p25_saves_override_and_restarts_op25(client, op25):
|
|
r = client.post("/api/listen", json={
|
|
"mode": "p25", "frequencies": [851.0125, 851.5125], "talkgroups": [{"id": 1001, "name": "Fire"}],
|
|
})
|
|
assert r.status_code == 200, r.text
|
|
cfg = config_manager.load_node_config()
|
|
assert cfg.override_config.type == "P25"
|
|
assert cfg.override_config.config["control_channels"] == [851.0125, 851.5125]
|
|
assert cfg.override_config.config["talkgroups"] == [{"id": 1001, "name": "Fire"}]
|
|
op25["gen"].assert_awaited_once()
|
|
op25["start"].assert_awaited_once()
|
|
op25["checkin"].assert_not_awaited() # MQTT not connected in tests
|
|
|
|
|
|
def test_listen_analog_single_frequency(client, op25):
|
|
r = client.post("/api/listen", json={"mode": "analog", "frequencies": [154.43], "name": "Fire"})
|
|
assert r.status_code == 200, r.text
|
|
cfg = config_manager.load_node_config().override_config
|
|
assert (cfg.type, cfg.name, cfg.config["control_channels"]) == ("NBFM", "Fire", [154.43])
|
|
|
|
|
|
@pytest.mark.parametrize("body,needle", [
|
|
({"mode": "p25", "frequencies": []}, "at least one"),
|
|
({"mode": "p25", "frequencies": [851012500]}, "MHz"),
|
|
({"mode": "analog", "frequencies": [154.43, 155.1]}, "exactly one"),
|
|
({"mode": "system", "system_id": "nope"}, "saved list"),
|
|
])
|
|
def test_listen_rejects_bad_input_with_a_readable_reason(client, op25, body, needle):
|
|
r = client.post("/api/listen", json=body)
|
|
assert r.status_code in (400, 404)
|
|
assert needle in r.json()["detail"]
|
|
op25["gen"].assert_not_awaited()
|
|
|
|
|
|
def test_listen_reports_op25_start_failure(client, op25):
|
|
op25["start"].return_value = "No OP25 config yet — choose a system or frequency first."
|
|
r = client.post("/api/listen", json={"mode": "analog", "frequencies": [154.43]})
|
|
assert r.status_code == 502
|
|
assert "did not start" in r.json()["detail"]
|
|
|
|
|
|
def test_dmr_is_refused_up_front(client, op25):
|
|
r = client.post("/api/config/override", json={"system_config": {
|
|
"system_id": "x", "name": "x", "type": "DMR", "config": {"control_channels": [450.1]},
|
|
}})
|
|
assert r.status_code == 400 and "DMR" in r.json()["detail"]
|
|
op25["gen"].assert_not_awaited()
|
|
|
|
|
|
def test_config_system_goes_through_op25_generation(client, op25, tmp_path, monkeypatch):
|
|
"""It used to dump the raw blob into active.cfg.json — not an OP25 config."""
|
|
monkeypatch.setattr(settings, "config_path", str(tmp_path))
|
|
r = client.post("/api/config/system", json={
|
|
"system_id": "s1", "name": "County", "type": "P25", "config": {"control_channels": [851.0125]},
|
|
})
|
|
assert r.status_code == 200, r.text
|
|
op25["gen"].assert_awaited_once()
|
|
assert not (tmp_path / "active.cfg.json").exists()
|
|
cfg = config_manager.load_node_config()
|
|
assert cfg.configured and cfg.system_config.name == "County"
|
|
|
|
|
|
def test_revert_without_c2_system_stops_op25(client, op25):
|
|
client.post("/api/listen", json={"mode": "analog", "frequencies": [154.43]})
|
|
op25["stop"].reset_mock()
|
|
r = client.post("/api/config/revert")
|
|
assert r.status_code == 200
|
|
assert config_manager.load_node_config().override_config is None
|
|
op25["stop"].assert_awaited_once()
|
|
|
|
|
|
def test_status_exposes_stream_parts_and_source(client, op25):
|
|
client.post("/api/listen", json={"mode": "analog", "frequencies": [154.43]})
|
|
with patch("app.routers.api.op25_client.status", AsyncMock(return_value={"status": "running"})):
|
|
d = client.get("/api/status").json()
|
|
assert d["icecast_port"] == settings.icecast_port and d["icecast_mount"] == settings.icecast_mount
|
|
assert d["local_choice"] is True and d["has_c2_system"] is False
|
|
assert d["system_type"] == "NBFM"
|
|
|
|
|
|
def test_dashboard_page_has_no_dead_api_instruction(client):
|
|
client.post("/login", data={"username": "tester", "password": "s3cret-pass"}, follow_redirects=False)
|
|
html = client.get("/").text
|
|
assert "POST /api/config/system" not in html
|
|
assert "/api/listen" in html and "/api/health" in html
|