vacuum-wall/tests/test_common.py

"""Tests for lib.common apply-metadata and diff helpers."""

from __future__ import annotations

from lib.common import (
    _APPLY_HASH_KEY,
    _LAST_APPLIED_CONFIG_KEY,
    compute_pending,
    config_hash,
    deep_diff,
    load_json,
    revert_to_applied,
    save_json,
    stamp_applied,
    strip_apply_meta,
)


class TestStripApplyMeta:
    def test_strips_both_keys(self):
        cfg = {"a": 1, _APPLY_HASH_KEY: "h", _LAST_APPLIED_CONFIG_KEY: {}}
        assert strip_apply_meta(cfg) == {"a": 1}

    def test_missing_keys(self):
        assert strip_apply_meta({"a": 1}) == {"a": 1}

    def test_does_not_mutate_input(self):
        cfg = {"a": 1, _APPLY_HASH_KEY: "h"}
        strip_apply_meta(cfg)
        assert _APPLY_HASH_KEY in cfg


class TestConfigHashIgnoresMeta:
    def test_hash_unaffected_by_metadata(self):
        cfg = {"a": 1}
        stamped = {"a": 1, _APPLY_HASH_KEY: "x", _LAST_APPLIED_CONFIG_KEY: {"a": 1}}
        assert config_hash(cfg) == config_hash(stamped)


class TestStampApplied:
    def test_records_snapshot_and_hash(self):
        cfg = {"a": 1}
        stamp_applied(cfg)
        assert cfg[_LAST_APPLIED_CONFIG_KEY] == {"a": 1}
        assert cfg[_APPLY_HASH_KEY] == config_hash(cfg)

    def test_stable(self):
        cfg = {"a": 1}
        stamp_applied(cfg)
        # A pending-style check: hash matches the current (stripped) config.
        assert _APPLY_HASH_KEY in cfg and cfg[_APPLY_HASH_KEY] == config_hash(cfg)
        # No drift → no diff.
        assert (
            deep_diff(cfg.get(_LAST_APPLIED_CONFIG_KEY, {}), strip_apply_meta(cfg))
            == []
        )


class TestDeepDiff:
    def test_identical_empty(self):
        assert deep_diff({"a": 1, _APPLY_HASH_KEY: "h"}, {"a": 1}) == []

    def test_changed_scalar(self):
        diff = deep_diff({"a": 1}, {"a": 2})
        assert diff == [{"path": "a", "action": "changed", "old": 1, "new": 2}]

    def test_added_removed(self):
        added = deep_diff({}, {"a": 1})
        assert added[0]["action"] == "added" and added[0]["new"] == 1
        removed = deep_diff({"a": 1}, {})
        assert removed[0]["action"] == "removed" and removed[0]["old"] == 1

    def test_nested_and_list_index(self):
        old = {"z": {"svc": ["http"], "ranges": [{"ip": "10.0.0.1", "n": 1}]}}
        new = {"z": {"svc": ["http", "ssh"], "ranges": [{"ip": "10.0.0.2", "n": 1}]}}
        paths = {d["path"] for d in deep_diff(old, new)}
        assert "z.svc" in paths
        assert "z.ranges[0].ip" in paths
        assert not any(p.startswith("z.ranges[0].n") for p in paths)


class TestComputePending:
    def test_hash_match_no_pending(self):
        cfg = {"a": 1}
        stamp_applied(cfg)
        pending, diff = compute_pending(cfg)
        assert pending is False
        assert diff == []

    def test_never_applied_pending_no_snapshot(self):
        pending, diff = compute_pending({"a": 1})
        assert pending is True
        assert diff == []

    def test_never_applied_pending_with_foreign_snapshot(self):
        # A recorded snapshot that does not match the current hash is still
        # used for the diff.
        cfg = {"a": 2, _LAST_APPLIED_CONFIG_KEY: {"a": 1}}
        pending, diff = compute_pending(cfg)
        assert pending is True
        assert diff == [{"path": "a", "action": "changed", "old": 1, "new": 2}]

    def test_hash_mismatch_with_snapshot_diffs(self):
        applied = {"zones": {"lan": {"services": ["http"]}}}
        stamped = dict(applied)
        stamp_applied(stamped)
        drifted = {"zones": {"lan": {"services": ["http", "ssh"]}}}
        drifted[_LAST_APPLIED_CONFIG_KEY] = applied
        drifted[_APPLY_HASH_KEY] = stamped[_APPLY_HASH_KEY]
        pending, diff = compute_pending(drifted)
        assert pending is True
        paths = {d["path"] for d in diff}
        assert "zones.lan.services" in paths

    def test_hash_mismatch_snapshot_not_dict(self):
        cfg = {"a": 1, _LAST_APPLIED_CONFIG_KEY: "not-a-dict"}
        pending, diff = compute_pending(cfg)
        assert pending is True
        assert diff == []

    def test_meta_keys_excluded_from_diff(self):
        cfg = {"a": 1}
        stamp_applied(cfg)
        cfg["a"] = 2  # drift
        pending, diff = compute_pending(cfg)
        assert pending is True
        assert not any(
            p.startswith(("_last_applied",)) for d in diff for p in [d["path"]]
        )


class TestDashboardFallback:
    def test_hash_subsystem_unchanged_generic(self):
        # Guards that a pending status without a snapshot still yields a
        # renderable pending flag (frontend falls back to a generic line).
        status = {"pending_changes": True, "pending_diff": []}
        assert status["pending_changes"] is True
        assert status["pending_diff"] == []


class TestRevertToApplied:
    def test_restores_snapshot_and_clears_pending(self, tmp_path):
        path = tmp_path / "config.json"
        applied = {"zones": {"lan": {"services": ["http"]}}}
        stamped = dict(applied)
        stamp_applied(stamped)
        # Drift the file after apply (the "pending" state).
        dirty = {"zones": {"lan": {"services": ["http", "ssh"]}}}
        dirty[_LAST_APPLIED_CONFIG_KEY] = dict(applied)
        dirty[_APPLY_HASH_KEY] = stamped[_APPLY_HASH_KEY]
        save_json(path, dirty, indent=2)
        # Sanity: pending check would report drift.
        assert dirty[_APPLY_HASH_KEY] != config_hash(dirty)

        ok, reason = revert_to_applied(path)
        assert ok and reason == ""

        restored = load_json(path)
        assert _APPLY_HASH_KEY in restored and restored[_APPLY_HASH_KEY] == config_hash(
            restored
        )
        assert (
            _LAST_APPLIED_CONFIG_KEY in restored
            and restored[_LAST_APPLIED_CONFIG_KEY] == applied
        )
        assert (
            deep_diff(
                restored.get(_LAST_APPLIED_CONFIG_KEY, {}), strip_apply_meta(restored)
            )
            == []
        )

    def test_no_baseline(self, tmp_path):
        path = tmp_path / "config.json"
        save_json(path, {"zones": {}}, indent=2)
        ok, reason = revert_to_applied(path)
        assert not ok
        assert reason
        # File untouched.
        assert _LAST_APPLIED_CONFIG_KEY not in load_json(path)

    def test_missing_file(self, tmp_path):
        ok, reason = revert_to_applied(tmp_path / "nope.json")
        assert not ok
        assert reason

    def test_stale_hash_but_snapshot_present(self, tmp_path):
        # Baseline recorded, hash stale (drifted) → still revertable.
        path = tmp_path / "config.json"
        applied = {"a": 1}
        stamped = dict(applied)
        stamp_applied(stamped)
        stamp = dict(stamped)
        stamp["a"] = 99  # edited without re-stamping
        save_json(path, stamp, indent=2)
        assert stamp[_APPLY_HASH_KEY] != config_hash(stamp)

        ok, _ = revert_to_applied(path)
        assert ok
        restored = load_json(path)
        assert restored[_APPLY_HASH_KEY] == config_hash(restored)