152 lines
4.4 KiB
Python
152 lines
4.4 KiB
Python
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
|
|
|
|
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()
|