Add update-vendor.sh symlink support, unify install.sh vendor flow
- 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
This commit is contained in:
@@ -0,0 +1,342 @@
|
||||
"""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])
|
||||
Reference in New Issue
Block a user