mteehan/vacuum-wall
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
43b44ad340bb4230753f6b0b59b41ffa850b14b0
vacuum-wall/tests/test_system_import.py
T
mteehan 04417cf05c WireGuard access classes, firewall nftables fixes, network sync event refactor 
- WireGuard: refactor to multi-interface 'access classes' model; extract config
  generation and helpers into lib/wireguard.py; add per-class up/down endpoints
  and API routes; update UI with class management pages and QR code component
- Firewall: fix zone creation with --new-zone before --set-target; skip
  masquerade on public zone; add masquerade propagation for nftables backend
  so NAT works when internal zones exit via public
- Network: rename sync event subsystem 'network' -> 'networkd'; always stamp
  config hash even when deployment fails (fixes pending-changes detection)
- DHCP: add new API endpoint and update frontend page
- State/Sync: update state collectors and sync buses for new subsystems
- Docs: update API and config documentation for new endpoints and schemas
2026-07-20 03:57:16 +00:00

676 lines
27 KiB
Python
Raw Blame History

"""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
Reference in New Issue View Git Blame Copy Permalink