Add state management, WebSocket polling, html.js templating, and refactor pages
- lib/state.py: per-subsystem collectors with versioned state store - daemon/server.py: state refresh on request, batch routing updates - webui/static/hoover/html.js: new html tag template helper via htm.js - webui/static/hoover/websocket.js: real-time state change notifications - webui/static/hoover/vdom.js: VDOM improvements for keyed diff - All frontend pages refactored to use html templates - Add tests for state management and polling - Update docs and AGENTS.md
This commit is contained in:
@@ -0,0 +1,64 @@
|
||||
"""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
|
||||
|
||||
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):
|
||||
from daemon.server import _ws_subscribers, broadcast_tick
|
||||
|
||||
mock_ws = AsyncMock()
|
||||
mock_ws.send_str = AsyncMock()
|
||||
_ws_subscribers.add(mock_ws)
|
||||
try:
|
||||
asyncio.run(broadcast_tick(["firewall", "wireguard"]))
|
||||
mock_ws.send_str.assert_called_once()
|
||||
sent = json.loads(mock_ws.send_str.call_args[0][0])
|
||||
assert sent["type"] == "tick"
|
||||
assert sent["subsystems"] == ["firewall", "wireguard"]
|
||||
finally:
|
||||
_ws_subscribers.discard(mock_ws)
|
||||
|
||||
def test_prunes_dead_subscribers(self):
|
||||
from daemon.server import _ws_subscribers, broadcast_tick
|
||||
|
||||
mock_ws = AsyncMock()
|
||||
mock_ws.send_str = AsyncMock(side_effect=Exception("broken"))
|
||||
_ws_subscribers.add(mock_ws)
|
||||
asyncio.run(broadcast_tick(["firewall"]))
|
||||
assert mock_ws not in _ws_subscribers
|
||||
|
||||
|
||||
class TestPollTasks:
|
||||
def test_poll_tasks_is_a_set(self):
|
||||
from daemon.server import _poll_tasks
|
||||
|
||||
assert isinstance(_poll_tasks, set)
|
||||
Reference in New Issue
Block a user