From 7abe7700e9f5d15608615a301a581b45a4a541d7 Mon Sep 17 00:00:00 2001 From: Mike Teehan Date: Tue, 16 Jun 2026 03:37:00 +0000 Subject: [PATCH] test: update and add tests for all updated subsystems --- tests/test-reactive-dom.js | 120 ++++++++++++++++++++ tests/test_client.py | 74 ++++++++++++ tests/test_firewall.py | 2 +- tests/test_handler_acme.py | 84 ++++++++++++++ tests/test_handler_network.py | 108 +++++++++++++++++- tests/test_iface_sync.py | 145 ++++++++++++++++++++++++ tests/test_network.py | 16 +-- tests/test_network_integration.py | 8 +- tests/test_server.py | 181 ++++++++++++++++-------------- tests/test_state.py | 97 ++++++++++++++++ 10 files changed, 733 insertions(+), 102 deletions(-) create mode 100644 tests/test-reactive-dom.js create mode 100644 tests/test_client.py create mode 100644 tests/test_handler_acme.py create mode 100644 tests/test_iface_sync.py 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