"""Tests for lib.common apply-metadata and diff helpers.""" from __future__ import annotations from lib.common import ( _APPLY_HASH_KEY, _LAST_APPLIED_CONFIG_KEY, 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 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)