heartbeat/tests/test_configio.py

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


def test_read_roundtrip_missing_file_raises(tmp_path):
    with pytest.raises(FileNotFoundError):
        configio.read_roundtrip(str(tmp_path / "nonexistent.yaml"))