andreasandClaude Sonnet 5.5
85c6a4eebe
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>
107 lines
3.4 KiB
Python
107 lines
3.4 KiB
Python
import asyncio
|
|
import unittest
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
import hbd.server.dns as dns
|
|
|
|
|
|
class TestDNS(unittest.TestCase):
|
|
|
|
def test_create_nsupdate_payload_ipv4(self):
|
|
p = dns.create_nsupdate_payload("host", "1.2.3.4", "example")
|
|
self.assertIn("update add host.dy.example 5 A 1.2.3.4", p)
|
|
self.assertNotIn("AAAA", p)
|
|
|
|
def test_create_nsupdate_payload_ipv6(self):
|
|
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.server.dns.Popen")
|
|
def test_nsupdate_success(self, mock_popen):
|
|
proc = MagicMock()
|
|
proc.communicate.return_value = (b"status: NOERROR", None)
|
|
mock_popen.return_value = proc
|
|
|
|
err = dns.nsupdate(
|
|
"host",
|
|
"1.2.3.4",
|
|
"example",
|
|
nsupdate_bin="/usr/bin/nsupdate",
|
|
rndc_key="/etc/rndc.key",
|
|
)
|
|
|
|
self.assertIsNone(err)
|
|
mock_popen.assert_called_once_with(
|
|
["/usr/bin/nsupdate", "-k", "/etc/rndc.key", "-v"],
|
|
shell=False,
|
|
bufsize=0,
|
|
stdin=dns.PIPE,
|
|
stdout=dns.PIPE,
|
|
stderr=dns.STDOUT,
|
|
)
|
|
|
|
@patch("hbd.server.dns.Popen")
|
|
def test_nsupdate_failure(self, mock_popen):
|
|
proc = MagicMock()
|
|
proc.communicate.return_value = (b"some error", None)
|
|
mock_popen.return_value = proc
|
|
|
|
err = dns.nsupdate(
|
|
"host",
|
|
"1.2.3.4",
|
|
"example",
|
|
nsupdate_bin="/usr/bin/nsupdate",
|
|
rndc_key="/etc/rndc.key",
|
|
)
|
|
self.assertIsNotNone(err)
|
|
self.assertIn("some error", err)
|
|
|
|
@patch("hbd.server.dns.Popen")
|
|
def test_nsupdate_raises_oserror(self, mock_popen):
|
|
mock_popen.side_effect = OSError("noexec")
|
|
err = dns.nsupdate(
|
|
"h",
|
|
"1.2.3.4",
|
|
"example",
|
|
nsupdate_bin="/usr/bin/nsupdate",
|
|
rndc_key="/etc/rndc.key",
|
|
)
|
|
self.assertIsNotNone(err)
|
|
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()
|