mteehan
8c13ad55ce
- update-vendor.sh now creates webui/vendor symlinks (htm.js) - install.sh calls update-vendor.sh after package install - Add vendor/.empty and webui/vendor/.empty as directory placeholders in git
343 lines
12 KiB
Python
343 lines
12 KiB
Python
"""Tests for daemon/handlers/status.py — aggregate pending + apply-all."""
|
|
|
|
from typing import Any, ClassVar
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from daemon.handlers import status
|
|
from lib.state import State
|
|
|
|
|
|
def _make_state(**kwargs):
|
|
"""Create a minimal in-memory state for pending checks."""
|
|
st = State()
|
|
for name, data in kwargs.items():
|
|
st.set(name, data)
|
|
return st
|
|
|
|
|
|
def _mock_state_store(state_dict):
|
|
"""Return a mock that looks like state_store.get()."""
|
|
mock = MagicMock()
|
|
mock.get.side_effect = lambda name: state_dict.get(name)
|
|
return mock
|
|
|
|
|
|
class TestFwChangeSummary:
|
|
"""Test fw_change_summary helper from status module."""
|
|
|
|
def test_interfaces_summary(self):
|
|
s = status.fw_change_summary(
|
|
"internal", "interfaces", {"config": ["eth1"], "live": []}
|
|
)
|
|
assert "Zone internal: interfaces changed" in s
|
|
assert "eth1" in s
|
|
|
|
def test_services_summary(self):
|
|
s = status.fw_change_summary(
|
|
"dmz", "services", {"config": ["ssh", "dns"], "live": ["ssh"]}
|
|
)
|
|
assert "Zone dmz: services changed" in s
|
|
|
|
def test_rich_rules_summary(self):
|
|
s = status.fw_change_summary(
|
|
"public", "rich_rules", {"config_count": 2, "live_count": 1}
|
|
)
|
|
assert "config: 2" in s
|
|
assert "live: 1" in s
|
|
|
|
def test_forward_ports_summary(self):
|
|
s = status.fw_change_summary(
|
|
"wan", "forward_ports", {"config_count": 3, "live_count": 0}
|
|
)
|
|
assert "Zone wan: port forwards differ" in s
|
|
|
|
def test_masquerade_summary(self):
|
|
s = status.fw_change_summary(
|
|
"lan", "masquerade", {"config": True, "live": False}
|
|
)
|
|
assert "Zone lan: masquerade changed" in s
|
|
|
|
def test_target_summary(self):
|
|
s = status.fw_change_summary(
|
|
"vpn", "target", {"config": "ACCEPT", "live": "default"}
|
|
)
|
|
assert "Zone vpn: target changed" in s
|
|
|
|
def test_unknown_type_summary(self):
|
|
s = status.fw_change_summary("public", "weird", {})
|
|
assert "Zone public: weird changed" in s
|
|
|
|
|
|
class TestHashSubsystem:
|
|
"""Test _hash_subsystem helper from status module."""
|
|
|
|
def test_no_state(self):
|
|
result = status._hash_subsystem("nginx", None)
|
|
assert result["pending_changes"] is False
|
|
assert result["summary"] == "Up to date"
|
|
|
|
def test_pending_true(self):
|
|
st = {"status": {"pending_changes": True}}
|
|
result = status._hash_subsystem("wireguard", st)
|
|
assert result["pending_changes"] is True
|
|
assert "unapplied changes" in result["summary"]
|
|
assert len(result["changes"]) == 1
|
|
|
|
def test_pending_false(self):
|
|
st = {"status": {"pending_changes": False}}
|
|
result = status._hash_subsystem("networkd", st)
|
|
assert result["pending_changes"] is False
|
|
|
|
def test_empty_status(self):
|
|
st = {}
|
|
result = status._hash_subsystem("dnsmasq", st)
|
|
assert result["pending_changes"] is False
|
|
|
|
|
|
class TestStatusPending:
|
|
"""Test the aggregate pending endpoint."""
|
|
|
|
@patch("daemon.handlers.status.state_store")
|
|
def test_all_synced(self, mock_store):
|
|
mock_store.get.return_value = {
|
|
"firewall": {"pending": {"needs_apply": False, "pending": []}},
|
|
"dnsmasq": {"status": {"pending_changes": False}},
|
|
"nginx": {"status": {"pending_changes": False}},
|
|
"wireguard": {"status": {"pending_changes": False}},
|
|
"networkd": {"status": {"pending_changes": False}},
|
|
}
|
|
result = status.status_pending(None, None)
|
|
assert result["total_changes"] == 0
|
|
assert not result["firewall"]["needs_apply"]
|
|
assert not result["dnsmasq"]["pending_changes"]
|
|
|
|
def _patch_store(self, data):
|
|
mock = MagicMock()
|
|
mock.get.side_effect = lambda name: data.get(name)
|
|
return patch("daemon.handlers.status.state_store", mock)
|
|
|
|
def test_firewall_pending_only(self):
|
|
with self._patch_store(
|
|
{
|
|
"firewall": {
|
|
"pending": {
|
|
"needs_apply": True,
|
|
"pending": [
|
|
{
|
|
"zone": "internal",
|
|
"type": "interfaces",
|
|
"config": ["eth1"],
|
|
"live": [],
|
|
}
|
|
],
|
|
}
|
|
},
|
|
"dnsmasq": {"status": {"pending_changes": False}},
|
|
"nginx": {"status": {"pending_changes": False}},
|
|
"wireguard": {"status": {"pending_changes": False}},
|
|
"networkd": {"status": {"pending_changes": False}},
|
|
}
|
|
):
|
|
result = status.status_pending(None, None)
|
|
assert result["total_changes"] == 1
|
|
assert result["firewall"]["change_count"] == 1
|
|
|
|
def test_multiple_subsystems_pending(self):
|
|
with self._patch_store(
|
|
{
|
|
"firewall": {
|
|
"pending": {
|
|
"needs_apply": True,
|
|
"pending": [
|
|
{
|
|
"zone": "lan",
|
|
"type": "services",
|
|
"config": ["ssh"],
|
|
"live": [],
|
|
},
|
|
{
|
|
"zone": "wan",
|
|
"type": "masquerade",
|
|
"config": True,
|
|
"live": False,
|
|
},
|
|
],
|
|
}
|
|
},
|
|
"dnsmasq": {"status": {"pending_changes": True}},
|
|
"nginx": {"status": {"pending_changes": False}},
|
|
"wireguard": {"status": {"pending_changes": True}},
|
|
"networkd": {"status": {"pending_changes": False}},
|
|
}
|
|
):
|
|
result = status.status_pending(None, None)
|
|
assert result["total_changes"] == 4 # 2 FW + 1 DHCP + 1 WG
|
|
assert result["firewall"]["change_count"] == 2
|
|
assert result["firewall"]["needs_apply"] is True
|
|
assert result["dnsmasq"]["pending_changes"] is True
|
|
assert result["wireguard"]["pending_changes"] is True
|
|
|
|
def test_empty_state(self):
|
|
with self._patch_store({}):
|
|
result = status.status_pending(None, None)
|
|
assert result["total_changes"] == 0
|
|
assert not result["firewall"]["needs_apply"]
|
|
|
|
def test_firewall_no_pending_key(self):
|
|
with self._patch_store(
|
|
{
|
|
"firewall": {},
|
|
"dnsmasq": {"status": {"pending_changes": False}},
|
|
"nginx": None,
|
|
"wireguard": None,
|
|
"networkd": None,
|
|
}
|
|
):
|
|
result = status.status_pending(None, None)
|
|
assert result["total_changes"] == 0
|
|
assert not result["firewall"]["needs_apply"]
|
|
|
|
|
|
class TestStatusApplyAll:
|
|
"""Test the apply-all endpoint.
|
|
|
|
Patches SYS_APPLY dict entries directly since they hold function
|
|
references at import time.
|
|
"""
|
|
|
|
_fake_pending_all: ClassVar[dict[str, Any]] = {
|
|
"firewall": {"needs_apply": False, "change_count": 0, "changes": []},
|
|
"dnsmasq": {"pending_changes": False, "summary": "Up to date", "changes": []},
|
|
"nginx": {"pending_changes": False, "summary": "Up to date", "changes": []},
|
|
"wireguard": {"pending_changes": False, "summary": "Up to date", "changes": []},
|
|
"networkd": {"pending_changes": False, "summary": "Up to date", "changes": []},
|
|
}
|
|
|
|
@patch("daemon.handlers.status.status_pending")
|
|
@patch("daemon.handlers.status.refresh_state")
|
|
def test_nothing_to_apply(self, mock_refresh, mock_pending):
|
|
mock_pending.return_value = self._fake_pending_all
|
|
result = status.status_apply_all(None, None)
|
|
assert result["applied"] == []
|
|
assert result["errors"] == {}
|
|
mock_refresh.assert_called_once()
|
|
|
|
def test_applies_pending_subsystems(self):
|
|
mock_net = MagicMock()
|
|
mock_fw = MagicMock()
|
|
|
|
pending_data = {**self._fake_pending_all}
|
|
pending_data["firewall"]["needs_apply"] = True
|
|
pending_data["firewall"]["change_count"] = 1
|
|
pending_data["networkd"]["pending_changes"] = True
|
|
|
|
with (
|
|
patch("daemon.handlers.status.status_pending", return_value=pending_data),
|
|
patch("daemon.handlers.status.refresh_state"),
|
|
patch.dict(
|
|
"daemon.handlers.status.SYS_APPLY",
|
|
{
|
|
"networkd": mock_net,
|
|
"firewall": mock_fw,
|
|
},
|
|
),
|
|
):
|
|
result = status.status_apply_all(None, None)
|
|
assert "networkd" in result["applied"]
|
|
assert "firewall" in result["applied"]
|
|
mock_net.assert_called_once()
|
|
mock_fw.assert_called_once()
|
|
|
|
def test_error_in_subsystem(self):
|
|
mock_fw = MagicMock(side_effect=RuntimeError("firewalld not running"))
|
|
|
|
pending_data = {**self._fake_pending_all}
|
|
pending_data["firewall"]["needs_apply"] = True
|
|
pending_data["firewall"]["change_count"] = 1
|
|
|
|
with (
|
|
patch("daemon.handlers.status.status_pending", return_value=pending_data),
|
|
patch("daemon.handlers.status.refresh_state"),
|
|
patch.dict("daemon.handlers.status.SYS_APPLY", {"firewall": mock_fw}),
|
|
):
|
|
result = status.status_apply_all(None, None)
|
|
assert "firewall" not in result["applied"]
|
|
assert "Firewall" in result["errors"]
|
|
assert "firewalld not running" in result["errors"]["Firewall"]
|
|
|
|
def test_order_is_respected(self):
|
|
call_order = []
|
|
|
|
def track(name):
|
|
def wrapper(*args):
|
|
call_order.append(name)
|
|
|
|
return wrapper
|
|
|
|
mock_net = MagicMock(side_effect=track("networkd"))
|
|
mock_wg = MagicMock(side_effect=track("wireguard"))
|
|
|
|
pending_data = {**self._fake_pending_all}
|
|
pending_data["wireguard"]["pending_changes"] = True
|
|
pending_data["networkd"]["pending_changes"] = True
|
|
|
|
with (
|
|
patch("daemon.handlers.status.status_pending", return_value=pending_data),
|
|
patch("daemon.handlers.status.refresh_state"),
|
|
patch.dict(
|
|
"daemon.handlers.status.SYS_APPLY",
|
|
{
|
|
"networkd": mock_net,
|
|
"wireguard": mock_wg,
|
|
},
|
|
),
|
|
):
|
|
status.status_apply_all(None, None)
|
|
assert call_order == ["networkd", "wireguard"]
|
|
|
|
def test_partial_failure_still_applies_others(self):
|
|
mock_fw = MagicMock(side_effect=RuntimeError("fail"))
|
|
mock_nginx = MagicMock()
|
|
|
|
pending_data = {**self._fake_pending_all}
|
|
pending_data["firewall"]["needs_apply"] = True
|
|
pending_data["firewall"]["change_count"] = 1
|
|
pending_data["nginx"]["pending_changes"] = True
|
|
|
|
with (
|
|
patch("daemon.handlers.status.status_pending", return_value=pending_data),
|
|
patch("daemon.handlers.status.refresh_state"),
|
|
patch.dict(
|
|
"daemon.handlers.status.SYS_APPLY",
|
|
{
|
|
"firewall": mock_fw,
|
|
"nginx": mock_nginx,
|
|
},
|
|
),
|
|
):
|
|
result = status.status_apply_all(None, None)
|
|
assert "firewall" not in result["applied"]
|
|
assert "nginx" in result["applied"]
|
|
assert "Firewall" in result["errors"]
|
|
mock_nginx.assert_called_once()
|
|
|
|
|
|
class TestSysOrder:
|
|
"""Verify SYS_ORDER and SYS_LABELS constants."""
|
|
|
|
def test_order_network_first(self):
|
|
assert status.SYS_ORDER[0] == "networkd"
|
|
|
|
def test_all_subsystems_present(self):
|
|
expected = {"networkd", "firewall", "wireguard", "dnsmasq", "nginx"}
|
|
assert set(status.SYS_ORDER) == expected
|
|
|
|
def test_labels_match(self):
|
|
for name in status.SYS_ORDER:
|
|
assert name in status.SYS_LABELS
|
|
assert name in status.SYS_APPLY
|
|
|
|
def test_apply_functions_callable(self):
|
|
for name in status.SYS_ORDER:
|
|
assert callable(status.SYS_APPLY[name])
|