Skip to content

test: synchronise test_integration on browser first-query #1771

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
bdraco merged 2 commits into from
May 21, 2026
+20 −5
Merged

test: synchronise test_integration on browser first-query #1771

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
4fb1005
test: synchronise test_integration on browser first-query
bluetoothbot May 21, 2026
4d00bb6
test: isolate first startup query in test_integration to avoid count …
bluetoothbot May 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 20 additions & 5 deletions tests/test_asyncio.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1066,6 +1066,19 @@ def send(out, addr=const._MDNS_ADDR, port=const._MDNS_PORT, v6_flow_scope=()):
"ash-2.local.",
addresses=[socket.inet_aton("10.0.1.2")],
)
# Wait for the browser's first startup query to land (with an empty
# cache) before registering — otherwise on fast loopback the register
# may finish before the first query fires, and answers[0] picks up
# the known PTR via RFC 6762 §7.1 known-answer suppression.
await asyncio.wait_for(got_query.wait(), 1)
Comment thread
bdraco marked this conversation as resolved.
# Snapshot the first query's answers and reset the captures so the
# subsequent assertions don't have to predict whether further startup
# queries fire in real time (before the manual time_changed_millis
# loop) or under mock time.
first_answers = answers[0]
packets.clear()
answers.clear()
got_query.clear()
task = await aio_zeroconf_registrar.async_register_service(info)
await task
loop = asyncio.get_running_loop()
Expand All @@ -1084,7 +1097,9 @@ def send(out, addr=const._MDNS_ADDR, port=const._MDNS_PORT, v6_flow_scope=()):
got_query.clear()
assert not unexpected_ttl.is_set()

assert len(packets) == _services_browser.STARTUP_QUERIES
# The first startup query was captured separately, so only the
# remaining STARTUP_QUERIES - 1 land here.
assert len(packets) == _services_browser.STARTUP_QUERIES - 1
packets.clear()

# Wait for the first refresh query
Expand All @@ -1098,12 +1113,12 @@ def send(out, addr=const._MDNS_ADDR, port=const._MDNS_PORT, v6_flow_scope=()):
assert len(packets) == 1
packets.clear()

assert len(answers) == _services_browser.STARTUP_QUERIES + 1
# The first question should have no known answers
assert len(answers[0]) == 0
assert len(answers) == _services_browser.STARTUP_QUERIES
# The first question (captured separately) should have no known answers
assert len(first_answers) == 0
# The rest of the startup questions should have
# known answers
for answer_list in answers[1:-2]:
for answer_list in answers[:-2]:
# Allow 0 or 1 answers due to random delays and timing
assert len(answer_list) <= 1
# Once the TTL is reached, the last question should have no known answers
Expand Down
Loading