refactor

tests/test_dns.py

@@ -0,0 +1,128 @@
+import time
+import queue
+import unittest
+from unittest.mock import patch, MagicMock
+
+import hbd.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.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.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)
+
+    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
+
+            t = 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")
+    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)
+
+
+if __name__ == "__main__":
+    unittest.main()