50 lines
1.5 KiB
Python
50 lines
1.5 KiB
Python
|
"""Parser tests for the dependency-free resolver (no network)."""
|
||
|
|
import socket
|
||
|
|
import struct
|
||
|
|
|
||
|
|
from mdns_tools import _stdlib_resolve as sr
|
||
|
|
|
||
|
|
|
||
|
|
def _a_response(hostname: str, ips: list[str], ttl: int = 120) -> bytes:
|
||
|
|
"""Craft a minimal mDNS response with one question and N A answers."""
|
||
|
|
header = struct.pack(">HHHHHH", 0, 0x8400, 1, len(ips), 0, 0)
|
||
|
|
labels = b""
|
||
|
|
for part in hostname.rstrip(".").split("."):
|
||
|
|
labels += bytes([len(part)]) + part.encode()
|
||
|
|
labels += b"\x00"
|
||
|
|
question = labels + struct.pack(">HH", 1, 1)
|
||
|
|
answers = b""
|
||
|
|
for ip in ips:
|
||
|
|
answers += labels + struct.pack(">HHIH", 1, 1, ttl, 4)
|
||
|
|
answers += socket.inet_aton(ip)
|
||
|
|
return header + question + answers
|
||
|
|
|
||
|
|
|
||
|
|
def test_build_query_roundtrip():
|
||
|
|
q = sr.build_query("host.local", sr.QTYPE_A)
|
||
|
|
# header is 12 bytes; QDCOUNT must be 1
|
||
|
|
assert struct.unpack(">H", q[4:6])[0] == 1
|
||
|
|
assert q.endswith(struct.pack(">HH", 1, 1))
|
||
|
|
|
||
|
|
|
||
|
|
def test_skip_name_plain():
|
||
|
|
data = b"\x04host\x05local\x00rest"
|
||
|
|
off = sr.skip_name(data, 0)
|
||
|
|
assert data[off:off + 4] == b"rest"
|
||
|
|
|
||
|
|
|
||
|
|
def test_parse_all_a_single():
|
||
|
|
resp = _a_response("dell.local", ["192.168.1.17"])
|
||
|
|
recs = sr.parse_all_a(resp)
|
||
|
|
assert recs == [("192.168.1.17", 120)]
|
||
|
|
|
||
|
|
|
||
|
|
def test_parse_all_a_multiple():
|
||
|
|
resp = _a_response("multi.local", ["192.168.1.5", "192.168.122.1"])
|
||
|
|
ips = [ip for ip, _ in sr.parse_all_a(resp)]
|
||
|
|
assert ips == ["192.168.1.5", "192.168.122.1"]
|
||
|
|
|
||
|
|
|
||
|
|
def test_parse_all_a_empty_on_garbage():
|
||
|
|
assert sr.parse_all_a(b"\x00\x00") == []
|