python-zeroconf · bdraco · May 18, 2026 · May 17, 2026 · May 18, 2026
diff --git a/tests/__init__.py b/tests/__init__.py
@@ -35,6 +35,13 @@
 
 _MONOTONIC_RESOLUTION = time.get_clock_info("monotonic").resolution
 
+# get_service_info / async_request timeout for tests using the
+# `quick_request_timing` fixture. The fixture cuts the initial-query
+# delay to ~15ms (10ms _LISTENER_TIME + 1-5ms jitter), so 50ms is
+# ample headroom for tests that only need to observe the first one
+# or two queries.
+QUICK_REQUEST_TIMEOUT_MS = 50
+
 
 class QuestionHistoryWithoutSuppression(QuestionHistory):
     def suppresses(self, question: DNSQuestion, now: float, known_answers: set[DNSRecord]) -> bool:

diff --git a/tests/conftest.py b/tests/conftest.py
@@ -10,6 +10,7 @@
 
 from zeroconf import _core, const
 from zeroconf._handlers import query_handler
+from zeroconf._services import info as service_info
 
 
 @pytest.fixture(autouse=True)
@@ -59,3 +60,21 @@ def quick_timing() -> Generator[None]:
         patch.object(_core, "_UNREGISTER_TIME", 10),
     ):
         yield
+
+
+@pytest.fixture
+def quick_request_timing() -> Generator[None]:
+    """Shorten the initial-query delay used by AsyncServiceInfo.async_request.
+
+    The 200ms `_LISTENER_TIME` and 20-120ms random jitter (RFC 6762
+    §5.2) help spread queries from multiple clients on real networks.
+    On loopback they're pure overhead — get_service_info-style tests
+    wait ~250ms before the first query even fires. Opt in by adding
+    `quick_request_timing` to a test's argument list, then drop the
+    test's own timeouts (which had to accommodate that delay).
+    """
+    with (
+        patch.object(service_info, "_LISTENER_TIME", 10),
+        patch.object(service_info, "_AVOID_SYNC_DELAY_RANDOM_INTERVAL", (1, 5)),
+    ):
+        yield
diff --git a/tests/services/test_info.py b/tests/services/test_info.py
@@ -23,7 +23,7 @@
 from zeroconf._utils.net import IPVersion
 from zeroconf.asyncio import AsyncZeroconf
 
-from .. import _inject_response, has_working_ipv6
+from .. import QUICK_REQUEST_TIMEOUT_MS, _inject_response, has_working_ipv6
 
 log = logging.getLogger("zeroconf")
 original_logging_level = logging.NOTSET
@@ -506,6 +506,7 @@ def get_service_info_helper(zc, type, name):
                 zc.remove_all_service_listeners()
                 zc.close()
 
+    @pytest.mark.usefixtures("quick_request_timing")
     def test_get_info_single(self):
         zc = r.Zeroconf(interfaces=["127.0.0.1"])
 
@@ -551,6 +552,9 @@ def get_service_info_helper(zc, type, name):
                     args=(zc, service_type, service_name),
                 )
                 helper_thread.start()
+                # Positive wait — the first query fires within
+                # `_LISTENER_TIME` + jitter (~15ms under
+                # `quick_request_timing`, ~320ms without).
                 wait_time = 1
 
                 # Expect query for SRV, TXT, A, AAAA
@@ -563,7 +567,10 @@ def get_service_info_helper(zc, type, name):
                 assert r.DNSQuestion(service_name, const._TYPE_AAAA, const._CLASS_IN) in last_sent.questions
                 assert service_info is None
 
-                # Expect no further queries
+                # Expect no further queries — under `quick_request_timing`
+                # the next query would have fired ~15ms after the previous
+                # send, so 100ms is plenty of headroom for the negative
+                # assertion.
                 last_sent = None
                 send_event.clear()
                 _inject_response(
@@ -597,7 +604,7 @@ def get_service_info_helper(zc, type, name):
                         ]
                     ),
                 )
-                send_event.wait(wait_time)
+                send_event.wait(0.1)
                 assert last_sent is None
                 assert service_info is not None
 
@@ -980,7 +987,7 @@ def test_serviceinfo_accepts_bytes_or_string_dict():
     assert info_service.dns_text().text == b"\x0epath=/~paulsm/"
 
 
-def test_asking_qu_questions():
+def test_asking_qu_questions(quick_request_timing):
     """Verify explicitly asking QU questions."""
     type_ = "_quservice._tcp.local."
     zeroconf = r.Zeroconf(interfaces=["127.0.0.1"])
@@ -999,12 +1006,14 @@ def send(out, addr=const._MDNS_ADDR, port=const._MDNS_PORT):
 
     # patch the zeroconf send
     with patch.object(zeroconf, "async_send", send):
-        zeroconf.get_service_info(f"name.{type_}", type_, 500, question_type=r.DNSQuestionType.QU)
+        zeroconf.get_service_info(
+            f"name.{type_}", type_, QUICK_REQUEST_TIMEOUT_MS, question_type=r.DNSQuestionType.QU
+        )
         assert first_outgoing.questions[0].unicast is True  # type: ignore[union-attr]
         zeroconf.close()
 
 
-def test_asking_qm_questions():
+def test_asking_qm_questions(quick_request_timing):
     """Verify explicitly asking QM questions."""
     type_ = "_quservice._tcp.local."
     zeroconf = r.Zeroconf(interfaces=["127.0.0.1"])
@@ -1023,7 +1032,9 @@ def send(out, addr=const._MDNS_ADDR, port=const._MDNS_PORT):
 
     # patch the zeroconf send
     with patch.object(zeroconf, "async_send", send):
-        zeroconf.get_service_info(f"name.{type_}", type_, 500, question_type=r.DNSQuestionType.QM)
+        zeroconf.get_service_info(
+            f"name.{type_}", type_, QUICK_REQUEST_TIMEOUT_MS, question_type=r.DNSQuestionType.QM
+        )
         assert first_outgoing.questions[0].unicast is False  # type: ignore[union-attr]
         zeroconf.close()
 

diff --git a/tests/test_asyncio.py b/tests/test_asyncio.py
@@ -43,6 +43,7 @@
 from zeroconf.const import _LISTENER_TIME
 
 from . import (
+    QUICK_REQUEST_TIMEOUT_MS,
     QuestionHistoryWithoutSuppression,
     _clear_cache,
     has_working_ipv6,
@@ -1139,7 +1140,7 @@ def send(out, addr=const._MDNS_ADDR, port=const._MDNS_PORT, v6_flow_scope=()):
 
 
 @pytest.mark.asyncio
-async def test_info_asking_default_is_asking_qm_questions_after_the_first_qu():
+async def test_info_asking_default_is_asking_qm_questions_after_the_first_qu(quick_request_timing):
     """Verify the service info first question is QU and subsequent ones are QM questions."""
     type_ = "_quservice._tcp.local."
     aiozc = AsyncZeroconf(interfaces=["127.0.0.1"])
@@ -1182,11 +1183,11 @@ def send(out, addr=const._MDNS_ADDR, port=const._MDNS_PORT):
     # patch the zeroconf send
     with patch.object(zeroconf_info, "async_send", send):
         aiosinfo = AsyncServiceInfo(type_, registration_name)
-        # Patch _is_complete so we send multiple times. 500ms covers
-        # the QU query at 0ms plus the QM query at ~_LISTENER_TIME +
-        # max random delay (~320ms).
+        # Patch _is_complete so we send multiple times. Under
+        # `quick_request_timing` both the QU query at 0ms and the QM
+        # query at ~15ms land well inside QUICK_REQUEST_TIMEOUT_MS.
         with patch("zeroconf.asyncio.AsyncServiceInfo._is_complete", False):
-            await aiosinfo.async_request(aiozc.zeroconf, 500)
+            await aiosinfo.async_request(aiozc.zeroconf, QUICK_REQUEST_TIMEOUT_MS)
         try:
             assert first_outgoing.questions[0].unicast is True  # type: ignore[union-attr]
             assert second_outgoing.questions[0].unicast is False  # type: ignore[attr-defined]