mteehan
332d14e37d
- daemon: send full snapshot on connect; versions/tick now carry the full state of one subsystem (subsystem + data); no legacy updated/subsystems payloads; refresh_state and POST /status/refresh broadcast per-subsystem versions with data - client: modelSet() patches models in place; onMessage/topic refresh retired; 3s initial-load fallback via new POST /api/status/refresh - schema: lib/schema.py TypedDicts + hoover/schema.js defaults + docs/state-model.md as single source of truth for state shapes - system: poll at 1s, volatile metrics registered, dashboard uses a dedicated system model (status model removed) - firewall: refuse to strip both https and ssh from the default zone (409, force override via UI confirm); set_zone_services persists services to the declarative config; collector exposes default_zone - UI: pages migrate to flat state shapes; post-mutation modelFetch refreshes removed (WS delta covers it) - tests: ws snapshot/delta/broadcast, refresh-state, schema types, model-set/js ws handler and reconnect fallback
115 lines
4.1 KiB
Python
115 lines
4.1 KiB
Python
"""Tests for WS delta structure (daemon.server._poll_loop + broadcasts).
|
|
|
|
After the push-stream migration the poll loop drives per-subsystem deltas:
|
|
a structural diff bumps the version and broadcasts {type: versions,
|
|
subsystem, data}; a volatile-only diff broadcasts {type: tick, subsystem,
|
|
data}. No legacy `updated` dict / `subsystems` array is emitted.
|
|
"""
|
|
|
|
import asyncio
|
|
import hashlib
|
|
import json
|
|
from contextlib import suppress
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import daemon.server as server
|
|
from lib.state import State
|
|
|
|
|
|
def _zero_offset_subsystem(interval: int) -> str:
|
|
"""Find a subsystem name whose md5 offset is 0 so the loop starts at once."""
|
|
for i in range(100_000):
|
|
name = f"sub{i}"
|
|
offset = int(hashlib.md5(name.encode()).hexdigest(), 16) % interval
|
|
if offset == 0:
|
|
return name
|
|
raise AssertionError("could not find zero-offset name")
|
|
|
|
|
|
def _run_one_poll_iteration(poll_result):
|
|
"""Run _poll_loop for a single iteration and return the broadcast mocks."""
|
|
namespaced = _zero_offset_subsystem(60)
|
|
|
|
async def drive():
|
|
store = MagicMock()
|
|
store.poll.return_value = poll_result
|
|
store.bump = MagicMock()
|
|
store.get.return_value = {"value": 1}
|
|
bv = AsyncMock()
|
|
bt = AsyncMock()
|
|
task = None
|
|
with (
|
|
patch.object(server, "state_store", store),
|
|
patch.object(server, "broadcast_versions", bv),
|
|
patch.object(server, "broadcast_tick", bt),
|
|
patch.object(server, "blacklist_expired"),
|
|
):
|
|
task = asyncio.create_task(server._poll_loop(namespaced, 60))
|
|
await asyncio.sleep(0.02) # let one full iteration run
|
|
task.cancel()
|
|
with suppress(asyncio.CancelledError):
|
|
await task
|
|
return store, bv, bt
|
|
|
|
store, bv, bt = asyncio.run(drive())
|
|
return store, bv, bt
|
|
|
|
|
|
class TestPollLoopDeltas:
|
|
def test_structural_change_broadcasts_versions(self):
|
|
store, bv, bt = _run_one_poll_iteration((True, False))
|
|
store.bump.assert_called_once_with(_zero_offset_subsystem(60))
|
|
bv.assert_awaited_once()
|
|
bt.assert_not_awaited()
|
|
|
|
def test_volatile_change_broadcasts_tick(self):
|
|
store, bv, bt = _run_one_poll_iteration((False, True))
|
|
store.bump.assert_not_called()
|
|
bv.assert_not_awaited()
|
|
bt.assert_awaited_once()
|
|
|
|
def test_no_change_no_broadcast(self):
|
|
store, bv, bt = _run_one_poll_iteration((False, False))
|
|
store.bump.assert_not_called()
|
|
bv.assert_not_awaited()
|
|
bt.assert_not_awaited()
|
|
|
|
|
|
class TestDeltaMessageShape:
|
|
def test_versions_message_carries_subsystem_and_data(self):
|
|
"""broadcast_versions emits {type, subsystem, data} — no `updated`."""
|
|
store = State()
|
|
store.set("firewall", {"zones": {"public": {}}})
|
|
ws = AsyncMock()
|
|
ws.send_str = AsyncMock()
|
|
server._ws_subscribers.add(ws)
|
|
try:
|
|
with patch.object(server, "state_store", store):
|
|
asyncio.run(server.broadcast_versions("firewall"))
|
|
ws.send_str.assert_awaited_once()
|
|
msg = json.loads(ws.send_str.call_args[0][0])
|
|
assert msg["type"] == "versions"
|
|
assert msg["subsystem"] == "firewall"
|
|
assert msg["data"] == {"zones": {"public": {}}}
|
|
assert "updated" not in msg
|
|
finally:
|
|
server._ws_subscribers.discard(ws)
|
|
|
|
def test_tick_message_carries_subsystem_and_data(self):
|
|
store = State()
|
|
store.set("system", {"load": {"load1": 1.0}})
|
|
ws = AsyncMock()
|
|
ws.send_str = AsyncMock()
|
|
server._ws_subscribers.add(ws)
|
|
try:
|
|
with patch.object(server, "state_store", store):
|
|
asyncio.run(server.broadcast_tick("system"))
|
|
ws.send_str.assert_awaited_once()
|
|
msg = json.loads(ws.send_str.call_args[0][0])
|
|
assert msg["type"] == "tick"
|
|
assert msg["subsystem"] == "system"
|
|
assert msg["data"] == {"load": {"load1": 1.0}}
|
|
assert "subsystems" not in msg
|
|
finally:
|
|
server._ws_subscribers.discard(ws)
|