From 326f53f23dfa59525a6b6e87652513d9321e0415 Mon Sep 17 00:00:00 2001 From: Andreas Wrede Date: Sat, 9 May 2026 11:11:32 -0400 Subject: [PATCH] feat: add configio module for comment-preserving YAML round-trip writes Co-Authored-By: Claude Sonnet 4.6 --- hbd/server/configio.py | 91 ++++++++++++++++++++++++ pyproject.toml | 1 + tests/test_configio.py | 157 +++++++++++++++++++++++++++++++++++++++++ 3 files changed, 249 insertions(+) create mode 100644 hbd/server/configio.py create mode 100644 tests/test_configio.py diff --git a/hbd/server/configio.py b/hbd/server/configio.py new file mode 100644 index 0000000..d9cf1f4 --- /dev/null +++ b/hbd/server/configio.py @@ -0,0 +1,91 @@ +"""YAML round-trip read/write for .hb.yaml, with backup and atomic writes.""" + +import glob +import os +import threading +from datetime import datetime + +from ruamel.yaml import YAML + +_write_lock = threading.Lock() +_yaml = YAML() +_yaml.preserve_quotes = True + +# Top-level keys managed by the 'server' logical section +_SERVER_KEYS = [ + "hbd_port", "hbd_host", "ws_port", "wss_port", "hb_port", + "interval", "grace", "base_url", "threshold_renotify_interval", + "logfile", "pidfile", "pickfile", "journal_enabled", "journal_dir", + "journal_max_size", "journal_max_backups", "default_owner", +] + +# Top-level keys managed by the 'dns' logical section +_DNS_KEYS = ["nsupdate_bin", "dyndomains", "dyndnshosts", "drophosts"] + + +def read_roundtrip(path: str): + """Load .hb.yaml with ruamel.yaml, preserving comments and ordering.""" + with open(path, "r", encoding="utf-8") as f: + return _yaml.load(f) + + +def write_config(path: str, data) -> None: + """Backup current file then atomically write data. + + Backup naming: {path}.bak.YYYYMMDD-HHMMSS + Rotation: keep the 10 most recent backups, delete older ones. + Atomic write: write to {path}.tmp then os.replace({path}.tmp, path). + Acquires _write_lock for the full backup+write sequence. + """ + with _write_lock: + ts = datetime.now().strftime("%Y%m%d-%H%M%S") + backup_path = f"{path}.bak.{ts}" + if os.path.exists(path): + with open(path, "rb") as src, open(backup_path, "wb") as dst: + dst.write(src.read()) + backups = sorted(glob.glob(f"{path}.bak.*"), reverse=True) + for old in backups[10:]: + os.unlink(old) + tmp = f"{path}.tmp" + with open(tmp, "w", encoding="utf-8") as f: + _yaml.dump(data, f) + os.replace(tmp, path) + + +def list_backups(path: str) -> list: + """Return backup paths sorted newest-first.""" + return sorted(glob.glob(f"{path}.bak.*"), reverse=True) + + +def apply_structured_section(data, section: str, values: dict) -> None: + """Merge a dict of scalar/list values into data for the named logical section. + + For 'server': updates each known key individually, preserving comments on + unchanged keys. For 'users': replaces the entire users dict. + """ + if section == "server": + for key in _SERVER_KEYS: + if key in values: + data[key] = values[key] + elif section == "users": + data["users"] = values + else: + raise ValueError(f"Unknown structured section: {section!r}") + + +def apply_yaml_section(data, section: str, yaml_text: str) -> None: + """Replace the named logical section by parsing yaml_text.""" + parsed = _yaml.load(yaml_text) + if section == "notification_channels": + data["notification_channels"] = parsed + elif section == "thresholds": + data["threshold_configs"] = parsed + elif section == "hosts": + data["hosts"] = parsed + elif section == "dns": + if parsed: + for key in _DNS_KEYS: + if key in parsed: + data[key] = parsed[key] + else: + raise ValueError(f"Unknown YAML section: {section!r}") diff --git a/pyproject.toml b/pyproject.toml index e8ad320..2521610 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -32,6 +32,7 @@ server = [ "aiohttp>=3.11", "Jinja2>=3.1.6", "matrix-nio>=0.24", + "ruamel.yaml>=0.18", ] # Minimal client — hbc_mini only, no external dependencies diff --git a/tests/test_configio.py b/tests/test_configio.py new file mode 100644 index 0000000..e29c2ed --- /dev/null +++ b/tests/test_configio.py @@ -0,0 +1,157 @@ +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})