158 lines
5.1 KiB
Python
158 lines
5.1 KiB
Python
import glob
|
|
import os
|
|
import pytest
|
|
from hbd.server import configio
|
|
|
|
SAMPLE_YAML = """\
|
|
# Server configuration
|
|
hbd_port: 50004 # HTTP API port
|
|
interval: 20
|
|
users:
|
|
alice:
|
|
full_name: Alice Smith
|
|
admin: true
|
|
notification_channels:
|
|
pushover_ops:
|
|
type: pushover
|
|
token: abc123
|
|
"""
|
|
|
|
|
|
def test_read_roundtrip_loads_values(tmp_path):
|
|
f = tmp_path / ".hb.yaml"
|
|
f.write_text(SAMPLE_YAML)
|
|
data = configio.read_roundtrip(str(f))
|
|
assert data["hbd_port"] == 50004
|
|
assert data["interval"] == 20
|
|
assert data["users"]["alice"]["full_name"] == "Alice Smith"
|
|
|
|
|
|
def test_write_config_creates_backup(tmp_path):
|
|
f = tmp_path / ".hb.yaml"
|
|
f.write_text(SAMPLE_YAML)
|
|
data = configio.read_roundtrip(str(f))
|
|
data["interval"] = 30
|
|
configio.write_config(str(f), data)
|
|
backups = configio.list_backups(str(f))
|
|
assert len(backups) == 1
|
|
assert ".bak." in backups[0]
|
|
|
|
|
|
def test_write_config_preserves_comments(tmp_path):
|
|
f = tmp_path / ".hb.yaml"
|
|
f.write_text(SAMPLE_YAML)
|
|
data = configio.read_roundtrip(str(f))
|
|
data["interval"] = 30
|
|
configio.write_config(str(f), data)
|
|
content = f.read_text()
|
|
assert "# Server configuration" in content
|
|
assert "# HTTP API port" in content
|
|
|
|
|
|
def test_write_config_atomically_replaces_file(tmp_path):
|
|
f = tmp_path / ".hb.yaml"
|
|
f.write_text(SAMPLE_YAML)
|
|
data = configio.read_roundtrip(str(f))
|
|
data["interval"] = 99
|
|
configio.write_config(str(f), data)
|
|
assert not (tmp_path / ".hb.yaml.tmp").exists()
|
|
data2 = configio.read_roundtrip(str(f))
|
|
assert data2["interval"] == 99
|
|
|
|
|
|
def test_write_config_backup_rotation(tmp_path):
|
|
cfg = tmp_path / ".hb.yaml"
|
|
cfg.write_text(SAMPLE_YAML)
|
|
# Pre-create 10 existing backups with old timestamps
|
|
for i in range(10):
|
|
(tmp_path / f".hb.yaml.bak.20260101-{i:06d}").write_text("old")
|
|
data = configio.read_roundtrip(str(cfg))
|
|
configio.write_config(str(cfg), data)
|
|
backups = configio.list_backups(str(cfg))
|
|
assert len(backups) == 10
|
|
assert not (tmp_path / ".hb.yaml.bak.20260101-000000").exists()
|
|
|
|
|
|
def test_list_backups_newest_first(tmp_path):
|
|
cfg = tmp_path / ".hb.yaml"
|
|
cfg.write_text(SAMPLE_YAML)
|
|
for i in range(3):
|
|
(tmp_path / f".hb.yaml.bak.20260101-{i:02d}0000").write_text("b")
|
|
backups = configio.list_backups(str(cfg))
|
|
assert len(backups) == 3
|
|
assert backups == sorted(backups, reverse=True)
|
|
|
|
|
|
def test_apply_structured_section_server_updates_keys(tmp_path):
|
|
f = tmp_path / ".hb.yaml"
|
|
f.write_text(SAMPLE_YAML)
|
|
data = configio.read_roundtrip(str(f))
|
|
configio.apply_structured_section(data, "server", {"interval": 60, "hbd_port": 8080})
|
|
assert data["interval"] == 60
|
|
assert data["hbd_port"] == 8080
|
|
|
|
|
|
def test_apply_structured_section_server_ignores_unknown_keys(tmp_path):
|
|
f = tmp_path / ".hb.yaml"
|
|
f.write_text(SAMPLE_YAML)
|
|
data = configio.read_roundtrip(str(f))
|
|
configio.apply_structured_section(data, "server", {"interval": 60, "not_a_key": "x"})
|
|
assert "not_a_key" not in data
|
|
|
|
|
|
def test_apply_structured_section_users_replaces_dict(tmp_path):
|
|
f = tmp_path / ".hb.yaml"
|
|
f.write_text(SAMPLE_YAML)
|
|
data = configio.read_roundtrip(str(f))
|
|
new_users = {"bob": {"full_name": "Bob Jones", "admin": False}}
|
|
configio.apply_structured_section(data, "users", new_users)
|
|
assert "alice" not in data["users"]
|
|
assert data["users"]["bob"]["full_name"] == "Bob Jones"
|
|
|
|
|
|
def test_apply_yaml_section_notification_channels(tmp_path):
|
|
f = tmp_path / ".hb.yaml"
|
|
f.write_text(SAMPLE_YAML)
|
|
data = configio.read_roundtrip(str(f))
|
|
new_yaml = "email_ops:\n type: email\n recipients: [ops@example.com]\n"
|
|
configio.apply_yaml_section(data, "notification_channels", new_yaml)
|
|
assert "email_ops" in data["notification_channels"]
|
|
assert "pushover_ops" not in data["notification_channels"]
|
|
|
|
|
|
def test_apply_yaml_section_thresholds_maps_to_threshold_configs(tmp_path):
|
|
f = tmp_path / ".hb.yaml"
|
|
f.write_text(SAMPLE_YAML)
|
|
data = configio.read_roundtrip(str(f))
|
|
configio.apply_yaml_section(data, "thresholds", "default:\n cpu: 80\n")
|
|
assert "threshold_configs" in data
|
|
assert data["threshold_configs"]["default"]["cpu"] == 80
|
|
|
|
|
|
def test_apply_yaml_section_dns_replaces_each_key(tmp_path):
|
|
f = tmp_path / ".hb.yaml"
|
|
f.write_text(SAMPLE_YAML)
|
|
data = configio.read_roundtrip(str(f))
|
|
configio.apply_yaml_section(
|
|
data, "dns",
|
|
"nsupdate_bin: /usr/bin/nsupdate\ndyndomains: [dyn.example.com]\n"
|
|
)
|
|
assert data["nsupdate_bin"] == "/usr/bin/nsupdate"
|
|
assert data["dyndomains"] == ["dyn.example.com"]
|
|
|
|
|
|
def test_apply_yaml_section_unknown_raises(tmp_path):
|
|
f = tmp_path / ".hb.yaml"
|
|
f.write_text(SAMPLE_YAML)
|
|
data = configio.read_roundtrip(str(f))
|
|
with pytest.raises(ValueError, match="Unknown YAML section"):
|
|
configio.apply_yaml_section(data, "nope", "x: 1\n")
|
|
|
|
|
|
def test_apply_structured_section_unknown_raises(tmp_path):
|
|
f = tmp_path / ".hb.yaml"
|
|
f.write_text(SAMPLE_YAML)
|
|
data = configio.read_roundtrip(str(f))
|
|
with pytest.raises(ValueError, match="Unknown structured section"):
|
|
configio.apply_structured_section(data, "nope", {"x": 1})
|