"""Tests for lib/system_import module.""" import json from pathlib import Path from unittest.mock import patch import pytest from lib import system_import from lib.common import save_json @pytest.fixture def temp_project(tmp_path): """Patch all module-level path constants to tmp_path subdirs.""" originals = { "PROJECT_DIR": system_import.PROJECT_DIR, "DNSMASQ_CONF": system_import.DNSMASQ_CONF, "WG_CONF": system_import.WG_CONF, "NETWORKD_DIR": system_import.NETWORKD_DIR, "NGINX_SITES_DIR": system_import.NGINX_SITES_DIR, } system_import.PROJECT_DIR = tmp_path system_import.DNSMASQ_CONF = tmp_path / "etc" / "dnsmasq.d" / "vacuum-wall.conf" system_import.WG_CONF = tmp_path / "etc" / "wireguard" / "wg0.conf" system_import.NETWORKD_DIR = tmp_path / "etc" / "systemd" / "network" system_import.NGINX_SITES_DIR = tmp_path / "data" / "nginx" / "sites-enabled" yield tmp_path system_import.PROJECT_DIR = originals["PROJECT_DIR"] system_import.DNSMASQ_CONF = originals["DNSMASQ_CONF"] system_import.WG_CONF = originals["WG_CONF"] system_import.NETWORKD_DIR = originals["NETWORKD_DIR"] system_import.NGINX_SITES_DIR = originals["NGINX_SITES_DIR"] # ────────────────────────────────────────────────────────────────────── # Dnsmasq # ────────────────────────────────────────────────────────────────────── class TestImportDnsmasq: def _write_conf(self, tmp_path, content: str) -> Path: p = tmp_path / "etc" / "dnsmasq.d" p.mkdir(parents=True, exist_ok=True) (p / "vacuum-wall.conf").write_text(content) return p / "vacuum-wall.conf" def _read_json(self, tmp_path) -> dict: p = tmp_path / "config" / "dnsmasq" / "config.json" return json.loads(p.read_text()) if p.exists() else {} def test_no_conf_file(self, temp_project): assert not system_import.import_dnsmasq() def test_no_markers(self, temp_project, tmp_path): self._write_conf(tmp_path, "# some random config\nserver=1.1.1.1\n") assert not system_import.import_dnsmasq() def test_empty_managed_block(self, temp_project, tmp_path): self._write_conf(tmp_path, f"{system_import.DNSTART}\n{system_import.DNEND}") assert system_import.import_dnsmasq() cfg = self._read_json(tmp_path) assert cfg["dns"]["upstreams"] == [] assert cfg["dhcp"]["ranges"] == [] def test_upstreams_only(self, temp_project, tmp_path): conf = ( f"{system_import.DNSTART}\n" "server=8.8.8.8\n" "server=1.1.1.1\n" f"{system_import.DNEND}" ) self._write_conf(tmp_path, conf) assert system_import.import_dnsmasq() cfg = self._read_json(tmp_path) assert cfg["dns"]["upstreams"] == ["8.8.8.8", "1.1.1.1"] def test_full_config(self, temp_project, tmp_path): conf = ( f"{system_import.DNSTART}\n" "server=8.8.8.8\n" "server=1.1.1.1\n" "domain=lan\n" "expand-hosts\n" "dhcp-range=set:eth1,192.168.2.100,192.168.2.200,12h\n" "dhcp-option=tag:eth1,3,192.168.2.1\n" "dhcp-option=tag:eth1,6,192.168.2.1\n" "dhcp-host=aa:bb:cc:dd:ee:ff,192.168.2.50,printer\n" "addr/nas.lan/192.168.2.10\n" f"{system_import.DNEND}" ) self._write_conf(tmp_path, conf) assert system_import.import_dnsmasq() cfg = self._read_json(tmp_path) assert cfg["dns"]["upstreams"] == ["8.8.8.8", "1.1.1.1"] assert cfg["dns"]["domain"] == "lan" assert len(cfg["dhcp"]["ranges"]) == 1 rng = cfg["dhcp"]["ranges"][0] assert rng["interface"] == "eth1" assert rng["start"] == "192.168.2.100" assert rng["end"] == "192.168.2.200" assert rng["lease_time"] == "12h" assert rng["gateway"] == "192.168.2.1" assert rng["dns"] == "192.168.2.1" assert len(cfg["dhcp"]["static_leases"]) == 1 lease = cfg["dhcp"]["static_leases"][0] assert lease["mac"] == "aa:bb:cc:dd:ee:ff" assert lease["ip"] == "192.168.2.50" assert lease["hostname"] == "printer" assert len(cfg["dns"]["custom_records"]) == 1 assert cfg["dns"]["custom_records"][0] == { "name": "nas.lan", "address": "192.168.2.10", } def test_no_resolv_resets_upstreams(self, temp_project, tmp_path): conf = f"{system_import.DNSTART}\nno-resolv\n{system_import.DNEND}" self._write_conf(tmp_path, conf) assert system_import.import_dnsmasq() cfg = self._read_json(tmp_path) assert cfg["dns"]["upstreams"] == [] def test_idempotent(self, temp_project, tmp_path): conf = f"{system_import.DNSTART}\nserver=8.8.8.8\n{system_import.DNEND}" self._write_conf(tmp_path, conf) assert system_import.import_dnsmasq() assert not system_import.import_dnsmasq() def test_parse_error_returns_false(self, temp_project, tmp_path): # Conf with markers — parses fine, so this tests the exception handler # by mocking _parse_dnsmasq_block to raise conf = f"{system_import.DNSTART}\nserver=8.8.8.8\n{system_import.DNEND}" self._write_conf(tmp_path, conf) with patch( "lib.system_import._parse_dnsmasq_block", side_effect=ValueError("bad") ): assert not system_import.import_dnsmasq() # ────────────────────────────────────────────────────────────────────── # WireGuard # ────────────────────────────────────────────────────────────────────── class TestImportWireguard: def _write_conf(self, tmp_path, content: str) -> Path: p = tmp_path / "etc" / "wireguard" p.mkdir(parents=True, exist_ok=True) (p / "wg0.conf").write_text(content) return p / "wg0.conf" def _read_json(self, tmp_path) -> dict: p = tmp_path / "config" / "wireguard" / "config.json" return json.loads(p.read_text()) if p.exists() else {} def test_no_conf_file(self, temp_project): assert not system_import.import_wireguard() def test_basic_interface(self, temp_project, tmp_path): conf = ( "[Interface]\n" " PrivateKey = abc123\n" " Address = 10.137.0.1/24\n" " ListenPort = 51820\n" ) self._write_conf(tmp_path, conf) assert system_import.import_wireguard() cfg = self._read_json(tmp_path) assert cfg["interface"]["private_key"] == "abc123" assert cfg["interface"]["addresses"] == ["10.137.0.1/24"] assert cfg["interface"]["listen_port"] == 51820 assert cfg["interface"]["name"] == "wg0" assert cfg["peers"] == {} def test_full_with_peers(self, temp_project, tmp_path): conf = ( "[Interface]\n" " PrivateKey = srv-priv\n" " Address = 10.137.0.1/24\n" " ListenPort = 51820\n" " PostUp = iptables -I FORWARD -i wg0 -j ACCEPT\n" " PostDown = iptables -D FORWARD -i wg0 -j ACCEPT\n" "\n" "[Peer] # alice\n" " PublicKey = alice-pub\n" " Endpoint = 203.0.113.1:51820\n" " AllowedIPs = 0.0.0.0/0\n" " PersistentKeepalive = 25\n" "\n" "[Peer] # bob\n" " PublicKey = bob-pub\n" " AllowedIPs = 10.0.0.0/8,172.16.0.0/12\n" ) self._write_conf(tmp_path, conf) assert system_import.import_wireguard() cfg = self._read_json(tmp_path) assert cfg["interface"]["post_up"] == "iptables -I FORWARD -i wg0 -j ACCEPT" assert cfg["interface"]["post_down"] == "iptables -D FORWARD -i wg0 -j ACCEPT" assert "alice" in cfg["peers"] assert cfg["peers"]["alice"]["public_key"] == "alice-pub" assert cfg["peers"]["alice"]["endpoint"] == "203.0.113.1:51820" assert cfg["peers"]["alice"]["allowed_ips"] == ["0.0.0.0/0"] assert cfg["peers"]["alice"]["persistent_keepalive"] == 25 assert "bob" in cfg["peers"] assert cfg["peers"]["bob"]["allowed_ips"] == ["10.0.0.0/8", "172.16.0.0/12"] def test_peer_without_name_uses_pubkey(self, temp_project, tmp_path): conf = ( "[Interface]\n" " PrivateKey = srv-priv\n" " Address = 10.137.0.1/24\n" " ListenPort = 51820\n" "\n" "[Peer]\n" " PublicKey = anon-pub\n" " AllowedIPs = 0.0.0.0/0\n" ) self._write_conf(tmp_path, conf) assert system_import.import_wireguard() cfg = self._read_json(tmp_path) assert "anon-pub" in cfg["peers"] def test_idempotent(self, temp_project, tmp_path): conf = ( "[Interface]\n" " PrivateKey = abc123\n" " Address = 10.137.0.1/24\n" " ListenPort = 51820\n" ) self._write_conf(tmp_path, conf) assert system_import.import_wireguard() assert not system_import.import_wireguard() # ────────────────────────────────────────────────────────────────────── # Networkd # ────────────────────────────────────────────────────────────────────── class TestImportNetworkd: def _write_network(self, tmp_path, name: str, content: str) -> Path: p = tmp_path / "etc" / "systemd" / "network" p.mkdir(parents=True, exist_ok=True) file_path = p / f"99-{name}.network" file_path.write_text(content) return file_path def _read_json(self, tmp_path) -> dict: p = tmp_path / "config" / "network" / "config.json" return json.loads(p.read_text()) if p.exists() else {} def test_no_network_dir(self, temp_project): assert not system_import.import_networkd() def test_no_files(self, temp_project, tmp_path): (tmp_path / "etc" / "systemd" / "network").mkdir(parents=True, exist_ok=True) assert not system_import.import_networkd() def test_basic_interface(self, temp_project, tmp_path): conf = ( "[Match]\n" "Name=eth0\n" "\n" "[Network]\n" "DHCP=no\n" "Addresses=192.168.1.1/24\n" "Gateway=192.168.1.254\n" "DNS=8.8.8.8\n" "DNS=1.1.1.1\n" ) self._write_network(tmp_path, "eth0", conf) assert system_import.import_networkd() cfg = self._read_json(tmp_path) eth0 = cfg["interfaces"]["eth0"] assert eth0["dhcp"] == "no" assert eth0["gateway"] == "192.168.1.254" assert eth0["dns"] == ["8.8.8.8", "1.1.1.1"] def test_multiple_interfaces(self, temp_project, tmp_path): self._write_network( tmp_path, "eth0", "[Match]\nName=eth0\n\n[Network]\nDHCP=yes\n" ) self._write_network( tmp_path, "eth1", "[Match]\nName=eth1\n\n[Network]\nDHCP=no\n" ) assert system_import.import_networkd() cfg = self._read_json(tmp_path) assert "eth0" in cfg["interfaces"] assert "eth1" in cfg["interfaces"] assert cfg["interfaces"]["eth0"]["dhcp"] == "yes" assert cfg["interfaces"]["eth1"]["dhcp"] == "no" def test_preserves_existing_interfaces(self, temp_project, tmp_path): # Pre-existing JSON has eth2 with no .network file cfg_path = tmp_path / "config" / "network" cfg_path.mkdir(parents=True, exist_ok=True) save_json(cfg_path / "config.json", {"interfaces": {"eth2": {"dhcp": "no"}}}) self._write_network( tmp_path, "eth0", "[Match]\nName=eth0\n\n[Network]\nDHCP=yes\n" ) assert system_import.import_networkd() cfg = self._read_json(tmp_path) assert "eth0" in cfg["interfaces"] assert "eth2" in cfg["interfaces"] def test_idempotent(self, temp_project, tmp_path): self._write_network( tmp_path, "eth0", "[Match]\nName=eth0\n\n[Network]\nDHCP=yes\n" ) assert system_import.import_networkd() assert not system_import.import_networkd() def test_address_section_parsed(self, temp_project, tmp_path): conf = ( "[Match]\n" "Name=eth0\n" "\n" "[Network]\n" "DHCP=no\n" "\n" "[Address]\n" "Address=192.168.1.1/24\n" ) self._write_network(tmp_path, "eth0", conf) assert system_import.import_networkd() cfg = self._read_json(tmp_path) eth0 = cfg["interfaces"]["eth0"] assert "192.168.1.1/24" in eth0.get("addresses", []) def test_route_section_parsed(self, temp_project, tmp_path): conf = ( "[Match]\n" "Name=eth0\n" "\n" "[Network]\n" "DHCP=no\n" "\n" "[Route]\n" "Destination=10.0.0.0/8\n" "Gateway=192.168.1.254\n" "Metric=100\n" ) self._write_network(tmp_path, "eth0", conf) assert system_import.import_networkd() cfg = self._read_json(tmp_path) eth0 = cfg["interfaces"]["eth0"] assert len(eth0.get("routes", [])) == 1 route = eth0["routes"][0] assert route["destination"] == "10.0.0.0/8" assert route["gateway"] == "192.168.1.254" assert route["metric"] == 100 # ────────────────────────────────────────────────────────────────────── # Nginx # ────────────────────────────────────────────────────────────────────── class TestImportNginx: def _write_site(self, tmp_path, domain: str, content: str) -> Path: p = tmp_path / "data" / "nginx" / "sites-enabled" p.mkdir(parents=True, exist_ok=True) file_path = p / f"{domain}.conf" file_path.write_text(content) return file_path def _read_json(self, tmp_path) -> dict: p = tmp_path / "config" / "nginx" / "config.json" return json.loads(p.read_text()) if p.exists() else {} def test_no_sites_dir(self, temp_project): assert not system_import.import_nginx() def test_no_files(self, temp_project, tmp_path): (tmp_path / "data" / "nginx" / "sites-enabled").mkdir( parents=True, exist_ok=True ) assert not system_import.import_nginx() def test_acme_challenge_skipped(self, temp_project, tmp_path): (tmp_path / "data" / "nginx" / "sites-enabled").mkdir( parents=True, exist_ok=True ) ( tmp_path / "data" / "nginx" / "sites-enabled" / "_acme-challenge.conf" ).write_text("# stuff\n") assert not system_import.import_nginx() def test_unrecognized_file_skipped(self, temp_project, tmp_path): self._write_site(tmp_path, "my-site", "# some random nginx config\nserver {}\n") assert not system_import.import_nginx() def test_basic_site(self, temp_project, tmp_path): conf = ( "# Auto-generated by Vacuum Wall — do not edit manually\n" "# Domain: example.com\n" "\n" "server {\n" " listen 80;\n" " listen [::]:80;\n" " server_name example.com;\n" " return 301 https://$host$request_uri;\n" "}\n" "\n" "server {\n" " listen 443 ssl;\n" " listen [::]:443 ssl;\n" " server_name example.com;\n" "\n" " ssl_certificate /home/wall/vacuum-wall/data/acme/example.com/fullchain.cer;\n" " ssl_certificate_key /home/wall/vacuum-wall/data/acme/example.com/example.com.key;\n" "\n" " # / -> 192.168.2.50:8080\n" " location / {\n" " auth_basic off;\n" " proxy_pass http://192.168.2.50:8080;\n" " }\n" "}\n" ) self._write_site(tmp_path, "example.com", conf) assert system_import.import_nginx() cfg = self._read_json(tmp_path) assert "example.com" in cfg["domains"] dom = cfg["domains"]["example.com"] assert dom["force_ssl"] is True assert dom["cert"] == "acme" assert "/" in dom["paths"] assert dom["paths"]["/"]["backend"]["host"] == "192.168.2.50" assert dom["paths"]["/"]["backend"]["port"] == 8080 def test_websocket_path(self, temp_project, tmp_path): conf = ( "# Auto-generated by Vacuum Wall — do not edit manually\n" "# Domain: example.com\n" "\n" "server {\n" " listen 443 ssl;\n" " server_name example.com;\n" "\n" " ssl_certificate /data/certs/example.com.crt;\n" " ssl_certificate_key /data/certs/example.com.key;\n" "\n" " # / -> 127.0.0.1:9090\n" " location / {\n" " auth_basic off;\n" " proxy_pass http://127.0.0.1:9090;\n" " }\n" "\n" " # /ws -> 127.0.0.1:9091 (WebSocket)\n" " location /ws {\n" " auth_basic off;\n" " proxy_pass http://127.0.0.1:9091;\n" " }\n" "}\n" ) self._write_site(tmp_path, "example.com", conf) assert system_import.import_nginx() cfg = self._read_json(tmp_path) dom = cfg["domains"]["example.com"] assert dom["cert"] == "selfsigned" assert dom["paths"]["/ws"]["is_websocket"] is True assert dom["paths"]["/ws"]["backend"]["port"] == 9091 def test_idempotent(self, temp_project, tmp_path): conf = ( "# Auto-generated by Vacuum Wall — do not edit manually\n" "server {\n" " listen 443 ssl;\n" " server_name example.com;\n" " ssl_certificate /data/acme/example.com/fullchain.cer;\n" " ssl_certificate_key /data/acme/example.com/example.com.key;\n" " # / -> 127.0.0.1:9090\n" " location / {\n" " auth_basic off;\n" " proxy_pass http://127.0.0.1:9090;\n" " }\n" "}\n" ) self._write_site(tmp_path, "example.com", conf) assert system_import.import_nginx() assert not system_import.import_nginx() # ────────────────────────────────────────────────────────────────────── # Firewall # ────────────────────────────────────────────────────────────────────── FIREWALL_ZONES_OUTPUT = ( "public (active)\n" " target: default\n" " interfaces: eth0 eth1\n" " sources: \n" " services: dhcpv6-cidr dns mdns ssh\n" " ports: \n" " protocols: \n" " forward-ports: \n" " source-ports: \n" " icmp-blocks: \n" " rich rules: \n" "\n" "internal (active)\n" " target: DEFAULT\n" " interfaces: eth2\n" " sources: \n" " services: dhcpv6-cidr dns mdns samba-client ssh\n" " ports: \n" " protocols: \n" " forward-ports: \n" " source-ports: \n" " icmp-blocks: \n" " rich rules: \n" "\n" "dmz (active)\n" " target: DROP\n" " interfaces: \n" " sources: \n" " services: dns\n" " ports: \n" " protocols: \n" " forward-ports: \n" " source-ports: \n" " icmp-blocks: \n" " rich rules: \n" ) class TestImportFirewall: def _read_json(self, tmp_path) -> dict: p = tmp_path / "config" / "firewall" / "config.json" return json.loads(p.read_text()) if p.exists() else {} def test_no_config_file_and_firewalld_down(self, temp_project, tmp_path): with patch( "lib.system_import.run", side_effect=RuntimeError("firewalld not running") ): assert not system_import.import_firewall() def test_existing_config_not_overwritten(self, temp_project, tmp_path): cfg_path = tmp_path / "config" / "firewall" cfg_path.mkdir(parents=True, exist_ok=True) save_json( cfg_path / "config.json", {"zones": {"public": {"interfaces": ["eth0"]}}} ) assert not system_import.import_firewall() def test_import_zones(self, temp_project, tmp_path): with patch("lib.system_import.run", return_value=FIREWALL_ZONES_OUTPUT): assert system_import.import_firewall() cfg = self._read_json(tmp_path) assert "zones" in cfg assert "public" in cfg["zones"] assert "internal" in cfg["zones"] assert cfg["zones"]["public"]["target"] == "DEFAULT" assert cfg["zones"]["public"]["interfaces"] == ["eth0", "eth1"] assert cfg["zones"]["public"]["services"] == [ "dhcpv6-cidr", "dns", "mdns", "ssh", ] assert cfg["zones"]["internal"]["target"] == "DEFAULT" assert cfg["zones"]["internal"]["interfaces"] == ["eth2"] def test_empty_interface_zones_skipped(self, temp_project, tmp_path): with patch("lib.system_import.run", return_value=FIREWALL_ZONES_OUTPUT): assert system_import.import_firewall() cfg = self._read_json(tmp_path) assert "dmz" not in cfg["zones"] def test_parse_error_returns_false(self, temp_project, tmp_path): with patch("lib.system_import.run", return_value="garbage with no valid zones"): assert not system_import.import_firewall() # ────────────────────────────────────────────────────────────────────── # import_all # ────────────────────────────────────────────────────────────────────── class TestImportAll: def test_all_missing(self, temp_project, tmp_path): from unittest.mock import MagicMock, patch mock_run = MagicMock(side_effect=RuntimeError("command not found")) with patch.object(system_import, "run", mock_run): result = system_import.import_all() assert result == [] def test_returns_updated_subsystems(self, temp_project, tmp_path): # Create dnsmasq conf etc = tmp_path / "etc" / "dnsmasq.d" etc.mkdir(parents=True, exist_ok=True) conf = f"{system_import.DNSTART}\nserver=8.8.8.8\n{system_import.DNEND}" (etc / "vacuum-wall.conf").write_text(conf) # Create wireguard conf wg_etc = tmp_path / "etc" / "wireguard" wg_etc.mkdir(parents=True, exist_ok=True) (wg_etc / "wg0.conf").write_text( "[Interface]\n PrivateKey = abc\n Address = 10.137.0.1/24\n ListenPort = 51820\n" ) result = system_import.import_all() assert "dnsmasq" in result assert "wireguard" in result assert "network" not in result assert "nginx" not in result def test_parse_error_does_not_crash(self, temp_project, tmp_path): # Create a dnsmasq conf that will parse fine etc = tmp_path / "etc" / "dnsmasq.d" etc.mkdir(parents=True, exist_ok=True) conf = f"{system_import.DNSTART}\nserver=8.8.8.8\n{system_import.DNEND}" (etc / "vacuum-wall.conf").write_text(conf) # Make wireguard import fail wg_etc = tmp_path / "etc" / "wireguard" wg_etc.mkdir(parents=True, exist_ok=True) (wg_etc / "wg0.conf").write_text("[Interface]\n") # This should not raise, just log warning result = system_import.import_all() assert "dnsmasq" in result # ────────────────────────────────────────────────────────────────────── # _cfgs_equal # ────────────────────────────────────────────────────────────────────── class TestCfgsEqual: def test_equal(self): assert system_import._cfgs_equal({"a": 1}, {"a": 1}) def test_not_equal(self): assert not system_import._cfgs_equal({"a": 1}, {"a": 2}) def test_ignores_applied_hash(self): a = {"a": 1, "_last_applied_hash": "abc"} b = {"a": 1, "_last_applied_hash": "xyz"} assert system_import._cfgs_equal(a, b) def test_nested(self): a = {"dhcp": {"ranges": [{"start": "1.2.3.4"}]}} b = {"dhcp": {"ranges": [{"start": "1.2.3.4"}]}} assert system_import._cfgs_equal(a, b) # ────────────────────────────────────────────────────────────────────── # Template marker consistency # ────────────────────────────────────────────────────────────────────── class TestTemplateMarker: """Verify nginx templates contain the expected auto-generated marker.""" def test_acme_challenge_has_marker(self): content = ( Path(__file__).resolve().parent.parent / "system" / "nginx" / "acme-challenge.conf" ).read_text() assert "# Auto-generated by Vacuum Wall" in content def test_server_block_has_marker(self): content = ( Path(__file__).resolve().parent.parent / "system" / "nginx" / "server_block.conf" ).read_text() assert "# Auto-generated by Vacuum Wall" in content