chore: initial import from KNEL/PFVCluster@041d311 [#769]

Split per O&M lane work order. Full history: KNEL/PFVCluster. https://projects.knownelement.com/issues/769#note-4152
2026-09-03 21:30:32 -05:00
commit 33827de0dc
236 changed files with 9587 additions and 0 deletions
@@ -0,0 +1,62 @@
+#!/usr/bin/env python3
+"""Unit tests for hap-bridge record handling. [#619]
+
+Run: python3 -m unittest discover -s netinfra/mdns -p 'test_*.py'
+"""
+
+import importlib.util
+import os
+import unittest
+
+_SPEC = importlib.util.spec_from_file_location(
+    "hap_bridge", os.path.join(os.path.dirname(__file__), "hap-bridge.py")
+)
+hap_bridge = importlib.util.module_from_spec(_SPEC)
+_SPEC.loader.exec_module(hap_bridge)
+
+PTR = {"name": "_hap._tcp.local", "type": 12, "ttl": 4500, "ptr_target": "dev1._hap._tcp.local"}
+SRV_OLD = {"name": "dev1._hap._tcp.local", "type": 33, "ttl": 120, "srv_prio": 0,
+           "srv_weight": 0, "srv_port": 8080, "srv_target": "dev1.local"}
+SRV_NEW = {"name": "dev1._hap._tcp.local", "type": 33, "ttl": 120, "srv_prio": 0,
+           "srv_weight": 0, "srv_port": 8081, "srv_target": "dev1.local"}
+TXT = {"name": "dev1._hap._tcp.local", "type": 16, "ttl": 4500, "rdata_hex": "0141"}
+
+
+class TestMergeRecords(unittest.TestCase):
+    def keyset(self, recs):
+        return {(r["name"], r["type"]): r for r in recs}
+
+    def test_empty_prev_starts_fresh(self):
+        merged = hap_bridge.merge_records([], [PTR, SRV_OLD, TXT])
+        self.assertEqual(len(merged), 3)
+        self.assertEqual(self.keyset(merged)[(PTR["name"], 12)], PTR)
+
+    def test_same_key_replaces(self):
+        merged = hap_bridge.merge_records([PTR, SRV_OLD], [SRV_NEW])
+        by_key = self.keyset(merged)
+        self.assertEqual(by_key[(SRV_NEW["name"], 33)]["srv_port"], 8081)
+        self.assertEqual(by_key[(PTR["name"], 12)], PTR)
+
+    def test_new_keys_union_in(self):
+        merged = hap_bridge.merge_records([PTR], [SRV_NEW, TXT])
+        self.assertEqual(len(merged), 3)
+
+    def test_empty_new_keeps_prev(self):
+        merged = hap_bridge.merge_records([PTR, SRV_OLD], [])
+        self.assertEqual(len(merged), 2)
+        self.assertEqual(self.keyset(merged)[(SRV_OLD["name"], 33)]["srv_port"], 8080)
+
+
+class TestWireRoundTrip(unittest.TestCase):
+    def test_ptr_record_encodes_without_compression(self):
+        wire = hap_bridge.encode_record(PTR)
+        name, off = hap_bridge.read_name(wire, 0)
+        self.assertEqual(name, "_hap._tcp.local")
+        rtype, rclass, ttl, rlen = hap_bridge.decode_rr_header(wire, off)
+        self.assertEqual((rtype, rclass, rlen), (12, 1, len(hap_bridge.encode_name(PTR["ptr_target"]))))
+        self.assertEqual(ttl, 4500)
+        self.assertNotIn(b"\xc0", wire)  # no compression pointers in replay
+
+
+if __name__ == "__main__":
+    unittest.main()