diff --git a/tests/test-reactive-dom.js b/tests/test-reactive-dom.js
new file mode 100644
index 0000000..3b73e1f
--- /dev/null
+++ b/tests/test-reactive-dom.js
@@ -0,0 +1,120 @@
+import { reactive, h, html, render, Router, Link } from '../webui/static/reactive-dom.js';
+
+let passed = 0;
+let failed = 0;
+
+function test(name, fn) {
+ try {
+ fn();
+ console.log(` ✓ ${name}`);
+ passed++;
+ } catch (e) {
+ console.error(` ✗ ${name}: ${e.message}`);
+ failed++;
+ }
+}
+
+function assert(cond, msg) {
+ if (!cond) throw new Error(msg || 'Assertion failed');
+}
+
+console.log('Testing reactive-dom.js\n');
+
+// === Reactive ===
+test('reactive() returns proxy', () => {
+ const s = reactive({ x: 1 });
+ assert(s.x === 1);
+});
+
+test('reactive() mutation triggers render callback', () => {
+ const s = reactive({ x: 1 });
+ let fired = false;
+ // We can't easily test the render callback without a DOM, but we can check the proxy works
+ s.x = 2;
+ assert(s.x === 2);
+});
+
+// === h() ===
+test('h() creates element VNode', () => {
+ const v = h('div', { class: 'foo' });
+ assert(v.tag === 'div' && v.props.class === 'foo');
+});
+
+test('h() flattens children array', () => {
+ const v = h('div', null, h('span', null, 'hi'));
+ assert(v.ch.length === 1 && v.ch[0].tag === 'span');
+});
+
+test('h() converts strings to text nodes', () => {
+ const v = h('div', null, 'hello', 42);
+ assert(v.ch.length === 2 && v.ch[0].tag === '#text' && v.ch[0].text === 'hello');
+ assert(v.ch[1].text === '42');
+});
+
+test('h() drops null/boolean children', () => {
+ const v = h('div', null, null, undefined, true, false, 'x');
+ assert(v.ch.length === 1 && v.ch[0].text === 'x');
+});
+
+// === html() ===
+test('html() parses static element', () => {
+ const nodes = html`
hello
`;
+ assert(nodes[0].tag === 'div' && nodes[0].ch[0].text === 'hello');
+});
+
+test('html() interpolates text into element children', () => {
+ const name = 'World';
+ const nodes = html`Hello ${name}
`;
+ assert(nodes[0].tag === 'div');
+ // Should have: text "Hello ", then text "World"
+ assert(nodes[0].ch[0].text && nodes[0].ch[0].text === 'Hello ');
+ assert(nodes[0].ch[1].tag === '#text' && nodes[0].ch[1].text === 'World');
+});
+
+test('html() interpolates class attribute value', () => {
+ const cls = 'active';
+ const nodes = html`x
`;
+ assert(nodes[0].tag === 'div' && nodes[0].props.class === 'active');
+});
+
+test('html() interpolates on:click attribute value', () => {
+ const handler = function click() {};
+ const nodes = html``;
+ assert(nodes[0].tag === 'button' && typeof nodes[0].props['on:click'] === 'function');
+});
+
+test('html() handles multiple interpolations', () => {
+ const a = 'first', b = 'second';
+ const nodes = html`${a} ${b}
`;
+ assert(nodes[0].tag === 'div');
+ assert(nodes[0].ch[0].tag === 'span' && nodes[0].ch[0].ch[0].text === 'first');
+});
+
+test('html() interpolates VNode into element children', () => {
+ const nodes = html``;
+ assert(nodes[0].tag === 'ul' && nodes[0].ch[0].tag === 'li');
+});
+
+// === Router ===
+test('Router initializes with current hash', () => {
+ globalThis.location = { hash: '' };
+ globalThis.window = { addEventListener: () => {} };
+ const router = Router({ '/home': () => {} });
+ assert(router.state.path === '/');
+});
+
+// === Link ===
+test('Link creates anchor with hash href', () => {
+ const link = Link({ path: '/dashboard' });
+ assert(link.tag === 'a' && link.props.href === '#/dashboard');
+});
+
+// === DOM functions (basic, no actual DOM) ===
+test('createDom produces document.createElement call', () => {
+ const v = h('div', { class: 'foo' }, 'hi', h('span', null, 'nested'));
+ // Can't test actual DOM without jsdom, but we can verify the VNode structure
+ assert(v.tag === 'div' && v.ch.length === 2);
+});
+
+console.log(`\n${passed} passed, ${failed} failed`);
+process.exit(failed > 0 ? 1 : 0);
diff --git a/tests/test_client.py b/tests/test_client.py
new file mode 100644
index 0000000..7f72094
--- /dev/null
+++ b/tests/test_client.py
@@ -0,0 +1,74 @@
+"""Tests for daemon client path parameter substitution."""
+
+import contextlib
+from unittest.mock import patch
+
+from daemon.client import _format_path, request
+
+
+class TestFormatPath:
+ def test_simple_substitution(self):
+ assert (
+ _format_path("/network/interfaces/", {"name": "eth0"})
+ == "/network/interfaces/eth0"
+ )
+
+ def test_multiple_params(self):
+ assert _format_path("/a//b/", {"x": "1", "y": "2"}) == "/a/1/b/2"
+
+ def test_no_params_unchanged(self):
+ assert (
+ _format_path("/network/interfaces/", None)
+ == "/network/interfaces/"
+ )
+
+ def test_empty_params_unchanged(self):
+ assert (
+ _format_path("/network/interfaces/", {})
+ == "/network/interfaces/"
+ )
+
+ def test_partial_substitution(self):
+ assert _format_path("/a//b/", {"x": "1"}) == "/a/1/b/"
+
+ def test_url_encodes_special_chars(self):
+ assert _format_path("/a/", {"x": "foo bar"}) == "/a/foo%20bar"
+
+ def test_preserves_non_param_brackets(self):
+ assert _format_path("/foo[bar]/", {"x": "z"}) == "/foo[bar]/z"
+
+ def test_numeric_value(self):
+ assert _format_path("/items/", {"id": 42}) == "/items/42"
+
+ def test_path_without_params(self):
+ assert _format_path("/health", {"foo": "bar"}) == "/health"
+
+
+class TestRequestPathSubstitution:
+ @patch("daemon.client.requests_unixsocket.Session")
+ def test_post_substitutes_name_from_body(self, mock_session_cls):
+ mock_sess = mock_session_cls.return_value
+ mock_resp = mock_sess.request.return_value
+ mock_resp.status_code = 200
+ mock_resp.json.return_value = {"ok": True, "data": {"name": "eth0"}}
+
+ with contextlib.suppress(Exception):
+ request("POST", "/network/interfaces/", json_body={"name": "eth0"})
+
+ call_args = mock_sess.request.call_args
+ url = call_args[0][1] if call_args else ""
+ assert "/interfaces/eth0" in url
+
+ @patch("daemon.client.requests_unixsocket.Session")
+ def test_get_substitutes_name_from_query(self, mock_session_cls):
+ mock_sess = mock_session_cls.return_value
+ mock_resp = mock_sess.request.return_value
+ mock_resp.status_code = 200
+ mock_resp.json.return_value = {"ok": True, "data": {}}
+
+ with contextlib.suppress(Exception):
+ request("GET", "/network/interfaces/", query_params={"name": "eth0"})
+
+ call_args = mock_sess.request.call_args
+ url = call_args[0][1] if call_args else ""
+ assert "/interfaces/eth0" in url
diff --git a/tests/test_firewall.py b/tests/test_firewall.py
index bf0000e..d44d7a3 100644
--- a/tests/test_firewall.py
+++ b/tests/test_firewall.py
@@ -435,7 +435,7 @@ class TestDaemonGetState:
class TestDaemonConfigApply:
@patch(
- "daemon.handlers.firewall._get_config",
+ "lib.firewall.get_config",
return_value={
"zones": {
"public": {
diff --git a/tests/test_handler_acme.py b/tests/test_handler_acme.py
new file mode 100644
index 0000000..cd5ce08
--- /dev/null
+++ b/tests/test_handler_acme.py
@@ -0,0 +1,84 @@
+"""Tests for daemon/handlers/acme.py — handler endpoint logic."""
+
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+from daemon.handlers.acme import generate_self_signed
+
+
+class TestGenerateSelfSigned:
+ def test_generate_creates_files(self, tmp_path):
+ with (
+ patch("daemon.handlers.acme._ACME_HOME", tmp_path / "acme"),
+ ):
+ result = generate_self_signed(None, {"domain": "test.local"})
+
+ assert result["domain"] == "test.local"
+ assert result["generated"] is True
+ cert_dir = tmp_path / "acme" / "test.local"
+ assert result["cert"] == str(cert_dir / "fullchain.cer")
+ assert result["key"] == str(cert_dir / "test.local.key")
+ assert (cert_dir / "fullchain.cer").is_file()
+ assert (cert_dir / "test.local.key").is_file()
+
+ def test_generate_idempotent_skips_existing(self, tmp_path):
+ cert_dir = tmp_path / "acme" / "test.local"
+ cert_dir.mkdir(parents=True)
+ (cert_dir / "fullchain.cer").write_text("dummy-cert")
+ (cert_dir / "test.local.key").write_text("dummy-key")
+
+ with patch("daemon.handlers.acme._ACME_HOME", tmp_path / "acme"):
+ result = generate_self_signed(None, {"domain": "test.local"})
+
+ assert result["generated"] is False
+
+ def test_generate_partial_existing(self, tmp_path):
+ cert_dir = tmp_path / "acme" / "test.local"
+ cert_dir.mkdir(parents=True)
+ (cert_dir / "fullchain.cer").write_text("dummy-cert")
+ # key missing -> should regenerate
+
+ with patch("daemon.handlers.acme._ACME_HOME", tmp_path / "acme"):
+ result = generate_self_signed(None, {"domain": "test.local"})
+
+ assert result["generated"] is True
+
+ def test_generate_custom_days(self, tmp_path):
+ with (
+ patch("daemon.handlers.acme._ACME_HOME", tmp_path / "acme"),
+ patch("subprocess.run") as mock_run,
+ ):
+
+ def _create_files(*args, **kwargs):
+ cert_dir = tmp_path / "acme" / "test.local"
+ cert_dir.mkdir(parents=True, exist_ok=True)
+ (cert_dir / "fullchain.cer").touch()
+ (cert_dir / "test.local.key").touch()
+ return Path("")
+
+ mock_run.side_effect = _create_files
+ generate_self_signed(None, {"domain": "test.local", "days": 730})
+ args = mock_run.call_args[0][0]
+ assert "-days" in args
+ idx = args.index("-days")
+ assert args[idx + 1] == "730"
+
+ cert_dir = tmp_path / "acme" / "test.local"
+ if (cert_dir / "fullchain.cer").is_file():
+ assert cert_dir.is_dir()
+
+ def test_generate_creates_directory(self, tmp_path):
+ with patch("daemon.handlers.acme._ACME_HOME", tmp_path / "acme"):
+ generate_self_signed(None, {"domain": "test.local"})
+
+ assert (tmp_path / "acme" / "test.local").is_dir()
+
+ def test_generate_requires_domain(self):
+ with pytest.raises(ValueError, match="domain"):
+ generate_self_signed(None, {"foo": "bar"})
+
+ def test_generate_requires_body(self):
+ with pytest.raises(ValueError, match="body"):
+ generate_self_signed(None, None)
diff --git a/tests/test_handler_network.py b/tests/test_handler_network.py
index 3e04ebf..07e9b11 100644
--- a/tests/test_handler_network.py
+++ b/tests/test_handler_network.py
@@ -13,6 +13,7 @@ from daemon.handlers.network import (
get_interfaces,
reload_interface,
save_interface,
+ set_sysctl,
)
from lib import network as _net
@@ -65,8 +66,8 @@ class TestSaveInterface:
)
data_dir = tmp_network / "data" / "networkd"
- assert (data_dir / "50-eth0.network").exists()
- content = (data_dir / "50-eth0.network").read_text()
+ assert (data_dir / "99-eth0.network").exists()
+ content = (data_dir / "99-eth0.network").read_text()
assert "Name=eth0" in content
assert "Address=10.0.0.1/24" in content
@@ -78,6 +79,54 @@ class TestSaveInterface:
with pytest.raises(ValueError, match="body"):
save_interface(None, None)
+ def test_save_interface_rejects_invalid_name(self, tmp_network):
+ invalid_names = [
+ "../../etc/passwd",
+ "eth 0",
+ "",
+ "eth/0",
+ "eth..0",
+ ]
+ for invalid in invalid_names:
+ with (
+ patch("daemon.handlers.network.run") as mock_run,
+ patch(
+ "daemon.handlers.network.DATA_DIR",
+ tmp_network / "data" / "networkd",
+ ),
+ ):
+ mock_run.return_value = (
+ "1: eth0 ethernet routable\n State: routable\n"
+ )
+ with pytest.raises(ValueError, match="name"):
+ save_interface(None, {"name": invalid})
+
+
+class TestReloadInterfaceValidation:
+ def test_reload_interface_rejects_invalid_name(self):
+ invalid_names = [
+ "../../etc/passwd",
+ "eth 0",
+ "",
+ "eth/0",
+ ]
+ for invalid in invalid_names:
+ with pytest.raises(ValueError, match="name"):
+ reload_interface(None, {"name": invalid})
+
+
+class TestGetInterfaceValidation:
+ def test_get_interface_rejects_invalid_name(self, tmp_network):
+ invalid_names = [
+ "../../etc/passwd",
+ "eth 0",
+ "",
+ "eth/0",
+ ]
+ for invalid in invalid_names:
+ with pytest.raises(ValueError, match="name"):
+ get_interface(None, {"name": invalid})
+
class TestReloadInterface:
def test_reload_interface(self):
@@ -117,7 +166,7 @@ class TestApplyAll:
patch("daemon.handlers.network.collect_upstream_dns", return_value=[]),
):
mock_gen.return_value = {
- "generated": [_net.DATA_DIR / "50-eth0.network"],
+ "generated": [_net.DATA_DIR / "99-eth0.network"],
"cleaned": [],
}
mock_run.return_value = ""
@@ -146,7 +195,7 @@ class TestApplyAll:
patch("daemon.handlers.network.collect_upstream_dns") as mock_collect,
):
mock_gen.return_value = {
- "generated": [_net.DATA_DIR / "50-eth0.network"],
+ "generated": [_net.DATA_DIR / "99-eth0.network"],
"cleaned": [],
}
mock_run.return_value = ""
@@ -169,7 +218,7 @@ class TestApplyAll:
patch("daemon.handlers.network.collect_upstream_dns") as mock_collect,
):
mock_gen.return_value = {
- "generated": [_net.DATA_DIR / "50-eth0.network"],
+ "generated": [_net.DATA_DIR / "99-eth0.network"],
"cleaned": [],
}
mock_run.return_value = ""
@@ -190,7 +239,7 @@ class TestApplyAll:
patch("daemon.handlers.network.collect_upstream_dns", return_value=[]),
):
mock_gen.return_value = {
- "generated": [_net.DATA_DIR / "50-eth0.network"],
+ "generated": [_net.DATA_DIR / "99-eth0.network"],
"cleaned": [],
}
mock_run.return_value = ""
@@ -295,3 +344,50 @@ class TestInferEndpoints:
assert "zones" in result
assert result["zones"]["wg0"] == "wan"
assert result["zones"]["eth0"] == "lan"
+
+
+class TestSetSysctl:
+ def test_set_sysctl_success(self):
+ with (
+ patch("daemon.handlers.network.run") as mock_run,
+ patch.object(Path, "read_text", return_value="1"),
+ ):
+ mock_run.return_value = "" # sysctl -w call
+ result = set_sysctl(None, {"name": "net.ipv4.ip_forward", "value": "1"})
+
+ assert result["name"] == "net.ipv4.ip_forward"
+ assert result["value"] == "1"
+ assert mock_run.call_count == 1
+ assert mock_run.call_args_list[0].args == (
+ ["sysctl", "-w", "net.ipv4.ip_forward=1"],
+ )
+ assert mock_run.call_args_list[0].kwargs == {"sudo": True}
+
+ def test_set_sysctl_rejects_slash_in_name(self):
+ with pytest.raises(ValueError, match="valid sysctl key"):
+ set_sysctl(None, {"name": "net.ipv4/ip_forward", "value": "1"})
+
+ def test_set_sysctl_rejects_double_dot(self):
+ with pytest.raises(ValueError, match="valid sysctl key"):
+ set_sysctl(None, {"name": "net..ipv4", "value": "1"})
+
+ def test_set_sysctl_requires_name(self):
+ with pytest.raises(ValueError, match="name"):
+ set_sysctl(None, {"value": "1"})
+
+ def test_set_sysctl_requires_value(self):
+ with pytest.raises(ValueError, match="value"):
+ set_sysctl(None, {"name": "net.ipv4.ip_forward"})
+
+ def test_set_sysctl_requires_body(self):
+ with pytest.raises(ValueError, match="body"):
+ set_sysctl(None, None)
+
+ def test_set_sysctl_verify_failure(self):
+ with (
+ patch("daemon.handlers.network.run") as mock_run,
+ patch.object(Path, "read_text", return_value="0"),
+ ):
+ mock_run.return_value = "" # sysctl -w call succeeds
+ with pytest.raises(RuntimeError, match="verify failed"):
+ set_sysctl(None, {"name": "net.ipv4.ip_forward", "value": "1"})
diff --git a/tests/test_iface_sync.py b/tests/test_iface_sync.py
new file mode 100644
index 0000000..83c243f
--- /dev/null
+++ b/tests/test_iface_sync.py
@@ -0,0 +1,145 @@
+"""Tests that daemon.iface stays in sync with registered server routes.
+
+Verifies a two-way contract:
+1. Every iface constant has a matching handler registered.
+2. Every registered handler has a matching iface constant.
+
+Run with: pytest tests/test_iface_sync.py -v
+"""
+
+from collections import defaultdict
+
+import pytest
+
+
+@pytest.fixture(autouse=True)
+def _load_handlers():
+ """Load all handler modules so registry is populated."""
+ # Import server to get registry, then load handlers
+ from daemon import server
+
+ # Force route registration
+ server._register_routes()
+
+
+@pytest.fixture()
+def registry():
+ from daemon import server
+
+ return server.registry
+
+
+@pytest.fixture()
+def iface_module():
+ import daemon.iface as iface
+
+ return iface
+
+
+def _get_iface_pairs(iface_module):
+ """Extract all (method, path) pairs from iface module."""
+ iface = iface_module
+ return {
+ name: val
+ for name, val in iface.__dict__.items()
+ if isinstance(val, tuple) and len(val) == 2 and isinstance(val[0], str)
+ }
+
+
+def _get_registered_routes(registry):
+ """Extract all (METHOD, path) keys from the registry."""
+ return {(method.upper(), path) for (method, path) in registry._routes}
+
+
+class TestIfaceSync:
+ """Verify iface constants match registered routes."""
+
+ def test_iface_constants_non_empty(self, iface_module):
+ pairs = _get_iface_pairs(iface_module)
+ assert len(pairs) >= 50, f"Expected many iface constants, got {len(pairs)}"
+
+ def test_iface_constants_have_registered_handlers(self, registry, iface_module):
+ """Every iface constant should map to a registered route."""
+ registered = _get_registered_routes(registry)
+ iface_pairs = _get_iface_pairs(iface_module)
+
+ # These 5 routes go through add_route() in create_app(), not @registry.register
+ add_route_paths = {
+ "/health",
+ "/status/all",
+ "/status/refresh",
+ "/ws",
+ "/batch",
+ }
+
+ missing = []
+ for name, (method, path) in iface_pairs.items():
+ key = (method.upper(), path)
+ if path not in add_route_paths and key not in registered:
+ missing.append(
+ (
+ name,
+ {
+ "method": method,
+ "path": path,
+ },
+ )
+ )
+
+ if missing:
+ detail = "\n".join(f" {name}: {pair}" for name, pair in missing)
+ pytest.fail(
+ f"{len(missing)} iface constant(s) have no matching handler:\n{detail}"
+ )
+
+ def test_registered_routes_have_iface_constants(self, registry, iface_module):
+ """Every registered route should have a matching iface constant."""
+ registered = _get_registered_routes(registry)
+ iface_pairs = _get_iface_pairs(iface_module)
+
+ iface_keys = set(iface_pairs.values())
+ missing = registered - iface_keys
+
+ if missing:
+ detail = "\n".join(f" {method} {path}" for method, path in sorted(missing))
+ pytest.fail(
+ f"{len(missing)} registered route(s) have no matching iface constant:\n{detail}"
+ )
+
+ def test_no_duplicate_iface_constants(self, iface_module):
+ """All iface constants should have unique (method, path) pairs."""
+ iface_pairs = _get_iface_pairs(iface_module)
+ seen = defaultdict(list)
+ for name, val in iface_pairs.items():
+ seen[val].append(name)
+
+ dupes = {pair: names for pair, names in seen.items() if len(names) > 1}
+ assert not dupes, "Duplicate iface constants:\n" + "".join(
+ f" {pair}: {names}\n" for pair, names in dupes.items()
+ )
+
+
+class TestIfaceFormat:
+ """Verify iface constants follow the expected format."""
+
+ def test_all_constants_are_tuples_of_str(self, iface_module):
+ iface_pairs = _get_iface_pairs(iface_module)
+ for name, val in iface_pairs.items():
+ assert isinstance(val, tuple), f"{name} should be a tuple"
+ assert len(val) == 2, f"{name} should have length 2"
+ assert isinstance(val[0], str), f"{name} method should be a string"
+ assert isinstance(val[1], str), f"{name} path should be a string"
+
+ def test_all_constants_have_uppercase_methods(self, iface_module):
+ iface_pairs = _get_iface_pairs(iface_module)
+ valid_methods = {"GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"}
+ for name, val in iface_pairs.items():
+ assert val[0].upper() in valid_methods, (
+ f"{name} has invalid method: {val[0]} — "
+ f"should be one of {valid_methods}"
+ )
+
+ def test_all_constants_have_leading_slash_path(self, iface_module):
+ iface_pairs = _get_iface_pairs(iface_module)
+ for name, val in iface_pairs.items():
+ assert val[1].startswith("/"), f"{name} path should start with /: {val[1]}"
diff --git a/tests/test_network.py b/tests/test_network.py
index 4fe9eb1..5379a8e 100644
--- a/tests/test_network.py
+++ b/tests/test_network.py
@@ -341,9 +341,9 @@ class TestGenerateNetworkFiles:
assert "cleaned" in result
paths = result["generated"]
assert len(paths) == 2
- # Check 50- prefix
- assert (tmp_network / "data" / "networkd" / "50-eth0.network").exists()
- content = (tmp_network / "data" / "networkd" / "50-eth0.network").read_text()
+ # Check 99- prefix
+ assert (tmp_network / "data" / "networkd" / "99-eth0.network").exists()
+ content = (tmp_network / "data" / "networkd" / "99-eth0.network").read_text()
assert "Name=eth0" in content
assert "Gateway=10.0.0.254" in content
@@ -366,7 +366,7 @@ class TestGenerateNetworkFiles:
data_dir.mkdir(parents=True)
# Simulate old files
(data_dir / "old-eth0.network").write_text("[Match]\nName=old-eth0\n")
- (data_dir / "50-old-eth0.network").write_text("[Match]\nName=old-eth0\n")
+ (data_dir / "99-old-eth0.network").write_text("[Match]\nName=old-eth0\n")
cfg = {"interfaces": {"eth0": {"addresses": ["10.0.0.1/24"]}}}
_net.save_config(cfg)
@@ -376,9 +376,9 @@ class TestGenerateNetworkFiles:
assert len(result["cleaned"]) == 2
# Old files are gone
assert not (data_dir / "old-eth0.network").exists()
- assert not (data_dir / "50-old-eth0.network").exists()
+ assert not (data_dir / "99-old-eth0.network").exists()
# New file exists
- assert (data_dir / "50-eth0.network").exists()
+ assert (data_dir / "99-eth0.network").exists()
def test_cleanup_only_when_no_new_interfaces(self, tmp_network):
"""Only stale cleanup, no new files."""
@@ -781,8 +781,8 @@ class TestInferDhcpRanges:
assert "eth0" in result
r = result["eth0"]
assert r["prefix"] == 24
- assert r["start"] == "192.168.1.1"
- assert r["end"] == "192.168.1.254"
+ assert r["start"] == "192.168.1.100"
+ assert r["end"] == "192.168.1.200"
def test_bare_string_address(self):
cfg = {
diff --git a/tests/test_network_integration.py b/tests/test_network_integration.py
index 93eb37a..ce1ee99 100644
--- a/tests/test_network_integration.py
+++ b/tests/test_network_integration.py
@@ -130,10 +130,10 @@ class TestDhcpRangesIntegration:
assert "lan2" in ranges
assert "wan0" not in ranges
- assert ranges["lan1"]["start"] == "192.168.1.1"
- assert ranges["lan1"]["end"] == "192.168.1.254"
- assert ranges["lan2"]["start"] == "10.10.0.1"
- assert ranges["lan2"]["end"] == "10.10.255.254"
+ assert ranges["lan1"]["start"] == "192.168.1.100"
+ assert ranges["lan1"]["end"] == "192.168.1.200"
+ assert ranges["lan2"]["start"] == "10.10.0.100"
+ assert ranges["lan2"]["end"] == "10.10.0.200"
def test_generate_then_infer(self, tmp_network):
"""End-to-end: save, generate, infer ranges."""
diff --git a/tests/test_server.py b/tests/test_server.py
index 099cc04..9388ad3 100644
--- a/tests/test_server.py
+++ b/tests/test_server.py
@@ -1,3 +1,5 @@
+import os
+from pathlib import Path
from unittest.mock import patch
import pytest
@@ -5,93 +7,106 @@ import pytest
@pytest.fixture
def client():
- from webui.server import app
-
- app.config["TESTING"] = True
- return app.test_client()
-
-
-class TestTemplateFilters:
- @pytest.fixture
- def env(self):
+ with patch("lib.logging.setup_logging"):
from webui.server import app
- return app.jinja_env
-
- def test_timestamp_filter_valid(self, env):
- result = env.filters["timestamp"]("2026-04-01T12:00:00Z")
- assert "2026-04-01" in result
-
- def test_timestamp_filter_empty(self, env):
- assert env.filters["timestamp"]("") == ""
- assert env.filters["timestamp"](None) == ""
-
- def test_timestamp_filter_invalid(self, env):
- result = env.filters["timestamp"]("not-a-date")
- assert result == "not-a-date"
-
- def test_bytes_filter_zero(self, env):
- assert env.filters["bytes"](0) == "0.0 B"
-
- def test_bytes_filter_kb(self, env):
- result = env.filters["bytes"](1536)
- assert "KB" in result
-
- def test_bytes_filter_mb(self, env):
- result = env.filters["bytes"](1500000)
- assert "MB" in result
-
- def test_bytes_filter_negative(self, env):
- assert env.filters["bytes"](-1) == "0 B"
-
- def test_bytes_filter_invalid(self, env):
- assert env.filters["bytes"]("not-a-number") == "not-a-number"
-
- def test_duration_filter_zero(self, env):
- assert env.filters["duration"](0) == "0s"
-
- def test_duration_filter_seconds(self, env):
- assert env.filters["duration"](65) == "1m 5s"
-
- def test_duration_filter_hours(self, env):
- result = env.filters["duration"](3661)
- assert "1h" in result
-
- def test_duration_filter_days(self, env):
- result = env.filters["duration"](90000)
- assert "1d" in result
-
- def test_duration_filter_invalid(self, env):
- assert env.filters["duration"]("bad") == "bad"
-
- def test_json_pretty_filter(self, env):
- result = env.filters["json_pretty"]({"key": "value"})
- assert '{"key": "value"}' in result or "key" in result
+ app.config["TESTING"] = True
+ return app.test_client()
-class TestSafelyHelper:
- def test_returns_result(self):
- from webui.server import _safely
+class TestSPARoutes:
+ def test_root_serves_index(self, client):
+ resp = client.get("/")
+ assert resp.status_code == 200
+ assert b'id="app"' in resp.data
- result = _safely(lambda: 42)
- assert result == 42
-
- def test_returns_default_on_exception(self):
- from webui.server import _safely
-
- result = _safely(lambda: 1 / 0, default=None)
- assert result is None
-
- def test_returns_custom_default(self):
- from webui.server import _safely
-
- result = _safely(lambda: 1 / 0, default="fallback")
- assert result == "fallback"
-
-
-class TestPageRoutes:
- @patch("webui.server.get")
- def test_dashboard_no_crash(self, mock_get, client):
- mock_get.return_value = {}
+ def test_spa_catch_all_serves_index(self, client):
resp = client.get("/dashboard")
assert resp.status_code == 200
+ assert b"index.html" in resp.data or b'id="app"' in resp.data
+
+ def test_spa_catch_all_other_page(self, client):
+ resp = client.get("/zones")
+ assert resp.status_code == 200
+
+ def test_api_routes_still_work(self, client):
+ resp = client.get("/api/firewall/zones")
+ assert resp.status_code in (200, 502, 503)
+
+
+class TestWsUrlGeneration:
+ def test_ws_url_ipv4_host(self, client):
+ resp = client.get("/", headers={"Host": "192.168.1.1:9090"})
+ assert b"ws://192.168.1.1:9090/ws" in resp.data
+
+ def test_ws_url_ipv6_host(self, client):
+ resp = client.get("/", headers={"Host": "[::1]:9090"})
+ assert b"ws://[::1]:9090/ws" in resp.data
+
+
+class TestApiStatusAll:
+ @patch("webui.server.get")
+ def test_success(self, mock_get, client):
+ mock_get.return_value = {"firewall": {"zones": {}}, "dnsmasq": {}}
+ resp = client.get("/api/status/all")
+ assert resp.status_code == 200
+ data = resp.get_json()
+ assert data["ok"] is True
+ assert "firewall" in data["data"]
+
+ @patch("webui.server.get")
+ def test_error(self, mock_get, client):
+ mock_get.side_effect = RuntimeError("connection refused")
+ resp = client.get("/api/status/all")
+ assert resp.status_code == 500
+ data = resp.get_json()
+ assert data["ok"] is False
+
+
+class TestBlueprintsRegistered:
+ def test_all_blueprints_registered(self, client):
+ from webui.server import BLUEPRINTS
+
+ assert len(BLUEPRINTS) == 7
+ names = [name for name, _ in BLUEPRINTS]
+ assert "firewall" in names
+ assert "network" in names
+ assert "dhcp" in names
+ assert "proxy" in names
+ assert "certs" in names
+ assert "wireguard" in names
+ assert "logs" in names
+
+
+class TestGroupWriteHandler:
+ def test_creates_file_with_group_write(self, tmp_path: Path) -> None:
+ """GroupWriteHandler creates new log files with group-write (0o664)."""
+ import contextlib
+ from logging.handlers import RotatingFileHandler
+
+ log_file = tmp_path / "test.log"
+ old = os.umask(0o022)
+ try:
+
+ class GroupWriteHandler(RotatingFileHandler):
+ def _open(self):
+ with contextlib.suppress(OSError):
+ os.chmod(self.baseFilename, 0o664)
+ saved = os.umask(0o002)
+ try:
+ fd = os.open(
+ self.baseFilename,
+ os.O_WRONLY | os.O_CREAT | os.O_APPEND,
+ 0o664,
+ )
+ finally:
+ os.umask(saved)
+ return os.fdopen(fd, "a", errors="backslashreplace")
+
+ fh = GroupWriteHandler(str(log_file))
+ fh.close()
+ finally:
+ os.umask(old)
+
+ mode = os.stat(log_file).st_mode & 0o777
+ assert mode == 0o664, f"Expected 0o664, got {oct(mode)}"
diff --git a/tests/test_state.py b/tests/test_state.py
index c468919..b4bf52d 100644
--- a/tests/test_state.py
+++ b/tests/test_state.py
@@ -56,6 +56,49 @@ class TestCollectAll:
assert "interfaces" in result
assert "timestamp" in result
+ @patch("lib.state.run")
+ def test_collect_firewall_vlan_ips_populated(self, mock_run):
+ """VLAN interfaces with @suffix in ip addr output get their IPs collected."""
+ from lib.state import _collect_firewall
+
+ def run_side(args, **kwargs):
+ if "--get-zones" in args:
+ return "public\ninternal"
+ if "--get-active-zones" in args:
+ return "public\n eth0\ninternal eth0.100"
+ if "--get-services" in args:
+ return "ssh http"
+ if "ip" in args[0]:
+ if "link" in args:
+ return (
+ "1: lo: mtu 65536\n"
+ "2: eth0: mtu 1500 link/ether aa:bb\n"
+ "3: eth0.100@eth0: mtu 1500 link/ether aa:bb\n"
+ )
+ if "addr" in args:
+ return (
+ "2: eth0 inet 192.168.1.1/24\n"
+ "3: eth0.100@if100 inet 10.0.0.1/24\n"
+ )
+ return ""
+ if "--list-all" in args:
+ return (
+ "target: default\ninterfaces: eth0\nsources: "
+ "services: \nports: \nprotocols: \nforward-ports: "
+ "masquerade: no\nics: no\nrich-rules: "
+ "icmp-blocks: \nmodule: \n"
+ )
+ return ""
+
+ mock_run.side_effect = run_side
+ result = _collect_firewall()
+ vlan_iface = next(
+ (i for i in result["interfaces"] if i["name"] == "eth0.100"), None
+ )
+ assert vlan_iface is not None, "VLAN interface should be present"
+ assert vlan_iface["ips"], "VLAN interface should have collected IPs"
+ assert "10.0.0.1/24" in vlan_iface["ips"]
+
@patch("lib.state.run_proc")
def test_collect_dnsmasq_returns_dict(self, mock_proc):
from unittest.mock import Mock
@@ -78,3 +121,57 @@ class TestCollectFailure:
s.set("firewall", None) # simulates failure
assert s.get("firewall") is None
assert s.is_populated() is False
+
+
+class TestStateVersions:
+ def test_version_starts_at_zero(self):
+ s = State()
+ versions = s.get_versions()
+ assert versions["firewall"] == 0
+ assert versions["dnsmasq"] == 0
+
+ def test_bump_increments_version(self):
+ s = State()
+ assert s.get_versions()["firewall"] == 0
+ s.bump("firewall")
+ assert s.get_versions()["firewall"] == 1
+
+ def test_bump_unknown_subsystem_noop(self):
+ s = State()
+ versions = s.get_versions()
+ s.bump("nonexistent")
+ assert versions == s.get_versions()
+
+ def test_get_updated_versions_first_call_empty(self):
+ s = State()
+ s.bump("firewall")
+ updated = s.get_updated_versions()
+ assert updated == {}
+ assert s.get_updated_versions() == {}
+
+ def test_get_updated_versions_detects_change(self):
+ s = State()
+ _ = s.get_updated_versions() # snapshot
+ s.bump("firewall")
+ updated = s.get_updated_versions()
+ assert updated["firewall"] == 1
+
+ def test_broadcast_maintains_snapshot(self):
+ s = State()
+ s.bump("firewall")
+ s.bump("dnsmasq")
+ _ = s.get_updated_versions() # snapshot at fw=1, dm=1
+ s.bump("wireguard")
+ updated = s.get_updated_versions()
+ assert updated["wireguard"] == 1
+ assert s.get_updated_versions() == {}
+
+ def test_multiple_bumps_aggregate(self):
+ s = State()
+ _ = s.get_updated_versions()
+ s.bump("firewall")
+ s.bump("firewall")
+ s.bump("dnsmasq")
+ updated = s.get_updated_versions()
+ assert updated["firewall"] == 2
+ assert updated["dnsmasq"] == 1