test: unit tests for dns drift/sync + hap-bridge, vendored framework [#769]

Moved from KNEL/PFVCluster tests/unit with repo-relative paths. https://projects.knownelement.com/issues/769#note-4152
2026-09-03 21:45:57 -05:00
parent 33827de0dc
commit da0c3bd868
23 changed files with 1153 additions and 0 deletions
@@ -0,0 +1,84 @@
+#!/bin/bash
+# Unit tests for mdns/hap-bridge.py [#619]
+# Verifies record capture normalization + packet encode/decode round-trip.
+set -euo pipefail
+
+PROJECT_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)"
+SCRIPT="$PROJECT_ROOT/mdns/hap-bridge.py"
+
+TMP="$(mktemp -d)"
+trap 'rm -rf "$TMP"' EXIT
+
+cat > "$TMP/test_bridge.py" <<'PY'
+import importlib.util
+import struct
+import sys
+
+spec = importlib.util.spec_from_file_location("hap_bridge", sys.argv[1])
+hb = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(hb)
+
+failed = 0
+
+def check(name, cond):
+    global failed
+    print(("PASS " if cond else "FAIL ") + name)
+    if not cond:
+        failed += 1
+
+# --- encode_name / read_name round trip
+enc = hb.encode_name("Main Floor._hap._tcp.local")
+back, off = hb.read_name(enc, 0)
+check("name round-trip", back == "Main Floor._hap._tcp.local" and off == len(enc))
+
+# --- encode_record + build_packet round trip for PTR/SRV/TXT
+recs = [
+    {"name": "_hap._tcp.local", "type": 12, "ttl": 10, "ptr_target": "Main Floor._hap._tcp.local"},
+    {"name": "Main Floor._hap._tcp.local", "type": 33, "ttl": 10,
+     "srv_prio": 0, "srv_weight": 0, "srv_port": 58528, "srv_target": "Main-Floor.local"},
+    {"name": "Main Floor._hap._tcp.local", "type": 16, "ttl": 10, "rdata_hex": "0268310568656c6c6f"},
+    {"name": "Main-Floor.local", "type": 1, "ttl": 10, "rdata_hex": "c0a801d4"},
+]
+packet = hb.build_packet(recs)
+check("packet header ancount", struct.unpack(">H", packet[6:8])[0] == 4)
+
+# parse the packet back
+qd = struct.unpack(">H", packet[4:6])[0]
+an = struct.unpack(">H", packet[6:8])[0]
+off = 12
+parsed = []
+for _ in range(qd):
+    _, off = hb.read_name(packet, off)
+    off += 4
+for _ in range(an):
+    name, name_end = hb.read_name(packet, off)
+    rtype, _rc, ttl, rlen = struct.unpack(">HHIH", packet[name_end:name_end + 10])
+    rdata_start = name_end + 10
+    rec = {"name": name, "type": rtype, "ttl": ttl}
+    if rtype == 12:
+        rec["ptr_target"] = hb.read_name(packet, rdata_start)[0]
+    elif rtype == 33:
+        pri, w, port = struct.unpack(">HHH", packet[rdata_start:rdata_start + 6])
+        rec["srv_target"] = hb.read_name(packet, rdata_start + 6)[0]
+        rec["srv_port"] = port
+    else:
+        rec["rdata_hex"] = packet[rdata_start:rdata_start + rlen].hex()
+    parsed.append(rec)
+    off = rdata_start + rlen
+
+check("ptr round-trip", parsed[0]["ptr_target"] == "Main Floor._hap._tcp.local")
+check("srv target round-trip", parsed[1]["srv_target"] == "Main-Floor.local")
+check("srv port round-trip", parsed[1]["srv_port"] == 58528)
+check("txt round-trip", parsed[2]["rdata_hex"] == "0268310568656c6c6f")
+check("a round-trip", parsed[3]["rdata_hex"] == "c0a801d4")
+check("consumed whole packet", off == len(packet))
+
+sys.exit(1 if failed else 0)
+PY
+
+if timeout 60 docker run --rm -v "$SCRIPT:/b.py" -v "$TMP/test_bridge.py:/t.py" python:3.12-alpine python /t.py /b.py; then
+    echo "✅ hap-bridge packet round-trip tests passed"
+else
+    echo "❌ hap-bridge tests failed"
+    exit 1
+fi