Alexander Whitestone
ade82f2e67
28 tests covering: - YAML loading (valid, empty, invalid, non-dict) - Nested flattening (dot paths, lists, deep nesting) - Config diffing (changed, missing, extra, ignored keys, secrets) - Sync patch generation (changed, missing, non-destructive) - Markdown and report formatting Closes #686
229 lines
8.8 KiB
Python
229 lines
8.8 KiB
Python
"""Tests for config drift detection."""
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
SCRIPT_DIR = Path(__file__).resolve().parent.parent / "scripts"
|
|
sys.path.insert(0, str(SCRIPT_DIR))
|
|
|
|
from config_drift_detector import (
|
|
load_yaml,
|
|
_flatten,
|
|
diff_configs,
|
|
generate_sync_patch,
|
|
format_drift_markdown,
|
|
format_report,
|
|
)
|
|
|
|
|
|
# ── YAML Loading ─────────────────────────────────────────────────────────────
|
|
|
|
class TestLoadYaml:
|
|
def test_valid_yaml(self):
|
|
result = load_yaml("model:\n default: hermes\n provider: nous")
|
|
assert result["model"]["default"] == "hermes"
|
|
|
|
def test_empty_string(self):
|
|
assert load_yaml("") == {}
|
|
|
|
def test_invalid_yaml(self):
|
|
assert load_yaml("{invalid yaml ][") == {}
|
|
|
|
def test_non_dict_root(self):
|
|
assert load_yaml("- item1\n- item2") == {}
|
|
|
|
|
|
# ── Flatten ──────────────────────────────────────────────────────────────────
|
|
|
|
class TestFlatten:
|
|
def test_flat_dict(self):
|
|
assert _flatten({"a": 1, "b": 2}) == {"a": 1, "b": 2}
|
|
|
|
def test_nested_dict(self):
|
|
result = _flatten({"model": {"default": "hermes", "provider": "nous"}})
|
|
assert result["model.default"] == "hermes"
|
|
assert result["model.provider"] == "nous"
|
|
|
|
def test_deeply_nested(self):
|
|
result = _flatten({"a": {"b": {"c": 42}}})
|
|
assert result["a.b.c"] == 42
|
|
|
|
def test_list_serialized_as_json(self):
|
|
result = _flatten({"toolsets": ["all", "terminal"]})
|
|
assert result["toolsets"] == '["all", "terminal"]'
|
|
|
|
def test_empty_dict(self):
|
|
assert _flatten({}) == {}
|
|
|
|
|
|
# ── Diff Configs ─────────────────────────────────────────────────────────────
|
|
|
|
class TestDiffConfigs:
|
|
def test_identical_configs(self):
|
|
cfg = {"model": {"default": "hermes"}, "agent": {"max_turns": 30}}
|
|
assert diff_configs(cfg, cfg) == []
|
|
|
|
def test_changed_value(self):
|
|
canon = {"model": {"default": "hermes"}}
|
|
remote = {"model": {"default": "gpt-4"}}
|
|
drifts = diff_configs(canon, remote)
|
|
assert len(drifts) == 1
|
|
assert drifts[0]["key"] == "model.default"
|
|
assert drifts[0]["drift_type"] == "changed"
|
|
assert drifts[0]["canonical_value"] == "hermes"
|
|
assert drifts[0]["remote_value"] == "gpt-4"
|
|
|
|
def test_missing_remote_key(self):
|
|
canon = {"model": {"default": "hermes"}, "agent": {"max_turns": 30}}
|
|
remote = {"model": {"default": "hermes"}}
|
|
drifts = diff_configs(canon, remote)
|
|
assert len(drifts) == 1
|
|
assert drifts[0]["key"] == "agent.max_turns"
|
|
assert drifts[0]["drift_type"] == "missing_remote"
|
|
|
|
def test_extra_remote_key(self):
|
|
canon = {"model": {"default": "hermes"}}
|
|
remote = {"model": {"default": "hermes"}, "extra": {"key": 1}}
|
|
drifts = diff_configs(canon, remote)
|
|
assert len(drifts) == 1
|
|
assert drifts[0]["drift_type"] == "extra_remote"
|
|
|
|
def test_ignored_keys(self):
|
|
canon = {"terminal": {"cwd": "/a"}, "model": {"default": "hermes"}}
|
|
remote = {"terminal": {"cwd": "/b"}, "model": {"default": "hermes"}}
|
|
# terminal.cwd is in the default ignore list
|
|
drifts = diff_configs(canon, remote)
|
|
assert len(drifts) == 0
|
|
|
|
def test_api_keys_ignored(self):
|
|
canon = {"auxiliary": {"vision": {"api_key": "sk-old"}}}
|
|
remote = {"auxiliary": {"vision": {"api_key": "sk-new"}}}
|
|
drifts = diff_configs(canon, remote)
|
|
assert len(drifts) == 0
|
|
|
|
def test_custom_ignored_keys(self):
|
|
canon = {"a": 1, "b": 2}
|
|
remote = {"a": 99, "b": 99}
|
|
drifts = diff_configs(canon, remote, ignored_keys=["a"])
|
|
assert len(drifts) == 1
|
|
assert drifts[0]["key"] == "b"
|
|
|
|
def test_multiple_drifts(self):
|
|
canon = {"model": {"default": "hermes", "provider": "nous"}, "agent": {"max_turns": 30}}
|
|
remote = {"model": {"default": "gpt-4", "provider": "openai"}, "agent": {"max_turns": 30}}
|
|
drifts = diff_configs(canon, remote)
|
|
assert len(drifts) == 2
|
|
keys = {d["key"] for d in drifts}
|
|
assert "model.default" in keys
|
|
assert "model.provider" in keys
|
|
|
|
def test_list_values_compared_by_content(self):
|
|
canon = {"toolsets": ["all"]}
|
|
remote = {"toolsets": ["terminal", "file"]}
|
|
drifts = diff_configs(canon, remote)
|
|
assert len(drifts) == 1
|
|
assert drifts[0]["drift_type"] == "changed"
|
|
|
|
|
|
# ── Sync Patch ───────────────────────────────────────────────────────────────
|
|
|
|
class TestGenerateSyncPatch:
|
|
def test_empty_when_identical(self):
|
|
cfg = {"model": {"default": "hermes"}}
|
|
assert generate_sync_patch(cfg, cfg) == {}
|
|
|
|
def test_patches_changed_values(self):
|
|
canon = {"model": {"default": "hermes"}}
|
|
remote = {"model": {"default": "gpt-4"}}
|
|
patch = generate_sync_patch(canon, remote)
|
|
assert patch["model.default"] == "hermes"
|
|
|
|
def test_patches_missing_keys(self):
|
|
canon = {"model": {"default": "hermes"}, "agent": {"max_turns": 30}}
|
|
remote = {"model": {"default": "hermes"}}
|
|
patch = generate_sync_patch(canon, remote)
|
|
assert patch["agent.max_turns"] == 30
|
|
|
|
def test_does_not_delete_extra_keys(self):
|
|
canon = {"model": {"default": "hermes"}}
|
|
remote = {"model": {"default": "hermes"}, "extra": 1}
|
|
patch = generate_sync_patch(canon, remote)
|
|
assert "extra" not in patch
|
|
|
|
|
|
# ── Formatting ───────────────────────────────────────────────────────────────
|
|
|
|
class TestFormatting:
|
|
def test_drift_markdown_no_drift(self):
|
|
text = format_drift_markdown([])
|
|
assert "no drift" in text.lower()
|
|
|
|
def test_drift_markdown_with_drifts(self):
|
|
drifts = [
|
|
{"key": "model.default", "canonical_value": "hermes", "remote_value": "gpt-4", "drift_type": "changed"},
|
|
]
|
|
text = format_drift_markdown(drifts)
|
|
assert "model.default" in text
|
|
assert "hermes" in text
|
|
assert "gpt-4" in text
|
|
assert "changed" in text
|
|
|
|
def test_drift_markdown_caps_at_30(self):
|
|
drifts = [
|
|
{"key": f"key.{i}", "canonical_value": "a", "remote_value": "b", "drift_type": "changed"}
|
|
for i in range(50)
|
|
]
|
|
text = format_drift_markdown(drifts)
|
|
assert "and 20 more" in text
|
|
|
|
def test_report_format_has_sections(self):
|
|
report = {
|
|
"timestamp": "2026-04-15T12:00:00+00:00",
|
|
"canonical_config": "config.yaml",
|
|
"total_nodes": 4,
|
|
"reachable": 3,
|
|
"unreachable": 1,
|
|
"drifted_nodes": 1,
|
|
"clean_nodes": 2,
|
|
"total_drifts": 3,
|
|
"nodes": {
|
|
"allegro": {
|
|
"host": "167.99.126.228",
|
|
"reachable": True,
|
|
"machine_type": "vps",
|
|
"config_keys": 25,
|
|
"has_drift": True,
|
|
"drift_count": 3,
|
|
"drifts": [
|
|
{"key": "model.default", "canonical_value": "hermes", "remote_value": "gpt-4", "drift_type": "changed"},
|
|
{"key": "agent.max_turns", "canonical_value": 30, "remote_value": 50, "drift_type": "changed"},
|
|
{"key": "terminal.timeout", "canonical_value": 180, "remote_value": None, "drift_type": "missing_remote"},
|
|
],
|
|
},
|
|
"timmy": {
|
|
"host": "localhost",
|
|
"reachable": True,
|
|
"machine_type": "mac",
|
|
"config_keys": 30,
|
|
"has_drift": False,
|
|
"drift_count": 0,
|
|
"drifts": [],
|
|
},
|
|
"ezra": {
|
|
"host": "143.198.27.163",
|
|
"reachable": False,
|
|
"drift_count": -1,
|
|
"error": "Could not fetch config",
|
|
"drifts": [],
|
|
},
|
|
},
|
|
}
|
|
text = format_report(report)
|
|
assert "CONFIG DRIFT DETECTION" in text
|
|
assert "allegro" in text.lower()
|
|
assert "UNREACHABLE" in text
|
|
assert "model.default" in text
|
|
assert "no drift" in text.lower() # timmy
|