vacuum-wall/tests/test_state.py

"""Tests for lib/state.py — state store and collect functions."""

from unittest.mock import patch

from lib.state import State, state


class TestState:
    def test_new_state_empty(self):
        s = State()
        assert s.get("firewall") is None
        assert s.is_populated() is False

    def test_set_and_get(self):
        s = State()
        s.set("firewall", {"zones": {"public": {}}})
        assert s.get("firewall") == {"zones": {"public": {}}}

    def test_populate_all(self):
        s = State()
        with patch.object(s, "_data", {}):
            pass
        # Just verify populate doesn't crash on empty collectors
        # (our collect functions need subprocess, so test mocks only)
        pass

    def test_singleton_exists(self):
        assert state is not None
        assert isinstance(state, State)

    def test_get_snapshot_empty(self):
        """Fresh store: snapshot lists every subsystem, all None."""
        s = State()
        snap = s.get_snapshot()
        assert set(snap) == set(s.SUBSYSTEMS)
        assert all(v is None for v in snap.values())

    def test_get_snapshot_reflects_set_and_none(self):
        """Snapshot carries set data; failed collections stay None."""
        s = State()
        s.set("firewall", {"zones": {}})
        s.set("system", {"load": {"load1": 0.0}})
        s.set("dnsmasq", None)
        snap = s.get_snapshot()
        assert snap["firewall"] == {"zones": {}}
        assert snap["system"] == {"load": {"load1": 0.0}}
        assert snap["dnsmasq"] is None
        assert snap["acme"] is None


class TestCollectAll:
    @patch("lib.state.run")
    def test_collect_firewall_returns_dict(self, mock_run):
        from lib.state import _collect_firewall

        def run_side(args, **kwargs):
            if "--get-active-zones" in args:
                return "public\n  eth0"
            if "--get-default-zone" in args:
                return "public\n"
            if "--get-services" in args:
                return "ssh http"
            if "ip" in args[0]:
                if "link" in args:
                    return "1: lo: <LOOPBACK> mtu 65536\n2: eth0: <UP> mtu 1500 link/ether aa:bb\n"
                return ""
            if "--list-all-zones" in args:
                return (
                    "public\n"
                    "  target: default\n"
                    "  interfaces: eth0\n"
                    "  services: \n"
                    "  ports: \n"
                    "  protocols: \n"
                    "  forward-ports: \n"
                    "  masquerade: no\n"
                    "  rich rules: \n"
                )

        mock_run.side_effect = run_side
        result = _collect_firewall()
        assert isinstance(result, dict)
        assert "active_zones" in result
        assert "default_zone" in result
        assert result["default_zone"] == "public"
        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-active-zones" in args:
                return "public\n  eth0\ninternal\n  eth0.100"
            if "--get-default-zone" in args:
                return "public\n"
            if "--get-services" in args:
                return "ssh http"
            if "ip" in args[0]:
                if "link" in args:
                    return (
                        "1: lo: <LOOPBACK> mtu 65536\n"
                        "2: eth0: <UP> mtu 1500 link/ether aa:bb\n"
                        "3: eth0.100@eth0: <UP> 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-zones" in args:
                return (
                    "public\n"
                    "  target: default\n"
                    "  interfaces: eth0\n"
                    "  services: \n"
                    "  ports: \n"
                    "  protocols: \n"
                    "  forward-ports: \n"
                    "  masquerade: no\n"
                    "  rich rules: \n"
                    "internal\n"
                    "  target: ACCEPT\n"
                    "  interfaces: eth0.100\n"
                    "  services: \n"
                    "  ports: \n"
                    "  protocols: \n"
                    "  forward-ports: \n"
                    "  masquerade: no\n"
                    "  rich rules: \n"
                )

        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

        from lib.state import _collect_dnsmasq

        mock_proc.return_value = Mock(stdout="active\n", returncode=0)
        result = _collect_dnsmasq()
        assert isinstance(result, dict)
        assert "status" in result
        assert "config" in result
        assert "leases" in result


class TestCollectFailure:
    def test_state_clears_on_failure(self):
        """State collection failure sets the subsystem to None."""
        s = State()
        s.set("firewall", {"zones": {"public": {}}})
        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


class TestStripVolatile:
    def test_list_of_dicts_strips_nested_keys(self):
        """Wireguard-style: status.peers[].transfer_received gets stripped."""
        from lib.state import _strip_volatile

        data = {
            "status": {
                "peers": [
                    {
                        "transfer_received": "100B",
                        "transfer_sent": "200B",
                        "latest_handshake": "ago",
                        "public_key": "abc",
                    },
                    {
                        "transfer_received": "300B",
                        "transfer_sent": "400B",
                        "latest_handshake": "now",
                        "public_key": "def",
                    },
                ]
            }
        }
        vol = frozenset(
            {"status.peers[].transfer_received", "status.peers[].latest_handshake"}
        )
        stripped = _strip_volatile(data, vol)
        assert stripped["status"]["peers"][0]["transfer_received"] is None
        assert stripped["status"]["peers"][0]["latest_handshake"] is None
        assert stripped["status"]["peers"][0]["transfer_sent"] == "200B"
        assert stripped["status"]["peers"][0]["public_key"] == "abc"
        assert stripped["status"]["peers"][1]["transfer_received"] is None

    def test_scalar_path_strips(self):
        """Scalar paths get zeroed to None."""
        from lib.state import _strip_volatile

        data = {"a": {"b": 1, "c": 2}}
        vol = frozenset({"a.b"})
        stripped = _strip_volatile(data, vol)
        assert stripped["a"]["b"] is None
        assert stripped["a"]["c"] == 2

    def test_empty_volatile_returns_copy(self):
        from lib.state import _strip_volatile

        data = {"x": 1}
        stripped = _strip_volatile(data, frozenset())
        assert stripped == data
        assert stripped is not data

    def test_firewall_volatile_strips_ips(self):
        """Firewall-style: interfaces[].ips gets stripped."""
        from lib.state import _strip_volatile

        data = {
            "interfaces": [
                {
                    "name": "eth0",
                    "ips": ["10.0.0.1/24"],
                    "ipv6": ["fe80::1"],
                    "zone": "internal",
                },
            ]
        }
        vol = frozenset({"interfaces[].ips", "interfaces[].ipv6"})
        stripped = _strip_volatile(data, vol)
        assert stripped["interfaces"][0]["ips"] is None
        assert stripped["interfaces"][0]["ipv6"] is None
        assert stripped["interfaces"][0]["name"] == "eth0"
        assert stripped["interfaces"][0]["zone"] == "internal"

    def test_networkd_volatile_strips_addresses(self):
        """Networkd-style: interfaces dict keyed by name, addresses stripped via fallback."""
        from lib.state import _strip_volatile

        data = {
            "interfaces": {
                "eth0": {
                    "addresses": ["10.0.0.1/24", "fe80::1"],
                    "state": "routable",
                    "type": "ether",
                },
                "lo": {
                    "addresses": ["127.0.0.1/8"],
                    "state": "degraded",
                    "type": "loopback",
                },
            }
        }
        vol = frozenset({"interfaces[].addresses"})
        stripped = _strip_volatile(data, vol)
        assert stripped["interfaces"]["eth0"]["addresses"] is None
        assert stripped["interfaces"]["eth0"]["state"] == "routable"
        assert stripped["interfaces"]["eth0"]["type"] == "ether"
        assert stripped["interfaces"]["lo"]["addresses"] is None
        assert stripped["interfaces"]["lo"]["state"] == "degraded"


class TestDiffLayers:
    def test_no_change(self):
        """Identical data (minus timestamp) returns (False, False)."""
        from lib.state import _diff_layers

        old = {"zones": {"public": {}}, "timestamp": "t1"}
        new = {"zones": {"public": {}}, "timestamp": "t2"}
        structural, volatile = _diff_layers(old, new, frozenset())
        assert structural is False
        assert volatile is False

    def test_structural_only(self):
        """Non-volatile change detected as structural."""
        from lib.state import _diff_layers

        old = {
            "zones": {"public": {}},
            "interfaces": [{"ips": None}],
            "timestamp": "t1",
        }
        new = {
            "zones": {"internal": {}},
            "interfaces": [{"ips": None}],
            "timestamp": "t2",
        }
        vol = frozenset({"interfaces[].ips"})
        structural, volatile = _diff_layers(old, new, vol)
        assert structural is True
        assert volatile is False

    def test_volatile_only(self):
        """Only volatile fields changed returns (False, True)."""
        from lib.state import _diff_layers

        old = {
            "status": {"peers": [{"transfer_received": "100B", "public_key": "abc"}]},
            "timestamp": "t1",
        }
        new = {
            "status": {"peers": [{"transfer_received": "200B", "public_key": "abc"}]},
            "timestamp": "t2",
        }
        vol = frozenset({"status.peers[].transfer_received"})
        structural, volatile = _diff_layers(old, new, vol)
        assert structural is False
        assert volatile is True

    def test_neither(self):
        """No change at all returns (False, False)."""
        from lib.state import _diff_layers

        old = {
            "status": {"peers": [{"transfer_received": "100B", "public_key": "abc"}]},
            "timestamp": "t1",
        }
        new = {
            "status": {"peers": [{"transfer_received": "100B", "public_key": "abc"}]},
            "timestamp": "t2",
        }
        vol = frozenset({"status.peers[].transfer_received"})
        structural, volatile = _diff_layers(old, new, vol)
        assert structural is False
        assert volatile is False

    def test_old_none_returns_both_true(self):
        """When old is None (first poll), both layers report True."""
        from lib.state import _diff_layers

        new = {"zones": {}}
        structural, volatile = _diff_layers(None, new, frozenset())
        assert structural is True
        assert volatile is True


class TestPoll:
    def test_poll_no_change(self):
        """poll() returns (False, False) when collector returns same data."""
        import uuid

        from lib.state import _COLLECTORS, State

        name = f"test_{uuid.uuid4().hex}"
        s = State()
        data = {"value": 1, "timestamp": "t1"}
        _COLLECTORS[name] = lambda: data
        s.set(name, data)
        structural, volatile = s.poll(name)
        assert structural is False
        assert volatile is False
        del _COLLECTORS[name]

    def test_poll_detects_structural(self):
        """poll() detects structural changes via collector."""
        import uuid

        from lib.state import _COLLECTORS, _VOLATILE, State

        name = f"test_{uuid.uuid4().hex}"
        s = State()
        old = {
            "zones": {"public": {}},
            "interfaces": [{"ips": None}],
            "timestamp": "t1",
        }
        new = {
            "zones": {"internal": {}},
            "interfaces": [{"ips": None}],
            "timestamp": "t2",
        }
        _COLLECTORS[name] = lambda: new
        _VOLATILE[name] = frozenset({"interfaces[].ips"})
        s.set(name, old)
        structural, _volatile = s.poll(name)
        assert structural is True
        del _COLLECTORS[name]
        del _VOLATILE[name]

    def test_poll_failure_returns_no_broadcast(self):
        """poll() returns (False, False) and preserves existing data on collector failure."""
        import uuid

        from lib.state import _COLLECTORS, State

        name = f"test_{uuid.uuid4().hex}"
        s = State()
        s.set(name, {"value": 1})
        _COLLECTORS[name] = lambda: (_ for _ in ()).throw(RuntimeError("fail"))
        structural, volatile = s.poll(name)
        assert structural is False
        assert volatile is False
        assert s.get(name) == {"value": 1}
        del _COLLECTORS[name]