Systems: P25 trunked / P25 conventional / analog as one choice (node-26#24)
- system_config.normalize(): a P25 system with config.mode "conventional"
carries config.channels [{freq MHz, name, nac?}] (1-2, within 1.5 MHz) and
can no longer carry control channels (422). A missing mode is trunked and
passes through untouched, so there is no migration. An analog channel is
mirrored into control_channels for older nodes.
- talkgroups.name_from_system resolves the edge node's synthetic per-channel
id (freq in 100 Hz units) to the channel name.
- The checkin persists radio_channels (conventional channels + encryption).
- Systems editor: Type = P25 trunked / P25 conventional / Analog (NBFM) /
DMR, plus a channel-row editor for conventional and a one-frequency form for
analog. The node page shows conventional channels and an encryption notice.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
ae0ece9d2c
commit
23860eb31b
@@ -0,0 +1,119 @@
|
||||
"""
|
||||
node-26#24 — system kinds: P25 trunked / P25 conventional / analog.
|
||||
|
||||
Missing mode = trunked and passes through untouched (no migration). A
|
||||
conventional system carries channels, never control channels, and the edge
|
||||
node's synthetic per-channel talkgroup id resolves back to the channel name.
|
||||
"""
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from app.main import app
|
||||
from app.internal import system_config as sc
|
||||
from app.internal import talkgroups
|
||||
from app.internal.auth import require_admin_token, require_node_service_or_firebase_token
|
||||
from app.routers import systems
|
||||
|
||||
client = TestClient(app)
|
||||
|
||||
WHITE_PLAINS = {
|
||||
"mode": "conventional",
|
||||
"channels": [
|
||||
{"freq": 460.4, "name": "PD1 Dispatch"},
|
||||
{"freq": "460.425", "name": " PD3 Tac ", "nac": "0x293"},
|
||||
],
|
||||
"talkgroups": [],
|
||||
}
|
||||
|
||||
|
||||
def setup_function():
|
||||
app.dependency_overrides[require_admin_token] = lambda: {"admin": True}
|
||||
app.dependency_overrides[require_node_service_or_firebase_token] = lambda: {"admin": True}
|
||||
|
||||
|
||||
def teardown_function():
|
||||
app.dependency_overrides.pop(require_admin_token, None)
|
||||
app.dependency_overrides.pop(require_node_service_or_firebase_token, None)
|
||||
|
||||
|
||||
def test_trunked_without_mode_is_untouched():
|
||||
cfg = {"control_channels": [851.0125], "nac": "0x293", "talkgroups": [{"id": 1, "name": "x"}]}
|
||||
assert sc.normalize("P25", cfg) == cfg
|
||||
|
||||
|
||||
def test_conventional_is_normalized():
|
||||
out = sc.normalize("P25", WHITE_PLAINS)
|
||||
assert out["mode"] == "conventional"
|
||||
assert out["channels"] == [
|
||||
{"freq": 460.4, "name": "PD1 Dispatch"},
|
||||
{"freq": 460.425, "name": "PD3 Tac", "nac": "293"},
|
||||
]
|
||||
assert "control_channels" not in out
|
||||
|
||||
|
||||
@pytest.mark.parametrize("cfg, needle", [
|
||||
({**WHITE_PLAINS, "control_channels": [460.4]}, "no control channel"),
|
||||
({"mode": "conventional", "channels": []}, "at least one"),
|
||||
({"mode": "conventional", "channels": [{"freq": 460.4}, {"freq": 460.425}, {"freq": 460.45}]}, "at most 2"),
|
||||
({"mode": "conventional", "channels": [{"freq": 460.4}, {"freq": 462.6}]}, "apart"),
|
||||
({"mode": "conventional", "channels": [{"freq": 460.4}, {"freq": 460.4}]}, "twice"),
|
||||
({"mode": "conventional", "channels": [{"freq": 460400000}]}, "MHz"),
|
||||
({"mode": "conventional", "channels": [{"freq": 460.4, "nac": "zz"}]}, "hex"),
|
||||
({"mode": "conventional", "channels": [{"freq": 460.4, "nac": "1000"}]}, "outside"),
|
||||
({"mode": "simulcast"}, "trunked"),
|
||||
])
|
||||
def test_conventional_rejections(cfg, needle):
|
||||
with pytest.raises(sc.SystemConfigError, match=needle):
|
||||
sc.normalize("P25", cfg)
|
||||
|
||||
|
||||
def test_analog_channel_mirrors_legacy_control_channel():
|
||||
out = sc.normalize("NBFM", {"channels": [{"freq": 154.43, "name": "Fire"}]})
|
||||
assert out["control_channels"] == [154.43]
|
||||
with pytest.raises(sc.SystemConfigError, match="exactly one"):
|
||||
sc.normalize("NBFM", {"channels": [{"freq": 154.43}, {"freq": 154.1}]})
|
||||
with pytest.raises(sc.SystemConfigError, match="at least one"):
|
||||
sc.normalize("NBFM", {"channels": []})
|
||||
with pytest.raises(sc.SystemConfigError, match="at least one"):
|
||||
sc.normalize("NBFM", {})
|
||||
# Legacy analog systems pass through untouched.
|
||||
assert sc.normalize("NBFM", {"control_channels": [154.43]}) == {"control_channels": [154.43]}
|
||||
|
||||
|
||||
def test_channel_tgid_matches_edge_node():
|
||||
# drb-edge-node internal/conventional.channel_tgid: 100 Hz units.
|
||||
assert sc.channel_tgid(460.4) == 4_604_000
|
||||
assert sc.channel_tgid(460_425_000) == 4_604_250
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_synthetic_talkgroup_resolves_to_channel_name():
|
||||
doc = {"config": sc.normalize("P25", WHITE_PLAINS)}
|
||||
with patch.object(talkgroups.fstore, "doc_get_cached", AsyncMock(return_value=doc)):
|
||||
assert await talkgroups.name_from_system("wp", 4_604_250) == "PD3 Tac"
|
||||
assert await talkgroups.name_from_system("wp", 101) is None
|
||||
|
||||
|
||||
def test_create_route_rejects_conventional_control_channels():
|
||||
with patch.object(systems.fstore, "doc_set", AsyncMock()) as doc_set:
|
||||
r = client.post("/systems", json={"name": "WP", "type": "P25",
|
||||
"config": {**WHITE_PLAINS, "control_channels": [460.4]}})
|
||||
assert r.status_code == 422
|
||||
assert "control channel" in r.json()["detail"]
|
||||
doc_set.assert_not_awaited()
|
||||
|
||||
|
||||
def test_update_route_normalizes_with_existing_type():
|
||||
existing = {"system_id": "wp", "name": "White Plains", "type": "P25",
|
||||
"config": {"control_channels": [460.4, 460.425]}}
|
||||
with patch.object(systems.fstore, "doc_get", AsyncMock(return_value=existing)), \
|
||||
patch.object(systems.fstore, "doc_update", AsyncMock()) as doc_update, \
|
||||
patch.object(systems.area_ctx, "schedule_refresh", lambda *_: None):
|
||||
r = client.put("/systems/wp", json={"name": "White Plains", "type": "P25", "config": WHITE_PLAINS})
|
||||
assert r.status_code == 200, r.text
|
||||
(_, _, patch_), _ = doc_update.await_args
|
||||
assert patch_["config"]["mode"] == "conventional"
|
||||
assert "control_channels" not in patch_["config"]
|
||||
assert [c["freq"] for c in patch_["config"]["channels"]] == [460.4, 460.425]
|
||||
Reference in New Issue
Block a user