mrcharles
babd6d2e25
Live proof: ecobee answers unicast mDNS with PTR only (_hap._tcp.local -> "Main Floor._hap._tcp.local"), so the old replace-on-probe wiped learned records every cycle. Merge by (name, type) instead. CONFIG_PATH now env-overridable for tests. Details: https://projects.knownelement.com/issues/619#note-5 💘 Generated with Crush Assisted-by: Crush:glm-5.2
63 lines
2.4 KiB
Python
63 lines
2.4 KiB
Python
#!/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()
|