Optimize firewall state collection and improve daemon shutdown

- Replace per-zone --list-all calls with single --list-all-zones in _collect_firewall - Add _parse_all_zones_output() parser with rich rules/rich-rules normalization - Convert daemon shutdown to async with proper runner cleanup and socket unlink - Add TimeoutStopSec=15 to vacuum-walld.service for graceful stop - Fix exception handling in _collect_dnsmasq - Remove management badge from proxy path rows
2026-06-28 00:54:01 +00:00
parent 80dd4e3272
commit 25a1943fce
7 changed files with 190 additions and 33 deletions
@@ -542,3 +542,89 @@ class TestLibParseForwardPorts:

    def test_empty_string(self):
        assert firewall._parse_forward_ports("") == []
+
+
+# ---------------------------------------------------------------------------
+# lib/firewall.py — parse all zones output (--list-all-zones)
+# ---------------------------------------------------------------------------
+
+
+class TestParseAllZonesOutput:
+    def test_parses_single_zone(self):
+        result = firewall._parse_all_zones_output(
+            "public\n"
+            "  target: default\n"
+            "  interfaces: eth0\n"
+            "  services: ssh http\n"
+            "  masquerade: yes\n"
+            "  rich rules: \n"
+        )
+        assert "public" in result
+        assert result["public"]["name"] == "public"
+        assert result["public"]["interfaces"] == ["eth0"]
+        assert result["public"]["services"] == ["ssh", "http"]
+        assert result["public"]["masquerade"] is True
+        assert result["public"]["rich-rules"] == []
+
+    def test_parses_multiple_zones(self):
+        result = firewall._parse_all_zones_output(
+            "public (default, active)\n"
+            "  target: default\n"
+            "  interfaces: eth0\n"
+            "  services: ssh\n"
+            "  masquerade: no\n"
+            "  rich rules: \n"
+            "internal (active)\n"
+            "  target: ACCEPT\n"
+            "  interfaces: eth1\n"
+            "  services: dhcp\n"
+            "  masquerade: no\n"
+            "  rich rules: \n"
+            "trusted\n"
+            "  target: ACCEPT\n"
+            "  interfaces: \n"
+            "  services: \n"
+            "  masquerade: no\n"
+            "  rich rules: \n"
+        )
+        assert set(result.keys()) == {"public", "internal", "trusted"}
+        assert result["public"]["interfaces"] == ["eth0"]
+        assert result["internal"]["target"] == "ACCEPT"
+        assert result["trusted"]["services"] == []
+
+    def test_empty_output(self):
+        assert firewall._parse_all_zones_output("") == {}
+        assert firewall._parse_all_zones_output("\n  \n") == {}
+
+    def test_handles_blank_lines_between_zones(self):
+        result = firewall._parse_all_zones_output(
+            "public\n"
+            "  target: default\n"
+            "  interfaces: eth0\n"
+            "  rich rules: \n"
+            "\n"
+            "internal\n"
+            "  target: ACCEPT\n"
+            "  interfaces: eth1\n"
+            "  rich rules: \n"
+        )
+        assert "public" in result
+        assert "internal" in result
+        assert result["public"]["interfaces"] == ["eth0"]
+        assert result["internal"]["interfaces"] == ["eth1"]
+
+    def test_all_default_fields_present(self):
+        result = firewall._parse_all_zones_output(
+            "dmz\n"
+            "  target: default\n"
+            "  interfaces: \n"
+            "  services: \n"
+            "  rich rules: \n"
+        )
+        zone = result["dmz"]
+        for field in (
+            "interfaces", "sources", "services", "ports", "protocols",
+            "forward-ports", "masquerade", "ics", "icmp-blocks", "module",
+            "target", "rich-rules",
+        ):
+            assert field in zone, f"Missing field: {field}"