"""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)