"""Unit tests for zeroconf._services.browser.""" from __future__ import annotations import asyncio import logging import os import socket import time import unittest from collections.abc import Iterable from threading import Event from typing import cast from unittest.mock import patch import pytest import zeroconf as r import zeroconf._services.browser as _services_browser from zeroconf import ( DNSPointer, DNSQuestion, Zeroconf, _engine, const, current_time_millis, millis_to_seconds, ) from zeroconf._services import ServiceStateChange from zeroconf._services.browser import ServiceBrowser, _ScheduledPTRQuery from zeroconf._services.info import ServiceInfo from zeroconf.asyncio import AsyncServiceBrowser, AsyncZeroconf from .. import ( QuestionHistoryWithoutSuppression, _inject_response, _wait_for_start, has_working_ipv6, time_changed_millis, ) log = logging.getLogger("zeroconf") original_logging_level = logging.NOTSET def setup_module(): global original_logging_level original_logging_level = log.level log.setLevel(logging.DEBUG) def teardown_module(): if original_logging_level != logging.NOTSET: log.setLevel(original_logging_level) def mock_incoming_msg(records: Iterable[r.DNSRecord]) -> r.DNSIncoming: generated = r.DNSOutgoing(const._FLAGS_QR_RESPONSE) for record in records: generated.add_answer_at_time(record, 0) return r.DNSIncoming(generated.packets()[0]) def test_service_browser_cancel_multiple_times(): """Test we can cancel a ServiceBrowser multiple times before close.""" # instantiate a zeroconf instance zc = Zeroconf(interfaces=["127.0.0.1"]) # start a browser type_ = "_hap._tcp.local." class MyServiceListener(r.ServiceListener): pass listener = MyServiceListener() browser = r.ServiceBrowser(zc, type_, None, listener) browser.cancel() browser.cancel() browser.cancel() zc.close() def test_service_browser_cancel_context_manager(): """Test we can cancel a ServiceBrowser with it being used as a context manager.""" # instantiate a zeroconf instance zc = Zeroconf(interfaces=["127.0.0.1"]) # start a browser type_ = "_hap._tcp.local." class MyServiceListener(r.ServiceListener): pass listener = MyServiceListener() browser = r.ServiceBrowser(zc, type_, None, listener) assert cast(bool, browser.done) is False with browser: pass # ensure call_soon_threadsafe in ServiceBrowser.cancel is run assert zc.loop is not None asyncio.run_coroutine_threadsafe(asyncio.sleep(0), zc.loop).result() assert cast(bool, browser.done) is True zc.close() def test_service_browser_cancel_multiple_times_after_close(): """Test we can cancel a ServiceBrowser multiple times after close.""" # instantiate a zeroconf instance zc = Zeroconf(interfaces=["127.0.0.1"]) # start a browser type_ = "_hap._tcp.local." class MyServiceListener(r.ServiceListener): pass listener = MyServiceListener() browser = r.ServiceBrowser(zc, type_, None, listener) zc.close() browser.cancel() browser.cancel() browser.cancel() def test_service_browser_started_after_zeroconf_closed(): """Test starting a ServiceBrowser after close raises RuntimeError.""" # instantiate a zeroconf instance zc = Zeroconf(interfaces=["127.0.0.1"]) # start a browser type_ = "_hap._tcp.local." class MyServiceListener(r.ServiceListener): pass listener = MyServiceListener() zc.close() with pytest.raises(RuntimeError): r.ServiceBrowser(zc, type_, None, listener) def test_multiple_instances_running_close(): """Test we can shutdown multiple instances.""" # instantiate a zeroconf instance zc = Zeroconf(interfaces=["127.0.0.1"]) zc2 = Zeroconf(interfaces=["127.0.0.1"]) zc3 = Zeroconf(interfaces=["127.0.0.1"]) assert zc.loop != zc2.loop assert zc.loop != zc3.loop class MyServiceListener(r.ServiceListener): pass listener = MyServiceListener() zc2.add_service_listener("zca._hap._tcp.local.", listener) zc.close() zc2.remove_service_listener(listener) zc2.close() zc3.close() class TestServiceBrowser(unittest.TestCase): def test_update_record(self): enable_ipv6 = has_working_ipv6() and not os.environ.get("SKIP_IPV6") service_name = "name._type._tcp.local." service_type = "_type._tcp.local." service_server = "ash-1.local." service_text = b"path=/~matt1/" service_address = "10.0.1.2" service_v6_address = "2001:db8::1" service_v6_second_address = "6001:db8::1" service_added_count = 0 service_removed_count = 0 service_updated_count = 0 service_add_event = Event() service_removed_event = Event() service_updated_event = Event() class MyServiceListener(r.ServiceListener): def add_service(self, zc, type_, name) -> None: # type: ignore[no-untyped-def] nonlocal service_added_count service_added_count += 1 service_add_event.set() def remove_service(self, zc, type_, name) -> None: # type: ignore[no-untyped-def] nonlocal service_removed_count service_removed_count += 1 service_removed_event.set() def update_service(self, zc, type_, name) -> None: # type: ignore[no-untyped-def] nonlocal service_updated_count service_updated_count += 1 service_info = zc.get_service_info(type_, name) assert socket.inet_aton(service_address) in service_info.addresses if enable_ipv6: assert socket.inet_pton( socket.AF_INET6, service_v6_address ) in service_info.addresses_by_version(r.IPVersion.V6Only) assert socket.inet_pton( socket.AF_INET6, service_v6_second_address ) in service_info.addresses_by_version(r.IPVersion.V6Only) assert service_info.text == service_text assert service_info.server.lower() == service_server.lower() service_updated_event.set() def mock_record_update_incoming_msg( service_state_change: r.ServiceStateChange, ) -> r.DNSIncoming: generated = r.DNSOutgoing(const._FLAGS_QR_RESPONSE) assert generated.is_response() is True if service_state_change == r.ServiceStateChange.Removed: ttl = 0 else: ttl = 120 generated.add_answer_at_time( r.DNSText( service_name, const._TYPE_TXT, const._CLASS_IN | const._CLASS_UNIQUE, ttl, service_text, ), 0, ) generated.add_answer_at_time( r.DNSService( service_name, const._TYPE_SRV, const._CLASS_IN | const._CLASS_UNIQUE, ttl, 0, 0, 80, service_server, ), 0, ) # Send the IPv6 address first since we previously # had a bug where the IPv4 would be missing if the # IPv6 was seen first if enable_ipv6: generated.add_answer_at_time( r.DNSAddress( service_server, const._TYPE_AAAA, const._CLASS_IN | const._CLASS_UNIQUE, ttl, socket.inet_pton(socket.AF_INET6, service_v6_address), ), 0, ) generated.add_answer_at_time( r.DNSAddress( service_server, const._TYPE_AAAA, const._CLASS_IN | const._CLASS_UNIQUE, ttl, socket.inet_pton(socket.AF_INET6, service_v6_second_address), ), 0, ) generated.add_answer_at_time( r.DNSAddress( service_server, const._TYPE_A, const._CLASS_IN | const._CLASS_UNIQUE, ttl, socket.inet_aton(service_address), ), 0, ) generated.add_answer_at_time( r.DNSPointer(service_type, const._TYPE_PTR, const._CLASS_IN, ttl, service_name), 0, ) return r.DNSIncoming(generated.packets()[0]) zeroconf = r.Zeroconf(interfaces=["127.0.0.1"]) service_browser = r.ServiceBrowser(zeroconf, service_type, listener=MyServiceListener()) try: wait_time = 3 # service added _inject_response(zeroconf, mock_record_update_incoming_msg(r.ServiceStateChange.Added)) service_add_event.wait(wait_time) assert service_added_count == 1 assert service_updated_count == 0 assert service_removed_count == 0 # service SRV updated service_updated_event.clear() service_server = "ash-2.local." _inject_response(zeroconf, mock_record_update_incoming_msg(r.ServiceStateChange.Updated)) service_updated_event.wait(wait_time) assert service_added_count == 1 assert service_updated_count == 1 assert service_removed_count == 0 # service TXT updated service_updated_event.clear() service_text = b"path=/~matt2/" _inject_response(zeroconf, mock_record_update_incoming_msg(r.ServiceStateChange.Updated)) service_updated_event.wait(wait_time) assert service_added_count == 1 assert service_updated_count == 2 assert service_removed_count == 0 # service TXT updated - duplicate update should not trigger another service_updated service_updated_event.clear() service_text = b"path=/~matt2/" _inject_response(zeroconf, mock_record_update_incoming_msg(r.ServiceStateChange.Updated)) service_updated_event.wait(wait_time) assert service_added_count == 1 assert service_updated_count == 2 assert service_removed_count == 0 # service A updated service_updated_event.clear() service_address = "10.0.1.3" # Verify we match on uppercase service_server = service_server.upper() _inject_response(zeroconf, mock_record_update_incoming_msg(r.ServiceStateChange.Updated)) service_updated_event.wait(wait_time) assert service_added_count == 1 assert service_updated_count == 3 assert service_removed_count == 0 # service all updated service_updated_event.clear() service_server = "ash-3.local." service_text = b"path=/~matt3/" service_address = "10.0.1.3" _inject_response(zeroconf, mock_record_update_incoming_msg(r.ServiceStateChange.Updated)) service_updated_event.wait(wait_time) assert service_added_count == 1 assert service_updated_count == 4 assert service_removed_count == 0 # service removed _inject_response(zeroconf, mock_record_update_incoming_msg(r.ServiceStateChange.Removed)) service_removed_event.wait(wait_time) assert service_added_count == 1 assert service_updated_count == 4 assert service_removed_count == 1 finally: assert len(zeroconf.listeners) == 1 service_browser.cancel() time.sleep(0.2) assert len(zeroconf.listeners) == 0 zeroconf.remove_all_service_listeners() zeroconf.close() class TestServiceBrowserMultipleTypes(unittest.TestCase): def test_update_record(self): service_names = [ "name2._type2._tcp.local.", "name._type._tcp.local.", "name._type._udp.local", ] service_types = ["_type2._tcp.local.", "_type._tcp.local.", "_type._udp.local."] service_added_count = 0 service_removed_count = 0 service_add_event = Event() service_removed_event = Event() class MyServiceListener(r.ServiceListener): def add_service(self, zc, type_, name) -> None: # type: ignore[no-untyped-def] nonlocal service_added_count service_added_count += 1 if service_added_count == 3: service_add_event.set() def remove_service(self, zc, type_, name) -> None: # type: ignore[no-untyped-def] nonlocal service_removed_count service_removed_count += 1 if service_removed_count == 3: service_removed_event.set() def mock_record_update_incoming_msg( service_state_change: r.ServiceStateChange, service_type: str, service_name: str, ttl: int, ) -> r.DNSIncoming: generated = r.DNSOutgoing(const._FLAGS_QR_RESPONSE) generated.add_answer_at_time( r.DNSPointer(service_type, const._TYPE_PTR, const._CLASS_IN, ttl, service_name), 0, ) return r.DNSIncoming(generated.packets()[0]) zeroconf = r.Zeroconf(interfaces=["127.0.0.1"]) service_browser = r.ServiceBrowser(zeroconf, service_types, listener=MyServiceListener()) try: wait_time = 3 # all three services added _inject_response( zeroconf, mock_record_update_incoming_msg( r.ServiceStateChange.Added, service_types[0], service_names[0], 120 ), ) _inject_response( zeroconf, mock_record_update_incoming_msg( r.ServiceStateChange.Added, service_types[1], service_names[1], 120 ), ) time.sleep(0.1) called_with_refresh_time_check = False def _mock_get_expiration_time(self, percent): nonlocal called_with_refresh_time_check if percent == const._EXPIRE_REFRESH_TIME_PERCENT: called_with_refresh_time_check = True return 0 return self.created + (percent * self.ttl * 10) # Set an expire time that will force a refresh with patch("zeroconf.DNSRecord.get_expiration_time", new=_mock_get_expiration_time): _inject_response( zeroconf, mock_record_update_incoming_msg( r.ServiceStateChange.Added, service_types[0], service_names[0], 120, ), ) # Add the last record after updating the first one # to ensure the service_add_event only gets set # after the update _inject_response( zeroconf, mock_record_update_incoming_msg( r.ServiceStateChange.Added, service_types[2], service_names[2], 120, ), ) service_add_event.wait(wait_time) assert called_with_refresh_time_check is True assert service_added_count == 3 assert service_removed_count == 0 _inject_response( zeroconf, mock_record_update_incoming_msg( r.ServiceStateChange.Updated, service_types[0], service_names[0], 0 ), ) # all three services removed _inject_response( zeroconf, mock_record_update_incoming_msg( r.ServiceStateChange.Removed, service_types[0], service_names[0], 0 ), ) _inject_response( zeroconf, mock_record_update_incoming_msg( r.ServiceStateChange.Removed, service_types[1], service_names[1], 0 ), ) _inject_response( zeroconf, mock_record_update_incoming_msg( r.ServiceStateChange.Removed, service_types[2], service_names[2], 0 ), ) service_removed_event.wait(wait_time) assert service_added_count == 3 assert service_removed_count == 3 except TypeError: # Cannot be patched with cython as get_expiration_time is immutable pass finally: assert len(zeroconf.listeners) == 1 service_browser.cancel() time.sleep(0.2) assert len(zeroconf.listeners) == 0 zeroconf.remove_all_service_listeners() zeroconf.close() def test_first_query_delay(): """Verify the first query is delayed. https://datatracker.ietf.org/doc/html/rfc6762#section-5.2 """ type_ = "_http._tcp.local." zeroconf_browser = Zeroconf(interfaces=["127.0.0.1"]) _wait_for_start(zeroconf_browser) # we are going to patch the zeroconf send to check query transmission old_send = zeroconf_browser.async_send first_query_time = None def send(out, addr=const._MDNS_ADDR, port=const._MDNS_PORT): """Sends an outgoing packet.""" nonlocal first_query_time if first_query_time is None: first_query_time = current_time_millis() old_send(out, addr=addr, port=port) # patch the zeroconf send with patch.object(zeroconf_browser, "async_send", send): # dummy service callback def on_service_state_change(zeroconf, service_type, state_change, name): pass start_time = current_time_millis() browser = ServiceBrowser(zeroconf_browser, type_, [on_service_state_change]) time.sleep(millis_to_seconds(_services_browser._FIRST_QUERY_DELAY_RANDOM_INTERVAL[1] + 5)) try: assert ( current_time_millis() - start_time > _services_browser._FIRST_QUERY_DELAY_RANDOM_INTERVAL[0] ) finally: browser.cancel() zeroconf_browser.close() @pytest.mark.asyncio async def test_asking_default_is_asking_qm_questions_after_the_first_qu(): """Verify the service browser's first questions are QU and refresh queries are QM.""" service_added = asyncio.Event() service_removed = asyncio.Event() unexpected_ttl = asyncio.Event() got_query = asyncio.Event() type_ = "_http._tcp.local." registration_name = f"xxxyyy.{type_}" def on_service_state_change(zeroconf, service_type, state_change, name): if name == registration_name: if state_change is ServiceStateChange.Added: service_added.set() elif state_change is ServiceStateChange.Removed: service_removed.set() aiozc = AsyncZeroconf(interfaces=["127.0.0.1"]) zeroconf_browser = aiozc.zeroconf zeroconf_browser.question_history = QuestionHistoryWithoutSuppression() await zeroconf_browser.async_wait_for_start() # we are going to patch the zeroconf send to check packet sizes old_send = zeroconf_browser.async_send expected_ttl = const._DNS_OTHER_TTL questions: list[list[DNSQuestion]] = [] def send(out, addr=const._MDNS_ADDR, port=const._MDNS_PORT, v6_flow_scope=()): """Sends an outgoing packet.""" pout = r.DNSIncoming(out.packets()[0]) questions.append(pout.questions) got_query.set() old_send(out, addr=addr, port=port, v6_flow_scope=v6_flow_scope) assert len(zeroconf_browser.engine.protocols) == 2 aio_zeroconf_registrar = AsyncZeroconf(interfaces=["127.0.0.1"]) zeroconf_registrar = aio_zeroconf_registrar.zeroconf await aio_zeroconf_registrar.zeroconf.async_wait_for_start() assert len(zeroconf_registrar.engine.protocols) == 2 # patch the zeroconf send so we can capture what is being sent with patch.object(zeroconf_browser, "async_send", send): service_added = asyncio.Event() service_removed = asyncio.Event() browser = AsyncServiceBrowser(zeroconf_browser, type_, [on_service_state_change]) info = ServiceInfo( type_, registration_name, 80, 0, 0, {"path": "/~paulsm/"}, "ash-2.local.", addresses=[socket.inet_aton("10.0.1.2")], ) task = await aio_zeroconf_registrar.async_register_service(info) await task loop = asyncio.get_running_loop() try: await asyncio.wait_for(service_added.wait(), 1) assert service_added.is_set() # Make sure the startup queries are sent original_now = loop.time() now_millis = original_now * 1000 for query_count in range(_services_browser.STARTUP_QUERIES): now_millis += (2**query_count) * 1000 time_changed_millis(now_millis) got_query.clear() now_millis = original_now * 1000 assert not unexpected_ttl.is_set() # Move time forward past when the TTL is no longer # fresh (AKA 75% of the TTL) now_millis += (expected_ttl * 1000) * 0.80 time_changed_millis(now_millis) await asyncio.wait_for(got_query.wait(), 1) assert not unexpected_ttl.is_set() assert len(questions) == _services_browser.STARTUP_QUERIES + 1 # The first question should be QU to try to # populate the known answers and limit the impact # of the QM questions that follow. We still # have to ask QM questions for the startup queries # because some devices will not respond to QU assert questions[0][0].unicast is True # The remaining questions should be QM questions for question in questions[1:]: assert question[0].unicast is False # Don't remove service, allow close() to cleanup finally: await aio_zeroconf_registrar.async_close() await asyncio.wait_for(service_removed.wait(), 1) assert service_removed.is_set() await browser.async_cancel() await aiozc.async_close() @pytest.mark.asyncio async def test_ttl_refresh_cancelled_rescue_query(): """Verify seeing a name again cancels the rescue query.""" service_added = asyncio.Event() service_removed = asyncio.Event() unexpected_ttl = asyncio.Event() got_query = asyncio.Event() type_ = "_http._tcp.local." registration_name = f"xxxyyy.{type_}" def on_service_state_change(zeroconf, service_type, state_change, name): if name == registration_name: if state_change is ServiceStateChange.Added: service_added.set() elif state_change is ServiceStateChange.Removed: service_removed.set() aiozc = AsyncZeroconf(interfaces=["127.0.0.1"]) zeroconf_browser = aiozc.zeroconf zeroconf_browser.question_history = QuestionHistoryWithoutSuppression() await zeroconf_browser.async_wait_for_start() # we are going to patch the zeroconf send to check packet sizes old_send = zeroconf_browser.async_send expected_ttl = const._DNS_OTHER_TTL packets = [] def send(out, addr=const._MDNS_ADDR, port=const._MDNS_PORT, v6_flow_scope=()): """Sends an outgoing packet.""" pout = r.DNSIncoming(out.packets()[0]) packets.append(pout) got_query.set() old_send(out, addr=addr, port=port, v6_flow_scope=v6_flow_scope) assert len(zeroconf_browser.engine.protocols) == 2 aio_zeroconf_registrar = AsyncZeroconf(interfaces=["127.0.0.1"]) zeroconf_registrar = aio_zeroconf_registrar.zeroconf await aio_zeroconf_registrar.zeroconf.async_wait_for_start() assert len(zeroconf_registrar.engine.protocols) == 2 # patch the zeroconf send so we can capture what is being sent with patch.object(zeroconf_browser, "async_send", send): service_added = asyncio.Event() service_removed = asyncio.Event() browser = AsyncServiceBrowser(zeroconf_browser, type_, [on_service_state_change]) info = ServiceInfo( type_, registration_name, 80, 0, 0, {"path": "/~paulsm/"}, "ash-2.local.", addresses=[socket.inet_aton("10.0.1.2")], ) task = await aio_zeroconf_registrar.async_register_service(info) await task loop = asyncio.get_running_loop() try: await asyncio.wait_for(service_added.wait(), 1) assert service_added.is_set() # Make sure the startup queries are sent original_now = loop.time() now_millis = original_now * 1000 for query_count in range(_services_browser.STARTUP_QUERIES): now_millis += (2**query_count) * 1000 time_changed_millis(now_millis) now_millis = original_now * 1000 assert not unexpected_ttl.is_set() await asyncio.wait_for(got_query.wait(), 1) got_query.clear() assert len(packets) == _services_browser.STARTUP_QUERIES packets.clear() # Move time forward past when the TTL is no longer # fresh (AKA 75% of the TTL) now_millis += (expected_ttl * 1000) * 0.80 # Inject a response that will reschedule # the rescue query so it does not happen with patch("time.monotonic", return_value=now_millis / 1000): zeroconf_browser.record_manager.async_updates_from_response( mock_incoming_msg([info.dns_pointer()]), ) time_changed_millis(now_millis) await asyncio.sleep(0) # Verify we did not send a rescue query assert not packets # We should still get a rescue query once the rescheduled # query time is reached now_millis += (expected_ttl * 1000) * 0.76 time_changed_millis(now_millis) await asyncio.wait_for(got_query.wait(), 1) assert len(packets) == 1 # Don't remove service, allow close() to cleanup finally: await aio_zeroconf_registrar.async_close() await asyncio.wait_for(service_removed.wait(), 1) assert service_removed.is_set() await browser.async_cancel() await aiozc.async_close() @pytest.mark.asyncio async def test_asking_qm_questions(): """Verify explicitly asking QM questions.""" type_ = "_quservice._tcp.local." aiozc = AsyncZeroconf(interfaces=["127.0.0.1"]) zeroconf_browser = aiozc.zeroconf await zeroconf_browser.async_wait_for_start() # we are going to patch the zeroconf send to check query transmission old_send = zeroconf_browser.async_send first_outgoing = None def send(out, addr=const._MDNS_ADDR, port=const._MDNS_PORT): """Sends an outgoing packet.""" nonlocal first_outgoing if first_outgoing is None: first_outgoing = out old_send(out, addr=addr, port=port) # patch the zeroconf send with patch.object(zeroconf_browser, "async_send", send): # dummy service callback def on_service_state_change(zeroconf, service_type, state_change, name): pass browser = AsyncServiceBrowser( zeroconf_browser, type_, [on_service_state_change], question_type=r.DNSQuestionType.QM, ) await asyncio.sleep(millis_to_seconds(_services_browser._FIRST_QUERY_DELAY_RANDOM_INTERVAL[1] + 5)) try: assert first_outgoing.questions[0].unicast is False # type: ignore[union-attr] finally: await browser.async_cancel() await aiozc.async_close() @pytest.mark.asyncio async def test_asking_qu_questions(): """Verify the service browser can ask QU questions.""" type_ = "_quservice._tcp.local." aiozc = AsyncZeroconf(interfaces=["127.0.0.1"]) zeroconf_browser = aiozc.zeroconf await zeroconf_browser.async_wait_for_start() # we are going to patch the zeroconf send to check query transmission old_send = zeroconf_browser.async_send first_outgoing = None def send(out, addr=const._MDNS_ADDR, port=const._MDNS_PORT): """Sends an outgoing packet.""" nonlocal first_outgoing if first_outgoing is None: first_outgoing = out old_send(out, addr=addr, port=port) # patch the zeroconf send with patch.object(zeroconf_browser, "async_send", send): # dummy service callback def on_service_state_change(zeroconf, service_type, state_change, name): pass browser = AsyncServiceBrowser( zeroconf_browser, type_, [on_service_state_change], question_type=r.DNSQuestionType.QU, ) await asyncio.sleep(millis_to_seconds(_services_browser._FIRST_QUERY_DELAY_RANDOM_INTERVAL[1] + 5)) try: assert first_outgoing.questions[0].unicast is True # type: ignore[union-attr] finally: await browser.async_cancel() await aiozc.async_close() def test_legacy_record_update_listener(): """Test a RecordUpdateListener that does not implement update_records.""" # instantiate a zeroconf instance zc = Zeroconf(interfaces=["127.0.0.1"]) with pytest.raises(RuntimeError): r.RecordUpdateListener().update_record( zc, 0, r.DNSRecord("irrelevant", const._TYPE_SRV, const._CLASS_IN, const._DNS_HOST_TTL), ) updates = [] class LegacyRecordUpdateListener(r.RecordUpdateListener): """A RecordUpdateListener that does not implement update_records.""" def update_record(self, zc: Zeroconf, now: float, record: r.DNSRecord) -> None: updates.append(record) listener = LegacyRecordUpdateListener() zc.add_listener(listener, None) # dummy service callback def on_service_state_change(zeroconf, service_type, state_change, name): pass # start a browser type_ = "_homeassistant._tcp.local." name = "MyTestHome" browser = ServiceBrowser(zc, type_, [on_service_state_change]) info_service = ServiceInfo( type_, f"{name}.{type_}", 80, 0, 0, {"path": "/~paulsm/"}, "ash-2.local.", addresses=[socket.inet_aton("10.0.1.2")], ) zc.register_service(info_service) time.sleep(0.001) browser.cancel() assert updates assert len([isinstance(update, r.DNSPointer) and update.name == type_ for update in updates]) >= 1 zc.remove_listener(listener) # Removing a second time should not throw zc.remove_listener(listener) zc.close() def test_service_browser_is_aware_of_port_changes(): """Test that the ServiceBrowser is aware of port changes.""" # instantiate a zeroconf instance zc = Zeroconf(interfaces=["127.0.0.1"]) # start a browser type_ = "_hap._tcp.local." registration_name = f"xxxyyy.{type_}" callbacks = [] # dummy service callback def on_service_state_change(zeroconf, service_type, state_change, name): """Dummy callback.""" if name == registration_name: callbacks.append((service_type, state_change, name)) browser = ServiceBrowser(zc, type_, [on_service_state_change]) desc = {"path": "/~paulsm/"} address_parsed = "10.0.1.2" address = socket.inet_aton(address_parsed) info = ServiceInfo(type_, registration_name, 80, 0, 0, desc, "ash-2.local.", addresses=[address]) _inject_response( zc, mock_incoming_msg( [ info.dns_pointer(), info.dns_service(), info.dns_text(), *info.dns_addresses(), ] ), ) time.sleep(0.1) assert callbacks == [("_hap._tcp.local.", ServiceStateChange.Added, "xxxyyy._hap._tcp.local.")] service_info = zc.get_service_info(type_, registration_name) assert service_info is not None assert service_info.port == 80 info.port = 400 info._dns_service_cache = None # we are mutating the record so clear the cache _inject_response( zc, mock_incoming_msg([info.dns_service()]), ) time.sleep(0.1) assert callbacks == [ ("_hap._tcp.local.", ServiceStateChange.Added, "xxxyyy._hap._tcp.local."), ("_hap._tcp.local.", ServiceStateChange.Updated, "xxxyyy._hap._tcp.local."), ] service_info = zc.get_service_info(type_, registration_name) assert service_info is not None assert service_info.port == 400 browser.cancel() zc.close() def test_service_browser_listeners_update_service(): """Test that the ServiceBrowser ServiceListener that implements update_service.""" # instantiate a zeroconf instance zc = Zeroconf(interfaces=["127.0.0.1"]) # start a browser type_ = "_hap._tcp.local." registration_name = f"xxxyyy.{type_}" callbacks = [] class MyServiceListener(r.ServiceListener): def add_service(self, zc, type_, name) -> None: # type: ignore[no-untyped-def] if name == registration_name: callbacks.append(("add", type_, name)) def remove_service(self, zc, type_, name) -> None: # type: ignore[no-untyped-def] if name == registration_name: callbacks.append(("remove", type_, name)) def update_service(self, zc, type_, name) -> None: # type: ignore[no-untyped-def] if name == registration_name: callbacks.append(("update", type_, name)) listener = MyServiceListener() browser = r.ServiceBrowser(zc, type_, None, listener) desc = {"path": "/~paulsm/"} address_parsed = "10.0.1.2" address = socket.inet_aton(address_parsed) info = ServiceInfo(type_, registration_name, 80, 0, 0, desc, "ash-2.local.", addresses=[address]) _inject_response( zc, mock_incoming_msg( [ info.dns_pointer(), info.dns_service(), info.dns_text(), *info.dns_addresses(), ] ), ) time.sleep(0.2) info._dns_service_cache = None # we are mutating the record so clear the cache info.port = 400 _inject_response( zc, mock_incoming_msg([info.dns_service()]), ) time.sleep(0.2) assert callbacks == [ ("add", type_, registration_name), ("update", type_, registration_name), ] browser.cancel() zc.close() def test_service_browser_listeners_no_update_service(): """Test that the ServiceBrowser ServiceListener that does not implement update_service.""" # instantiate a zeroconf instance zc = Zeroconf(interfaces=["127.0.0.1"]) # start a browser type_ = "_hap._tcp.local." registration_name = f"xxxyyy.{type_}" callbacks = [] class MyServiceListener(r.ServiceListener): def add_service(self, zc, type_, name) -> None: # type: ignore[no-untyped-def] if name == registration_name: callbacks.append(("add", type_, name)) def remove_service(self, zc, type_, name) -> None: # type: ignore[no-untyped-def] if name == registration_name: callbacks.append(("remove", type_, name)) listener = MyServiceListener() browser = r.ServiceBrowser(zc, type_, None, listener) desc = {"path": "/~paulsm/"} address_parsed = "10.0.1.2" address = socket.inet_aton(address_parsed) info = ServiceInfo(type_, registration_name, 80, 0, 0, desc, "ash-2.local.", addresses=[address]) _inject_response( zc, mock_incoming_msg( [ info.dns_pointer(), info.dns_service(), info.dns_text(), *info.dns_addresses(), ] ), ) time.sleep(0.2) info.port = 400 info._dns_service_cache = None # we are mutating the record so clear the cache _inject_response( zc, mock_incoming_msg([info.dns_service()]), ) time.sleep(0.2) assert callbacks == [ ("add", type_, registration_name), ] browser.cancel() zc.close() def test_service_browser_uses_non_strict_names(): """Verify we can look for technically invalid names as we cannot change what others do.""" # dummy service callback def on_service_state_change(zeroconf, service_type, state_change, name): pass zc = r.Zeroconf(interfaces=["127.0.0.1"]) browser = ServiceBrowser(zc, ["_tivo-videostream._tcp.local."], [on_service_state_change]) browser.cancel() # Still fail on completely invalid with pytest.raises(r.BadTypeInNameException): browser = ServiceBrowser(zc, ["tivo-videostream._tcp.local."], [on_service_state_change]) zc.close() def test_group_ptr_queries_with_known_answers(): questions_with_known_answers: _services_browser._QuestionWithKnownAnswers = {} now = current_time_millis() for i in range(120): name = f"_hap{i}._tcp._local." questions_with_known_answers[DNSQuestion(name, const._TYPE_PTR, const._CLASS_IN)] = { DNSPointer( name, const._TYPE_PTR, const._CLASS_IN, 4500, f"zoo{counter}.{name}", ) for counter in range(i) } outs = _services_browser.group_ptr_queries_with_known_answers(now, True, questions_with_known_answers) for out in outs: packets = out.packets() # If we generate multiple packets there must # only be one question assert len(packets) == 1 or len(out.questions) == 1 # This test uses asyncio because it needs to access the cache directly # which is not threadsafe @pytest.mark.asyncio async def test_generate_service_query_suppress_duplicate_questions(): """Generate a service query for sending with zeroconf.send.""" aiozc = AsyncZeroconf(interfaces=["127.0.0.1"]) zc = aiozc.zeroconf now = current_time_millis() name = "_suppresstest._tcp.local." question = r.DNSQuestion(name, const._TYPE_PTR, const._CLASS_IN) answer = r.DNSPointer( name, const._TYPE_PTR, const._CLASS_IN, 10000, f"known-to-other.{name}", ) other_known_answers: set[r.DNSRecord] = {answer} zc.question_history.add_question_at_time(question, now, other_known_answers) assert zc.question_history.suppresses(question, now, other_known_answers) # The known answer list is different, do not suppress outs = _services_browser.generate_service_query(zc, now, {name}, multicast=True, question_type=None) assert outs zc.cache.async_add_records([answer]) # The known answer list contains all the asked questions in the history # we should suppress outs = _services_browser.generate_service_query(zc, now, {name}, multicast=True, question_type=None) assert not outs # We do not suppress once the question history expires outs = _services_browser.generate_service_query( zc, now + 1000, {name}, multicast=True, question_type=None ) assert outs # We do not suppress QU queries ever outs = _services_browser.generate_service_query(zc, now, {name}, multicast=False, question_type=None) assert outs zc.question_history.async_expire(now + 2000) # No suppression after clearing the history outs = _services_browser.generate_service_query(zc, now, {name}, multicast=True, question_type=None) assert outs # The previous query we just sent is still remembered and # the next one is suppressed outs = _services_browser.generate_service_query(zc, now, {name}, multicast=True, question_type=None) assert not outs await aiozc.async_close() @pytest.mark.asyncio async def test_query_scheduler(): delay = const._BROWSER_TIME types_ = {"_hap._tcp.local.", "_http._tcp.local."} aiozc = AsyncZeroconf(interfaces=["127.0.0.1"]) await aiozc.zeroconf.async_wait_for_start() zc = aiozc.zeroconf sends: list[r.DNSIncoming] = [] def send(out, addr=const._MDNS_ADDR, port=const._MDNS_PORT, v6_flow_scope=()): """Sends an outgoing packet.""" pout = r.DNSIncoming(out.packets()[0]) sends.append(pout) query_scheduler = _services_browser.QueryScheduler(zc, types_, None, 0, True, delay, (0, 0), None) loop = asyncio.get_running_loop() # patch the zeroconf send so we can capture what is being sent with patch.object(zc, "async_send", send): query_scheduler.start(loop) original_now = loop.time() now_millis = original_now * 1000 for query_count in range(_services_browser.STARTUP_QUERIES): now_millis += (2**query_count) * 1000 time_changed_millis(now_millis) ptr_record = r.DNSPointer( "_hap._tcp.local.", const._TYPE_PTR, const._CLASS_IN, const._DNS_OTHER_TTL, "zoomer._hap._tcp.local.", ) ptr2_record = r.DNSPointer( "_hap._tcp.local.", const._TYPE_PTR, const._CLASS_IN, const._DNS_OTHER_TTL, "disappear._hap._tcp.local.", ) query_scheduler.reschedule_ptr_first_refresh(ptr_record) expected_when_time = ptr_record.get_expiration_time(const._EXPIRE_REFRESH_TIME_PERCENT) expected_expire_time = ptr_record.get_expiration_time(100) ptr_query = _ScheduledPTRQuery( ptr_record.alias, ptr_record.name, int(ptr_record.ttl), expected_expire_time, expected_when_time, ) assert query_scheduler._query_heap == [ptr_query] query_scheduler.reschedule_ptr_first_refresh(ptr2_record) expected_when_time = ptr2_record.get_expiration_time(const._EXPIRE_REFRESH_TIME_PERCENT) expected_expire_time = ptr2_record.get_expiration_time(100) ptr2_query = _ScheduledPTRQuery( ptr2_record.alias, ptr2_record.name, int(ptr2_record.ttl), expected_expire_time, expected_when_time, ) assert query_scheduler._query_heap == [ptr_query, ptr2_query] # Simulate PTR one goodbye query_scheduler.cancel_ptr_refresh(ptr_record) ptr_query.cancelled = True assert query_scheduler._query_heap == [ptr_query, ptr2_query] assert query_scheduler._query_heap[0].cancelled is True assert query_scheduler._query_heap[1].cancelled is False # Move time forward past when the TTL is no longer # fresh (AKA 75% of the TTL) now_millis += (ptr2_record.ttl * 1000) * 0.80 time_changed_millis(now_millis) assert len(query_scheduler._query_heap) == 1 first_heap = query_scheduler._query_heap[0] assert first_heap.cancelled is False assert first_heap.alias == ptr2_record.alias # Move time forward past when the record expires now_millis += (ptr2_record.ttl * 1000) * 0.20 time_changed_millis(now_millis) assert len(query_scheduler._query_heap) == 0 await aiozc.async_close() @pytest.mark.asyncio async def test_query_scheduler_rescue_records(): delay = const._BROWSER_TIME types_ = {"_hap._tcp.local.", "_http._tcp.local."} aiozc = AsyncZeroconf(interfaces=["127.0.0.1"]) await aiozc.zeroconf.async_wait_for_start() zc = aiozc.zeroconf sends: list[r.DNSIncoming] = [] def send(out, addr=const._MDNS_ADDR, port=const._MDNS_PORT, v6_flow_scope=()): """Sends an outgoing packet.""" pout = r.DNSIncoming(out.packets()[0]) sends.append(pout) query_scheduler = _services_browser.QueryScheduler(zc, types_, None, 0, True, delay, (0, 0), None) loop = asyncio.get_running_loop() # patch the zeroconf send so we can capture what is being sent with patch.object(zc, "async_send", send): query_scheduler.start(loop) original_now = loop.time() now_millis = original_now * 1000 for query_count in range(_services_browser.STARTUP_QUERIES): now_millis += (2**query_count) * 1000 time_changed_millis(now_millis) ptr_record = r.DNSPointer( "_hap._tcp.local.", const._TYPE_PTR, const._CLASS_IN, const._DNS_OTHER_TTL, "zoomer._hap._tcp.local.", ) query_scheduler.reschedule_ptr_first_refresh(ptr_record) expected_when_time = ptr_record.get_expiration_time(const._EXPIRE_REFRESH_TIME_PERCENT) expected_expire_time = ptr_record.get_expiration_time(100) ptr_query = _ScheduledPTRQuery( ptr_record.alias, ptr_record.name, int(ptr_record.ttl), expected_expire_time, expected_when_time, ) assert query_scheduler._query_heap == [ptr_query] assert query_scheduler._query_heap[0].cancelled is False # Move time forward past when the TTL is no longer # fresh (AKA 75% of the TTL) now_millis += (ptr_record.ttl * 1000) * 0.76 time_changed_millis(now_millis) assert len(query_scheduler._query_heap) == 1 new_when = query_scheduler._query_heap[0].when_millis assert query_scheduler._query_heap[0].cancelled is False assert new_when >= expected_when_time # Move time forward again, but not enough to expire the # record to make sure we try to rescue it now_millis += (ptr_record.ttl * 1000) * 0.11 time_changed_millis(now_millis) assert len(query_scheduler._query_heap) == 1 second_new_when = query_scheduler._query_heap[0].when_millis assert query_scheduler._query_heap[0].cancelled is False assert second_new_when >= new_when # Move time forward again, enough that we will no longer # try to rescue the record now_millis += (ptr_record.ttl * 1000) * 0.11 time_changed_millis(now_millis) assert len(query_scheduler._query_heap) == 0 await aiozc.async_close() def test_service_browser_matching(): """Test that the ServiceBrowser matching does not match partial names.""" # instantiate a zeroconf instance zc = Zeroconf(interfaces=["127.0.0.1"]) # start a browser type_ = "_http._tcp.local." registration_name = f"xxxyyy.{type_}" not_match_type_ = "_asustor-looksgood_http._tcp.local." not_match_registration_name = f"xxxyyy.{not_match_type_}" callbacks = [] class MyServiceListener(r.ServiceListener): def add_service(self, zc, type_, name) -> None: # type: ignore[no-untyped-def] if name == registration_name: callbacks.append(("add", type_, name)) def remove_service(self, zc, type_, name) -> None: # type: ignore[no-untyped-def] if name == registration_name: callbacks.append(("remove", type_, name)) def update_service(self, zc, type_, name) -> None: # type: ignore[no-untyped-def] if name == registration_name: callbacks.append(("update", type_, name)) listener = MyServiceListener() browser = r.ServiceBrowser(zc, type_, None, listener) desc = {"path": "/~paulsm/"} address_parsed = "10.0.1.2" address = socket.inet_aton(address_parsed) info = ServiceInfo(type_, registration_name, 80, 0, 0, desc, "ash-2.local.", addresses=[address]) should_not_match = ServiceInfo( not_match_type_, not_match_registration_name, 80, 0, 0, desc, "ash-2.local.", addresses=[address], ) _inject_response( zc, mock_incoming_msg( [ info.dns_pointer(), info.dns_service(), info.dns_text(), *info.dns_addresses(), ] ), ) _inject_response( zc, mock_incoming_msg( [ should_not_match.dns_pointer(), should_not_match.dns_service(), should_not_match.dns_text(), *should_not_match.dns_addresses(), ] ), ) time.sleep(0.2) info.port = 400 info._dns_service_cache = None # we are mutating the record so clear the cache _inject_response( zc, mock_incoming_msg([info.dns_service()]), ) should_not_match.port = 400 _inject_response( zc, mock_incoming_msg([should_not_match.dns_service()]), ) time.sleep(0.2) assert callbacks == [ ("add", type_, registration_name), ("update", type_, registration_name), ] browser.cancel() zc.close() @pytest.mark.skipif(os.environ.get("DEBIAN_TEST"), reason="disabled by Debian builder") @patch.object(_engine, "_CACHE_CLEANUP_INTERVAL", 0.01) def test_service_browser_expire_callbacks(): """Test that the ServiceBrowser matching does not match partial names.""" # instantiate a zeroconf instance zc = Zeroconf(interfaces=["127.0.0.1"]) # start a browser type_ = "_old._tcp.local." registration_name = f"uniquezip323.{type_}" callbacks = [] class MyServiceListener(r.ServiceListener): def add_service(self, zc, type_, name) -> None: # type: ignore[no-untyped-def] if name == registration_name: callbacks.append(("add", type_, name)) def remove_service(self, zc, type_, name) -> None: # type: ignore[no-untyped-def] if name == registration_name: callbacks.append(("remove", type_, name)) def update_service(self, zc, type_, name) -> None: # type: ignore[no-untyped-def] if name == registration_name: callbacks.append(("update", type_, name)) listener = MyServiceListener() browser = r.ServiceBrowser(zc, type_, None, listener) desc = {"path": "/~paul2/"} address_parsed = "10.0.1.3" address = socket.inet_aton(address_parsed) info = ServiceInfo( type_, registration_name, 80, 0, 0, desc, "newname-2.local.", host_ttl=1, other_ttl=1, addresses=[address], ) _inject_response( zc, mock_incoming_msg( [ info.dns_pointer(), info.dns_service(), info.dns_text(), *info.dns_addresses(), ] ), ) # Force the ttl to be 1 second now = current_time_millis() for cache_record in list(zc.cache.cache.values()): for record in cache_record: zc.cache._async_set_created_ttl(record, now, 1) time.sleep(0.3) info.port = 400 info._dns_service_cache = None # we are mutating the record so clear the cache _inject_response( zc, mock_incoming_msg([info.dns_service()]), ) for _ in range(10): time.sleep(0.05) if len(callbacks) == 2: break assert callbacks == [ ("add", type_, registration_name), ("update", type_, registration_name), ] for _ in range(25): time.sleep(0.05) if len(callbacks) == 3: break assert callbacks == [ ("add", type_, registration_name), ("update", type_, registration_name), ("remove", type_, registration_name), ] browser.cancel() zc.close() def test_scheduled_ptr_query_dunder_methods(): query75 = _ScheduledPTRQuery("zoomy._hap._tcp.local.", "_hap._tcp.local.", 120, 120, 75) query80 = _ScheduledPTRQuery("zoomy._hap._tcp.local.", "_hap._tcp.local.", 120, 120, 80) query75_2 = _ScheduledPTRQuery("zoomy._hap._tcp.local.", "_hap._tcp.local.", 120, 140, 75) other = object() stringified = str(query75) assert "zoomy._hap._tcp.local." in stringified assert "120" in stringified assert "75" in stringified assert "ScheduledPTRQuery" in stringified assert query75 == query75 assert query75 != query80 assert query75 == query75_2 assert query75 < query80 assert query75 <= query80 assert query80 > query75 assert query80 >= query75 assert query75 != other with pytest.raises(TypeError): assert query75 < other # type: ignore[operator] with pytest.raises(TypeError): assert query75 <= other # type: ignore[operator] with pytest.raises(TypeError): assert query75 > other # type: ignore[operator] with pytest.raises(TypeError): assert query75 >= other # type: ignore[operator] @pytest.mark.asyncio async def test_close_zeroconf_without_browser_before_start_up_queries(): """Test that we stop sending startup queries if zeroconf is closed out from under the browser.""" service_added = asyncio.Event() type_ = "_http._tcp.local." registration_name = f"xxxyyy.{type_}" def on_service_state_change(zeroconf, service_type, state_change, name): if name == registration_name: if state_change is ServiceStateChange.Added: service_added.set() aiozc = AsyncZeroconf(interfaces=["127.0.0.1"]) zeroconf_browser = aiozc.zeroconf zeroconf_browser.question_history = QuestionHistoryWithoutSuppression() await zeroconf_browser.async_wait_for_start() sends: list[r.DNSIncoming] = [] def send(out, addr=const._MDNS_ADDR, port=const._MDNS_PORT, v6_flow_scope=()): """Sends an outgoing packet.""" pout = r.DNSIncoming(out.packets()[0]) sends.append(pout) assert len(zeroconf_browser.engine.protocols) == 2 aio_zeroconf_registrar = AsyncZeroconf(interfaces=["127.0.0.1"]) zeroconf_registrar = aio_zeroconf_registrar.zeroconf await aio_zeroconf_registrar.zeroconf.async_wait_for_start() assert len(zeroconf_registrar.engine.protocols) == 2 # patch the zeroconf send so we can capture what is being sent with patch.object(zeroconf_browser, "async_send", send): service_added = asyncio.Event() browser = AsyncServiceBrowser(zeroconf_browser, type_, [on_service_state_change]) info = ServiceInfo( type_, registration_name, 80, 0, 0, {"path": "/~paulsm/"}, "ash-2.local.", addresses=[socket.inet_aton("10.0.1.2")], ) task = await aio_zeroconf_registrar.async_register_service(info) await task loop = asyncio.get_running_loop() try: await asyncio.wait_for(service_added.wait(), 1) assert service_added.is_set() await aiozc.async_close() sends.clear() # Make sure the startup queries are sent original_now = loop.time() now_millis = original_now * 1000 for query_count in range(_services_browser.STARTUP_QUERIES): now_millis += (2**query_count) * 1000 time_changed_millis(now_millis) # We should not send any queries after close assert not sends finally: await aio_zeroconf_registrar.async_close() await browser.async_cancel() @pytest.mark.asyncio async def test_close_zeroconf_without_browser_after_start_up_queries(): """Test that we stop sending rescue queries if zeroconf is closed out from under the browser.""" service_added = asyncio.Event() type_ = "_http._tcp.local." registration_name = f"xxxyyy.{type_}" def on_service_state_change(zeroconf, service_type, state_change, name): if name == registration_name: if state_change is ServiceStateChange.Added: service_added.set() aiozc = AsyncZeroconf(interfaces=["127.0.0.1"]) zeroconf_browser = aiozc.zeroconf zeroconf_browser.question_history = QuestionHistoryWithoutSuppression() await zeroconf_browser.async_wait_for_start() sends: list[r.DNSIncoming] = [] def send(out, addr=const._MDNS_ADDR, port=const._MDNS_PORT, v6_flow_scope=()): """Sends an outgoing packet.""" pout = r.DNSIncoming(out.packets()[0]) sends.append(pout) assert len(zeroconf_browser.engine.protocols) == 2 aio_zeroconf_registrar = AsyncZeroconf(interfaces=["127.0.0.1"]) zeroconf_registrar = aio_zeroconf_registrar.zeroconf await aio_zeroconf_registrar.zeroconf.async_wait_for_start() assert len(zeroconf_registrar.engine.protocols) == 2 # patch the zeroconf send so we can capture what is being sent with patch.object(zeroconf_browser, "async_send", send): service_added = asyncio.Event() browser = AsyncServiceBrowser(zeroconf_browser, type_, [on_service_state_change]) expected_ttl = const._DNS_OTHER_TTL info = ServiceInfo( type_, registration_name, 80, 0, 0, {"path": "/~paulsm/"}, "ash-2.local.", addresses=[socket.inet_aton("10.0.1.2")], ) task = await aio_zeroconf_registrar.async_register_service(info) await task loop = asyncio.get_running_loop() try: await asyncio.wait_for(service_added.wait(), 1) assert service_added.is_set() sends.clear() # Make sure the startup queries are sent original_now = loop.time() now_millis = original_now * 1000 for query_count in range(_services_browser.STARTUP_QUERIES): now_millis += (2**query_count) * 1000 time_changed_millis(now_millis) # We should not send any queries after close assert sends await aiozc.async_close() sends.clear() now_millis = original_now * 1000 # Move time forward past when the TTL is no longer # fresh (AKA 75% of the TTL) now_millis += (expected_ttl * 1000) * 0.80 time_changed_millis(now_millis) # We should not send the query after close assert not sends finally: await aio_zeroconf_registrar.async_close() await browser.async_cancel()