link and flake cleanup

tests/test_dns.py

@@ -47,7 +47,13 @@ class TestDNS(unittest.TestCase):
         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")
+        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)
 
@@ -71,7 +77,9 @@ class TestDNS(unittest.TestCase):
                 Host = FakeHost
 
             # start the thread (daemon) that processes the queue
-            t = dns.start_dns_thread(FakeHbd, {"dyndomains": ["example"]}, log=log, email=email)
+            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)
@@ -83,7 +91,9 @@ class TestDNS(unittest.TestCase):
                 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))
+            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
@@ -104,7 +114,9 @@ class TestDNS(unittest.TestCase):
             class FakeHbd:
                 Host = FakeHost
 
-            t = dns.start_dns_thread(FakeHbd, {"dyndomains": ["example"]}, log=log, email=email)
+            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"))
 
@@ -114,12 +126,23 @@ class TestDNS(unittest.TestCase):
                 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))
+            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")
+        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)
 

tests/test_handle_datagram.py

@@ -1,9 +1,11 @@
 from hbd.udp import handle_datagram, parse_message
 from hbd.proto import dicttos
 
+
 class FakeTransport:
     def __init__(self):
         self.sent = []
+
     def sendto(self, data, addr):
         self.sent.append((data, addr))
 
@@ -18,30 +20,30 @@ def test_handle_cmd_sends_command():
     import hbdclass
 
     ctx = {
-        'config': {'watchhosts':[], 'dyndnshosts':[]},
-        'hbdclass': hbdclass,
-        'log': dummy_noop,
-        'email': dummy_noop,
-        'pushmsg': dummy_noop,
-        'msg_to_websockets': dummy_noop,
-        'msgs': [],
-        'DEBUG': 0,
-        'verbose': False,
+        "config": {"watchhosts": [], "dyndnshosts": []},
+        "hbdclass": hbdclass,
+        "log": dummy_noop,
+        "email": dummy_noop,
+        "pushmsg": dummy_noop,
+        "msg_to_websockets": dummy_noop,
+        "msgs": [],
+        "DEBUG": 0,
+        "verbose": False,
     }
 
     # 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'
+    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"
 
     # queue a CMD for the host and send another heartbeat; expect command sent
-    h = hbdclass.Host.hosts['cmdhost']
-    h.cmds.append(('CMD', {'cmd': 'doit'}))
+    h = hbdclass.Host.hosts["cmdhost"]
+    h.cmds.append(("CMD", {"cmd": "doit"}))
     ftr.sent.clear()
-    msg2 = parse_message(dicttos('HTB', {'name':'cmdhost','interval':10}))
-    handle_datagram(msg2, ('127.0.0.1',50000), ftr, ctx)
+    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 cver 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)
+    assert any(b"doit" in s[0] for s in ftr.sent)

tests/test_proto.py

@@ -1,4 +1,3 @@
-import pytest
 from hbd.proto import dicttos, stodict, oldmtodict
 
 

tests/test_udp.py

@@ -3,12 +3,12 @@ from hbd.proto import dicttos
 
 
 def test_parse_message_uncompressed():
-    raw = dicttos('HTB', {'name': 'host', 'interval': 1})
+    raw = dicttos("HTB", {"name": "host", "interval": 1})
     m = parse_message(raw)
-    assert m['ID'].startswith('HTB')
+    assert m["ID"].startswith("HTB")
 
 
 def test_parse_message_compressed():
-    raw = dicttos('ACK', {'time': 1}, compress=True)
+    raw = dicttos("ACK", {"time": 1}, compress=True)
     m = parse_message(raw)
-    assert 'ID' in m
+    assert "ID" in m