alexei.dolgolyov
abc204c04e
The aggregated /api/v1/snapshot poll now emits a `scene_playlists` section (each playlist with its `is_running` flag) plus a companion `playlist_state` key carrying the single global cycling state (running playlist, current index/ preset, dwell) — so the HA-coordinator and other low-overhead pollers get playlist state in the same round trip as scenes/targets, matching the other entity sections. Gated by the `scene_playlists` include-section like the rest. Reuses the existing list_scene_playlists handler; snapshot route tests updated.
249 lines
9.4 KiB
Python
249 lines
9.4 KiB
Python
"""Tests for the aggregated /api/v1/snapshot endpoint.
|
|
|
|
The snapshot collapses the integration's per-target/per-device poll fan-out
|
|
into one response. These tests build a minimal app with the snapshot router and
|
|
override the store/manager getters, mirroring tests/api/routes/test_devices_routes.py.
|
|
Auth is left real (the conftest patches a test API key) so the rejection path is
|
|
also covered. System metrics + health run for real — they read module-level
|
|
providers and the patched config, no lifespan needed.
|
|
"""
|
|
|
|
import types
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
from fastapi import FastAPI, HTTPException
|
|
from fastapi.testclient import TestClient
|
|
|
|
from ledgrab.api import dependencies as deps
|
|
from ledgrab.api.routes import devices as devices_mod
|
|
from ledgrab.api.routes.devices import resolve_device_brightness
|
|
from ledgrab.api.routes.snapshot import SNAPSHOT_SECTIONS, _resolve_sections, router
|
|
from ledgrab.core.processing.processor_manager import ProcessorManager
|
|
from ledgrab.core.update.update_service import UpdateService
|
|
|
|
_AUTH = {"Authorization": "Bearer test-api-key-12345"}
|
|
|
|
_TOP_LEVEL_KEYS = (
|
|
"targets",
|
|
"target_states",
|
|
"target_metrics",
|
|
"devices",
|
|
"device_brightness",
|
|
"css_sources",
|
|
"value_sources",
|
|
"scene_presets",
|
|
"scene_playlists",
|
|
"playlist_state",
|
|
"sync_clocks",
|
|
"system",
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def client(test_config, monkeypatch):
|
|
# Pin the global config (with its test API key) so auth is deterministic
|
|
# regardless of test ordering — other suites mutate the config singleton.
|
|
import ledgrab.config as config_mod
|
|
|
|
monkeypatch.setattr(config_mod, "config", test_config)
|
|
|
|
target_store = MagicMock()
|
|
target_store.get_all_targets.return_value = []
|
|
device_store = MagicMock()
|
|
device_store.get_all_devices.return_value = []
|
|
css_store = MagicMock()
|
|
css_store.get_all_sources.return_value = []
|
|
value_store = MagicMock()
|
|
value_store.get_all_sources.return_value = []
|
|
preset_store = MagicMock()
|
|
preset_store.get_all_presets.return_value = []
|
|
playlist_store = MagicMock()
|
|
playlist_store.get_all_playlists.return_value = []
|
|
playlist_engine = MagicMock()
|
|
playlist_engine.get_running_playlist_id.return_value = None
|
|
playlist_engine.get_state.return_value = {
|
|
"is_running": False,
|
|
"playlist_id": None,
|
|
"playlist_name": None,
|
|
"current_index": 0,
|
|
"item_count": 0,
|
|
"current_preset_id": None,
|
|
"started_at": None,
|
|
"step_started_at": None,
|
|
"step_duration": 0.0,
|
|
}
|
|
clock_store = MagicMock()
|
|
clock_store.get_all_clocks.return_value = []
|
|
clock_manager = MagicMock()
|
|
|
|
manager = MagicMock(spec=ProcessorManager)
|
|
manager.get_all_target_states.return_value = {}
|
|
manager.get_all_target_metrics.return_value = {}
|
|
|
|
update_service = MagicMock(spec=UpdateService)
|
|
update_service.get_status.return_value = {"has_update": False, "current_version": "test"}
|
|
|
|
app = FastAPI()
|
|
app.include_router(router)
|
|
app.dependency_overrides[deps.get_output_target_store] = lambda: target_store
|
|
app.dependency_overrides[deps.get_device_store] = lambda: device_store
|
|
app.dependency_overrides[deps.get_color_strip_store] = lambda: css_store
|
|
app.dependency_overrides[deps.get_value_source_store] = lambda: value_store
|
|
app.dependency_overrides[deps.get_scene_preset_store] = lambda: preset_store
|
|
app.dependency_overrides[deps.get_scene_playlist_store] = lambda: playlist_store
|
|
app.dependency_overrides[deps.get_playlist_engine] = lambda: playlist_engine
|
|
app.dependency_overrides[deps.get_sync_clock_store] = lambda: clock_store
|
|
app.dependency_overrides[deps.get_sync_clock_manager] = lambda: clock_manager
|
|
app.dependency_overrides[deps.get_processor_manager] = lambda: manager
|
|
app.dependency_overrides[deps.get_update_service] = lambda: update_service
|
|
|
|
return TestClient(app, raise_server_exceptions=False)
|
|
|
|
|
|
def test_snapshot_returns_all_sections(client):
|
|
resp = client.get("/api/v1/snapshot", headers=_AUTH)
|
|
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
|
|
for key in _TOP_LEVEL_KEYS:
|
|
assert key in data, f"snapshot missing top-level key: {key}"
|
|
|
|
for list_key in (
|
|
"targets",
|
|
"devices",
|
|
"css_sources",
|
|
"value_sources",
|
|
"scene_presets",
|
|
"scene_playlists",
|
|
"sync_clocks",
|
|
):
|
|
assert data[list_key] == []
|
|
for dict_key in ("target_states", "target_metrics", "device_brightness"):
|
|
assert data[dict_key] == {}
|
|
|
|
# The single global cycling state rides along with the playlist list.
|
|
assert data["playlist_state"]["is_running"] is False
|
|
|
|
|
|
def test_snapshot_system_block_has_health_version(client):
|
|
data = client.get("/api/v1/snapshot", headers=_AUTH).json()
|
|
|
|
system = data["system"]
|
|
assert {"performance", "health", "update"}.issubset(system)
|
|
# health drives the coordinator's version + boot-time derivation
|
|
assert system["health"]["version"]
|
|
assert "uptime_seconds" in system["health"]
|
|
|
|
|
|
def test_snapshot_requires_auth(client):
|
|
resp = client.get("/api/v1/snapshot")
|
|
|
|
assert resp.status_code in (401, 403)
|
|
|
|
|
|
def test_snapshot_include_filters_to_requested_sections(client):
|
|
resp = client.get("/api/v1/snapshot", params={"include": "devices,system"}, headers=_AUTH)
|
|
|
|
assert resp.status_code == 200
|
|
# Only requested sections are present — excluded ones are omitted entirely.
|
|
assert set(resp.json().keys()) == {"devices", "system"}
|
|
|
|
|
|
def test_snapshot_include_rejects_unknown_section(client):
|
|
resp = client.get("/api/v1/snapshot", params={"include": "devices,bogus"}, headers=_AUTH)
|
|
|
|
assert resp.status_code == 422
|
|
assert "bogus" in resp.json()["detail"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _resolve_sections — query-param parsing edge cases
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_resolve_sections_defaults_to_all_when_empty():
|
|
assert _resolve_sections(None) == frozenset(SNAPSHOT_SECTIONS)
|
|
assert _resolve_sections("") == frozenset(SNAPSHOT_SECTIONS)
|
|
|
|
|
|
def test_resolve_sections_strips_whitespace_and_dedupes():
|
|
assert _resolve_sections("devices, system ,devices") == frozenset({"devices", "system"})
|
|
|
|
|
|
def test_resolve_sections_ignores_empty_segments():
|
|
assert _resolve_sections("devices,,system,") == frozenset({"devices", "system"})
|
|
|
|
|
|
def test_resolve_sections_is_case_sensitive():
|
|
with pytest.raises(HTTPException) as exc:
|
|
_resolve_sections("Devices")
|
|
assert exc.value.status_code == 422
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# resolve_device_brightness — cached / cold-fetch / graceful-degrade paths
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _fake_device(**kw):
|
|
return types.SimpleNamespace(
|
|
id=kw.get("id", "d1"),
|
|
device_type=kw.get("device_type", "wled"),
|
|
url=kw.get("url", "http://x"),
|
|
software_brightness=kw.get("software_brightness", 42),
|
|
)
|
|
|
|
|
|
async def test_brightness_none_without_capability(monkeypatch):
|
|
monkeypatch.setattr(devices_mod, "get_device_capabilities", lambda _dt: set())
|
|
manager = MagicMock()
|
|
|
|
assert await resolve_device_brightness(_fake_device(), manager) is None
|
|
manager.find_device_state.assert_not_called()
|
|
|
|
|
|
async def test_brightness_returns_cached(monkeypatch):
|
|
monkeypatch.setattr(devices_mod, "get_device_capabilities", lambda _dt: {"brightness_control"})
|
|
manager = MagicMock()
|
|
manager.find_device_state.return_value = types.SimpleNamespace(hardware_brightness=128)
|
|
|
|
assert await resolve_device_brightness(_fake_device(), manager) == 128
|
|
|
|
|
|
async def test_brightness_active_fetch_and_caches_on_cold_cache(monkeypatch):
|
|
monkeypatch.setattr(devices_mod, "get_device_capabilities", lambda _dt: {"brightness_control"})
|
|
provider = MagicMock()
|
|
provider.get_brightness = AsyncMock(return_value=200)
|
|
monkeypatch.setattr(devices_mod, "get_provider", lambda _dt: provider)
|
|
ds = types.SimpleNamespace(hardware_brightness=None)
|
|
manager = MagicMock()
|
|
manager.find_device_state.return_value = ds
|
|
|
|
assert await resolve_device_brightness(_fake_device(), manager) == 200
|
|
assert ds.hardware_brightness == 200 # cached so the next poll is I/O-free
|
|
|
|
|
|
async def test_brightness_degrades_to_none_on_provider_error(monkeypatch):
|
|
monkeypatch.setattr(devices_mod, "get_device_capabilities", lambda _dt: {"brightness_control"})
|
|
provider = MagicMock()
|
|
provider.get_brightness = AsyncMock(side_effect=OSError("unreachable"))
|
|
monkeypatch.setattr(devices_mod, "get_provider", lambda _dt: provider)
|
|
manager = MagicMock()
|
|
manager.find_device_state.return_value = types.SimpleNamespace(hardware_brightness=None)
|
|
|
|
# A single unreachable device must not raise — it degrades to None.
|
|
assert await resolve_device_brightness(_fake_device(), manager) is None
|
|
|
|
|
|
async def test_brightness_falls_back_to_software_when_unsupported(monkeypatch):
|
|
monkeypatch.setattr(devices_mod, "get_device_capabilities", lambda _dt: {"brightness_control"})
|
|
provider = MagicMock()
|
|
provider.get_brightness = AsyncMock(side_effect=NotImplementedError)
|
|
monkeypatch.setattr(devices_mod, "get_provider", lambda _dt: provider)
|
|
manager = MagicMock()
|
|
manager.find_device_state.return_value = types.SimpleNamespace(hardware_brightness=None)
|
|
|
|
assert await resolve_device_brightness(_fake_device(software_brightness=42), manager) == 42
|