tests/test_mdns.py
Rama master · réplica de lectura del repositorio original.
import sys
import unittest
from unittest.mock import MagicMock, patch
from peach.mdns import (
DnsSdPublisher, MdnsPublisher, answers_on_network, create_mdns_publisher, lan_ipv4,
)
class MdnsPublisherTests(unittest.TestCase):
def test_registers_and_unregisters_peach_local_on_all_interfaces(self):
with patch("peach.mdns.Zeroconf") as zeroconf_type, \
patch("peach.mdns.answers_on_network", return_value=True):
publisher = MdnsPublisher(
"Peach.local", 80, address_resolver=lambda: "192.0.2.10"
)
publisher.start()
info = zeroconf_type.return_value.register_service.call_args.args[0]
self.assertEqual(info.server, "peach.local.")
self.assertEqual(info.port, 80)
self.assertEqual(publisher.status, "peach.local")
self.assertEqual(publisher.backend, "zeroconf-all-interfaces")
publisher.stop()
zeroconf_type.assert_called_once_with()
zeroconf_type.return_value.register_service.assert_called_once_with(
info, allow_name_change=True,
)
zeroconf_type.return_value.unregister_service.assert_called_once_with(info)
zeroconf_type.return_value.close.assert_called_once()
def test_address_change_republishes_so_dhcp_moves_are_followed(self):
"""换 Wi-Fi、DHCP 换地址都只改地址、不通知进程。
不复查的话 `peach.local` 会一直指向旧地址；zeroconf 的套接字也绑在旧接口上，
所以是整体重建而不是只换地址字段。
"""
addresses = iter(["192.0.2.10", "192.0.2.10", "198.51.100.7"])
current = ["192.0.2.10"]
def resolver():
current[0] = next(addresses, current[0])
return current[0]
with patch("peach.mdns.Zeroconf") as zeroconf_type, \
patch("peach.mdns.answers_on_network", return_value=True):
publisher = MdnsPublisher(
"peach", 80, address_resolver=resolver, refresh_seconds=0,
)
publisher.start()
self.assertEqual(publisher.address, "192.0.2.10")
self.assertEqual(publisher.refresh(), "unchanged")
self.assertEqual(publisher.refresh(), "republished")
self.assertEqual(publisher.address, "198.51.100.7")
publisher.stop()
self.assertEqual(zeroconf_type.call_count, 2)
def test_losing_the_network_withdraws_the_record_instead_of_keeping_a_stale_one(self):
"""留着一条指向旧地址的记录比没有记录更糟：客户端会一直连一个不存在的地方。"""
states = iter(["192.0.2.10"])
def resolver():
try:
return next(states)
except StopIteration:
raise RuntimeError("no publishable LAN IPv4 address") from None
with patch("peach.mdns.Zeroconf"), \
patch("peach.mdns.answers_on_network", return_value=True):
publisher = MdnsPublisher(
"peach", 80, address_resolver=resolver, refresh_seconds=0,
)
publisher.start()
self.assertEqual(publisher.status, "peach.local")
self.assertEqual(publisher.refresh(), "unavailable")
self.assertEqual(publisher.status, "unavailable")
self.assertIsNone(publisher.address)
def test_a_pinned_address_never_republishes(self):
"""生产上显式钉住地址时复查必须是空转，不能自己换成别的网卡。"""
with patch("peach.mdns.Zeroconf") as zeroconf_type, \
patch("peach.mdns.answers_on_network", return_value=True):
publisher = MdnsPublisher(
"peach", 80, address_resolver=lambda: "192.0.2.10", refresh_seconds=0,
)
publisher.start()
self.assertEqual(publisher.refresh(), "unchanged")
self.assertEqual(publisher.refresh(), "unchanged")
publisher.stop()
self.assertEqual(zeroconf_type.call_count, 1)
def test_secure_publication_uses_https_service(self):
with patch("peach.mdns.Zeroconf") as zeroconf_type, \
patch("peach.mdns.answers_on_network", return_value=True):
publisher = MdnsPublisher(
"peach", 443, secure=True, address_resolver=lambda: "192.0.2.10"
)
publisher.start()
info = zeroconf_type.return_value.register_service.call_args.args[0]
self.assertEqual(info.type, "_https._tcp.local.")
self.assertEqual(info.port, 443)
publisher.stop()
def test_rejects_invalid_hostname(self):
with self.assertRaises(ValueError):
MdnsPublisher("bad name", 80)
def test_factory_picks_the_backend_that_works_on_this_platform(self):
"""macOS 走系统 mDNSResponder：自带 zeroconf 在 launchd 起的进程里会被
「本地网络」隐私门静默拒绝，注册看着成功、一个多播包都发不出去。"""
publisher = create_mdns_publisher("peach", 80)
expected = DnsSdPublisher if sys.platform == "darwin" else MdnsPublisher
self.assertIsInstance(publisher, expected)
def test_explicit_address_avoids_tunnel_route(self):
with patch("peach.mdns.Zeroconf") as zeroconf_type, \
patch("peach.mdns.answers_on_network", return_value=True):
publisher = MdnsPublisher(
"peach", 80, address_resolver=lambda: "198.51.100.162", refresh_seconds=0,
)
publisher.start()
self.assertEqual(publisher.address, "198.51.100.162")
publisher.stop()
zeroconf_type.assert_called_once_with()
def test_tunnel_route_falls_back_to_real_host_interface(self):
fake_socket = MagicMock()
fake_socket.getsockname.return_value = ("198.18.0.1", 50000)
with patch("peach.mdns.socket.socket", return_value=fake_socket), patch(
"peach.mdns.socket.gethostname", return_value="host",
), patch(
# 前两个是虚拟网卡那种网关端点（末段 1 或 2），`lan_ipv4` 按端点评分把它们
# 排在真实接口后面，所以第三个胜出。
"peach.mdns.socket.gethostbyname_ex",
return_value=("host", [], ["192.0.2.2", "203.0.113.1", "198.51.100.162"]),
):
self.assertEqual(lan_ipv4(), "198.51.100.162")
class ReachabilityTests(unittest.TestCase):
"""可达性不能由服务进程自证。
macOS 的「本地网络」隐私门按进程判：launchd 起的作业自己发的多播会被静默丢弃，
所以它去探自己必然收不到回应，会把好的记录判成不可达。运行时因此只报「注册还在
生效」，真正的可达性验证交给 `scripts/check_mdns.py`——那个从终端跑，有权限。
"""
def test_runtime_status_does_not_depend_on_a_self_probe(self):
with patch("peach.mdns.Zeroconf"), \
patch("peach.mdns.answers_on_network", return_value=False) as probe:
publisher = MdnsPublisher(
"peach", 80, address_resolver=lambda: "192.0.2.10", refresh_seconds=0)
publisher.start()
self.assertEqual(publisher.status, "peach.local")
probe.assert_not_called()
publisher.stop()
def test_probe_never_raises_on_a_dead_socket(self):
with patch("peach.mdns.socket.socket", side_effect=OSError("no multicast")):
self.assertFalse(answers_on_network("peach.local", timeout=0.1))
if __name__ == "__main__":
unittest.main()