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
73 lines
2.5 KiB
Python
73 lines
2.5 KiB
Python
"""Tests for daemon/server.py polling functions."""
|
|
|
|
import asyncio
|
|
import json
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
|
|
class TestPollIntervals:
|
|
def test_default_intervals_loaded(self):
|
|
from daemon.server import _POLL_INTERVALS
|
|
|
|
assert "firewall" in _POLL_INTERVALS
|
|
assert _POLL_INTERVALS["firewall"] == 30
|
|
assert _POLL_INTERVALS["wireguard"] == 10
|
|
assert _POLL_INTERVALS["dnsmasq"] == 10
|
|
assert _POLL_INTERVALS["networkd"] == 10
|
|
# Phase 5: real-time system metrics poll at 1s.
|
|
assert _POLL_INTERVALS["system"] == 1
|
|
# nginx/acme derive from config files; poll for drift self-heal.
|
|
assert _POLL_INTERVALS["nginx"] == 60
|
|
assert _POLL_INTERVALS["acme"] == 300
|
|
|
|
def test_env_override(self):
|
|
"""VACUUM_WALL_POLL_INTERVALS env var can override values."""
|
|
import importlib
|
|
|
|
with patch.dict(
|
|
"os.environ", {"VACUUM_WALL_POLL_INTERVALS": "firewall:60,wireguard:5"}
|
|
):
|
|
import daemon.server
|
|
|
|
importlib.reload(daemon.server)
|
|
assert daemon.server._POLL_INTERVALS["firewall"] == 60
|
|
assert daemon.server._POLL_INTERVALS["wireguard"] == 5
|
|
assert daemon.server._POLL_INTERVALS["dnsmasq"] == 10
|
|
importlib.reload(daemon.server)
|
|
|
|
|
|
class TestBroadcastTick:
|
|
def test_sends_tick_message(self):
|
|
import daemon.server as server
|
|
|
|
mock_ws = AsyncMock()
|
|
mock_ws.send_str = AsyncMock()
|
|
server._ws_subscribers.add(mock_ws)
|
|
try:
|
|
with patch.object(server.state_store, "get", return_value={"up": True}):
|
|
asyncio.run(server.broadcast_tick("firewall"))
|
|
mock_ws.send_str.assert_called_once()
|
|
sent = json.loads(mock_ws.send_str.call_args[0][0])
|
|
assert sent["type"] == "tick"
|
|
assert sent["subsystem"] == "firewall"
|
|
assert sent["data"] == {"up": True}
|
|
finally:
|
|
server._ws_subscribers.discard(mock_ws)
|
|
|
|
def test_prunes_dead_subscribers(self):
|
|
import daemon.server as server
|
|
|
|
mock_ws = AsyncMock()
|
|
mock_ws.send_str = AsyncMock(side_effect=Exception("broken"))
|
|
server._ws_subscribers.add(mock_ws)
|
|
with patch.object(server.state_store, "get", return_value={"up": True}):
|
|
asyncio.run(server.broadcast_tick("firewall"))
|
|
assert mock_ws not in server._ws_subscribers
|
|
|
|
|
|
class TestPollTasks:
|
|
def test_poll_tasks_is_a_set(self):
|
|
from daemon.server import _poll_tasks
|
|
|
|
assert isinstance(_poll_tasks, set)
|