andreas/heartbeat
Public Access
1
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases 54 Wiki Activity

fix(server): honour the legacy thresholds: config; repair the stale test suite

Browse Source 
Since the built-in defaults were seeded into threshold_configs, the legacy
top-level thresholds: format was silently ignored. get_thresholds_for_host
now uses it whenever it was configured.

Tests: point imports at hbd.server/hbd.common, rewrite the DNS worker,
proto, datagram and threshold tests for the current API, add pytest-asyncio
to the dev extras and set testpaths so a bare pytest runs cleanly.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
andreas
2026-10-28 04:53:43 -04:00
co-authored by Claude Sonnet 5.5
parent 952adb68f8
commit 85c6a4eebe
8 changed files with 78 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_dns.py
+37 -82
View File
@@ -1,9 +1,8 @@
import time
import queue
import asyncio
import unittest
from unittest.mock import patch, MagicMock
import hbd.dns as dns
import hbd.server.dns as dns
class TestDNS(unittest.TestCase):
@@ -17,7 +16,7 @@ class TestDNS(unittest.TestCase):
p = dns.create_nsupdate_payload("host", "2001:db8::1", "example")
self.assertIn("update add host.dy.example 5 AAAA 2001:db8::1", p)
@patch("hbd.dns.Popen")
@patch("hbd.server.dns.Popen")
def test_nsupdate_success(self, mock_popen):
proc = MagicMock()
proc.communicate.return_value = (b"status: NOERROR", None)
@@ -41,7 +40,7 @@ class TestDNS(unittest.TestCase):
stderr=dns.STDOUT,
)
@patch("hbd.dns.Popen")
@patch("hbd.server.dns.Popen")
def test_nsupdate_failure(self, mock_popen):
proc = MagicMock()
proc.communicate.return_value = (b"some error", None)
@@ -57,83 +56,7 @@ class TestDNS(unittest.TestCase):
self.assertIsNotNone(err)
self.assertIn("some error", err)
def test_dnsupdatethread_processes_queue(self):
# patch nsupdate to succeed
with patch("hbd.dns.nsupdate", return_value=None):
logs = []
def log(h, m):
logs.append((h, m))
emails = []
def email(s, m):
emails.append((s, m))
class FakeHost:
dnsQ = queue.Queue()
class FakeHbd:
Host = FakeHost
# start the thread (daemon) that processes the queue
t = dns.start_dns_thread(
FakeHbd, {"dyndomains": ["example"]}, log=log, email=email
)
self.assertTrue(t.is_alive())
# enqueue one item and wait for it to be processed (polling with timeout)
FakeHbd.Host.dnsQ.put(("testhost", "1.2.3.4"))
for _ in range(30):
if logs:
break
time.sleep(0.1)
self.assertTrue(logs, "dnsupdatethread did not call log")
self.assertTrue(
any("changed address" in m or "DNS updated" in m for (_h, m) in logs)
)
def test_dnsupdatethread_calls_email_on_failure(self):
# patch nsupdate to fail with an error message
with patch("hbd.dns.nsupdate", return_value="error: failed"):
logs = []
def log(h, m):
logs.append((h, m))
emails = []
def email(s, m):
emails.append((s, m))
class FakeHost:
dnsQ = queue.Queue()
class FakeHbd:
Host = FakeHost
dns.start_dns_thread(
FakeHbd, {"dyndomains": ["example"]}, log=log, email=email
)
# enqueue and wait for the email to be sent
FakeHbd.Host.dnsQ.put(("testhost", "1.2.3.4"))
for _ in range(30):
if emails:
break
time.sleep(0.1)
self.assertTrue(emails, "dnsupdatethread did not call email on failure")
self.assertTrue(
any(
"nsupdate failed" in s or "nsupdate failed" in m or "error" in m
for (s, m) in emails
)
)
@patch("hbd.dns.Popen")
@patch("hbd.server.dns.Popen")
def test_nsupdate_raises_oserror(self, mock_popen):
mock_popen.side_effect = OSError("noexec")
err = dns.nsupdate(
@@ -147,5 +70,37 @@ class TestDNS(unittest.TestCase):
self.assertIn("execution failed", err)
class TestDNSWorker(unittest.IsolatedAsyncioTestCase):
async def _run_worker(self, nsupdate_result):
"""Feed one update through dns_update_worker; returns the (name, level, message) log calls."""
logs = []
queue = asyncio.Queue()
await queue.put(("testhost", "1.2.3.4"))
await queue.put(None)
with patch("hbd.server.dns.nsupdate", return_value=nsupdate_result):
await dns.dns_update_worker(
None, {"dyndomains": ["example"]}, async_queue=queue,
log=lambda name, level, msg: logs.append((name, level, msg)),
)
return logs
async def test_worker_logs_successful_update(self):
logs = await self._run_worker(None)
self.assertEqual(len(logs), 1)
name, level, msg = logs[0]
self.assertEqual((name, level), ("testhost", "INFO"))
self.assertIn("DNS updated testhost.dy.example", msg)
self.assertIn("1.2.3.4", msg)
async def test_worker_logs_error_on_failed_update(self):
logs = await self._run_worker("error: failed")
self.assertEqual(len(logs), 1)
name, level, msg = logs[0]
self.assertEqual((name, level), ("testhost", "ERROR"))
self.assertIn("DNS update failed", msg)
self.assertIn("error: failed", msg)
if __name__ == "__main__":
unittest.main()
tests/test_handle_datagram.py
+7 -9
View File
@@ -1,5 +1,5 @@
from hbd.udp import handle_datagram, parse_message
from hbd.proto import dicttos
from hbd.server.udp import handle_datagram, parse_message
from hbd.common.proto import dicttos, stodict
class FakeTransport:
@@ -17,7 +17,7 @@ def dummy_noop(*a, **k):
def test_handle_cmd_sends_command():
ftr = FakeTransport()
# prepare ctx linking to hbdclass
import hbdclass
from hbd.server import hbdclass
ctx = {
"config": {"watchhosts": []},
@@ -34,7 +34,7 @@ def test_handle_cmd_sends_command():
# create host by sending initial heartbeat
msg = parse_message(dicttos("HTB", {"name": "cmdhost", "interval": 10}))
handle_datagram(msg, ("127.0.0.1", 50000), ftr, ctx)
assert ftr.sent[0][0] == b"ACK"
assert ftr.sent[0][0].startswith(b"!ACK:")
# queue a CMD for the host and send another heartbeat; expect command sent
h = hbdclass.Host.hosts["cmdhost"]
@@ -42,8 +42,6 @@ def test_handle_cmd_sends_command():
ftr.sent.clear()
msg2 = parse_message(dicttos("HTB", {"name": "cmdhost", "interval": 10}))
handle_datagram(msg2, ("127.0.0.1", 50000), ftr, ctx)
# should have sent ACK and the command; last send should be non-empty
assert len(ftr.sent) >= 1
# the command for ??0 will be sent as raw cmd string
# so at least one send contains b'doit' or similar
assert any(b"doit" in s[0] for s in ftr.sent)
# the ACK and the queued command are both sent; the command is a compressed CMD packet
decoded = [stodict(data) for data, _addr in ftr.sent]
assert any(d["ID"] == "CMD" and d["cmd"] == "doit" for d in decoded)
tests/test_nagios_runner.py
+4 -8
View File
@@ -1,15 +1,11 @@
import asyncio
import logging
import os
import stat
from hbd.client.plugins.nagios_runner import (
NagiosRunnerPlugin,
NAGIOS_OK,
NAGIOS_WARNING,
NAGIOS_CRITICAL,
NAGIOS_UNKNOWN,
)
from hbd.client.plugins.nagios_runner import NagiosRunnerPlugin, NAGIOS_UNKNOWN
# Nagios plugin exit codes (the module itself only needs UNKNOWN)
NAGIOS_OK, NAGIOS_WARNING, NAGIOS_CRITICAL = 0, 1, 2
def test_no_commands_sets_skip_reason():
tests/test_proto.py
+8 -8
View File
@@ -1,19 +1,19 @@
from hbd.proto import dicttos, stodict, oldmtodict
from hbd.common.proto import dicttos, stodict, oldmtodict
def test_dicttos_and_stodict_uncompressed():
def test_dicttos_and_stodict_roundtrip():
msg = dicttos("HTB", {"name": "host.example", "interval": 10})
assert msg.startswith(b"!HTB:")
d = stodict(msg)
assert d["ID"].startswith("HTB")
assert d["ID"] == "HTB"
assert d["name"] == "host.example"
assert d["interval"] == 10
def test_dicttos_and_stodict_compressed():
msg = dicttos("ACK", {"time": 12345}, compress=True)
d = stodict(msg)
# for compressed the original code included the colon in ID slice; ensure no crash
assert "ID" in d
def test_dicttos_and_stodict_ack():
d = stodict(dicttos("ACK", {"time": 12345}))
assert d["ID"] == "ACK"
assert d["time"] == 12345
def test_oldmtodict():
tests/test_threshold.py
+15 -10
View File
@@ -19,7 +19,7 @@ from pathlib import Path
# Add parent directory to path for imports
sys.path.insert(0, str(Path(__file__).parent.parent))
from hbd.threshold import (
from hbd.server.threshold import (
ThresholdChecker,
ThresholdConfig,
AlertLevel,
@@ -257,7 +257,16 @@ def test_threshold_checker_parsing():
print(f" ✓ Parsed {len(checker.thresholds)} thresholds correctly")
def test_check_plugin_data():
def _checker_with_notifications(config, notifications, monkeypatch):
"""ThresholdChecker with no grace period that records notifications instead of sending them."""
monkeypatch.setattr(
ThresholdChecker, "_send_notification",
lambda self, host, lvl, message, *a: notifications.append(f"[{lvl}] {host} {message}"),
)
return ThresholdChecker({"grace": 0, **config})
def test_check_plugin_data(monkeypatch):
"""Test checking plugin data against thresholds."""
print("\nTest 6: Plugin data checking...")
@@ -277,11 +286,7 @@ def test_check_plugin_data():
}
notifications = []
def notification_callback(msg):
notifications.append(msg)
checker = ThresholdChecker(config, notification_callback=notification_callback)
checker = _checker_with_notifications(config, notifications, monkeypatch)
alert_states = {}
# First check - OK
@@ -353,12 +358,12 @@ def test_check_plugin_data():
assert len(state_changes) == 2, "Two metrics should recover"
assert len(notifications) == 2, "Two recovery notifications"
assert any("RECOVERED" in n for n in notifications), "Should have recovery notification"
assert any("RECOVER" in n for n in notifications), "Should have recovery notification"
print(" ✓ Plugin data checking and notifications work")
def test_nested_metrics():
def test_nested_metrics(monkeypatch):
"""Test checking nested metrics like disk partitions."""
print("\nTest 7: Nested metrics (partitions)...")
@@ -384,7 +389,7 @@ def test_nested_metrics():
}
notifications = []
checker = ThresholdChecker(config, notification_callback=lambda m: notifications.append(m))
checker = _checker_with_notifications(config, notifications, monkeypatch)
alert_states = {}
plugin_data = {
tests/test_udp.py
+2 -2
View File
@@ -1,5 +1,5 @@
from hbd.udp import parse_message
from hbd.proto import dicttos
from hbd.server.udp import parse_message
from hbd.common.proto import dicttos
def test_parse_message_uncompressed():