mirror of
https://github.com/zvx-echo6/central.git
synced 2026-06-10 11:54:37 +02:00
02bc692bda
New CENTRAL_TRAFFIC_CAMERAS stream + state_511_atis_cameras adapter. Telemetry
half of Castle Rock (events shipped in v0.9.2). Each Idaho camera -> one
telemetry event on /telemetry; detail drawer renders <img> direct from the
source (no blob storage / proxy in Central -- URL only).
supervisor + gui + ARCHIVE restart (NEW event-bearing stream
central.traffic_cameras.>). Ships disabled; public-unauth (no api key). Idaho only.
- Full camera list via POST /List/GetData/Cameras (DataTables), PAGINATED at
100/page (Idaho ~455 = 5 pages). GetUserCameras was a red herring (4 default
cams). The 100-row page cap also means v0.9.2 state_511_atis silently
truncates its 114-row Construction layer -> separate v0.9.7 fix.
- Subject central.traffic_cameras.{state}.{camera_id}; category
camera.state_511_atis_cameras -> GUI event_type "camera". data_class=telemetry.
- Per-UTC-day dedup {state}:cam:{id}:{YYYY-MM-DD}: one event per camera per day
-- always shows today's cameras, no per-poll flooding, no retention
coordination. Inherits the v0.9.1 dedup mixin.
- All sources included (Idaho511/ITDNET/RWIS/UDOT/ODOT/WYDOT/MTD border cameras);
source surfaced in data + the drawer for provenance. WKT POINT (lon lat) -> geo.
- No upstream image-capture timestamp (lastUpdated is config-edit time); drawer
shows no false "Captured" line. Cadence 600s. Severity 1 (telemetry).
Full suite: 829 passed, 1 skipped (central and unprivileged zvx, 3x each).
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
104 lines
4 KiB
Python
104 lines
4 KiB
Python
"""Tests for v0.7.4 telemetry/event separation: SourceAdapter.data_class,
|
|
registry split, class-scoped filter options, and the data_class SQL filter.
|
|
Registry-derived (no hardcoded adapter lists beyond the nwis pin). No live DB.
|
|
"""
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from central.adapter import SourceAdapter
|
|
from central.adapter_discovery import discover_adapters
|
|
from central.gui import routes
|
|
|
|
# Adapters with data_class="telemetry" (the pinned split; grow as telemetry adapters land).
|
|
_TELEMETRY = ["nwis", "state_511_atis_cameras", "tomtom_flow"]
|
|
|
|
|
|
# --- data_class defaults / registry split -----------------------------------
|
|
|
|
def test_base_default_is_event():
|
|
assert SourceAdapter.data_class == "event"
|
|
|
|
|
|
def test_registry_split_11_event_1_telemetry():
|
|
reg = discover_adapters()
|
|
by_class = {}
|
|
for name, cls in reg.items():
|
|
by_class.setdefault(getattr(cls, "data_class", "event"), []).append(name)
|
|
assert sorted(by_class.get("telemetry", [])) == _TELEMETRY
|
|
# Everything else is event-class; the split must cover the whole registry.
|
|
assert sorted(by_class.get("event", [])) == sorted(n for n in reg if n not in _TELEMETRY)
|
|
assert len(by_class.get("event", [])) == len(reg) - len(_TELEMETRY)
|
|
|
|
|
|
def test_class_adapter_names():
|
|
assert "nwis" not in routes._class_adapter_names("event")
|
|
assert sorted(routes._class_adapter_names("telemetry")) == _TELEMETRY
|
|
assert "usgs_quake" in routes._class_adapter_names("event")
|
|
|
|
|
|
# --- class-scoped chip-picker / legend options -------------------------------
|
|
|
|
def test_event_options_exclude_nwis():
|
|
flat, grouped = routes._adapter_filter_options("event")
|
|
names = {a["name"] for a in flat}
|
|
assert "nwis" not in names
|
|
assert len(flat) == len(discover_adapters()) - len(_TELEMETRY)
|
|
grouped_values = {opt["value"] for _, items in grouped for opt in items}
|
|
assert "nwis" not in grouped_values
|
|
|
|
|
|
def test_telemetry_options_only_nwis():
|
|
flat, grouped = routes._adapter_filter_options("telemetry")
|
|
assert sorted(a["name"] for a in flat) == _TELEMETRY
|
|
grouped_values = [opt["value"] for _, items in grouped for opt in items]
|
|
assert sorted(grouped_values) == _TELEMETRY
|
|
|
|
|
|
def test_colors_stable_across_classes():
|
|
"""A given adapter keeps the same color on /events and /telemetry (colors
|
|
are keyed to the full registry, not the per-tab subset)."""
|
|
full, _ = routes._adapter_filter_options()
|
|
full_color = {a["name"]: a["color"] for a in full}
|
|
ev, _ = routes._adapter_filter_options("event")
|
|
for a in ev:
|
|
assert a["color"] == full_color[a["name"]]
|
|
|
|
|
|
# --- data_class SQL filter (captured SQL) ------------------------------------
|
|
|
|
async def _capture(parsed):
|
|
captured = {}
|
|
|
|
async def fake_fetch(query, *args):
|
|
captured["query"] = query
|
|
captured["params"] = list(args)
|
|
return []
|
|
|
|
conn = MagicMock()
|
|
conn.fetch = fake_fetch
|
|
pool = MagicMock()
|
|
pool.acquire.return_value.__aenter__ = AsyncMock(return_value=conn)
|
|
pool.acquire.return_value.__aexit__ = AsyncMock(return_value=None)
|
|
with patch("central.gui.routes.get_pool", return_value=pool):
|
|
await routes._fetch_events(parsed)
|
|
return captured
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_class_adapters_adds_adapter_any_condition():
|
|
parsed, _ = routes._parse_events_params({"time": "all"}, default_offset=0)
|
|
parsed["class_adapters"] = routes._class_adapter_names("event")
|
|
cap = await _capture(parsed)
|
|
assert "adapter = ANY($" in cap["query"]
|
|
assert routes._class_adapter_names("event") in cap["params"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_class_adapters_no_class_condition():
|
|
"""events.json path: no class_adapters -> no extra adapter filter (all classes)."""
|
|
parsed, _ = routes._parse_events_params({"time": "all"}) # cursor-mode, no class
|
|
assert parsed.get("class_adapters") is None
|
|
cap = await _capture(parsed)
|
|
# The only adapter=ANY would come from a user filter, which we didn't set.
|
|
assert "adapter = ANY($" not in cap["query"]
|