Skip to content

test(streams): harden take() ack tests against event-loop timing (fix PyPy flake) #709

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

Open
wbarnha wants to merge 3 commits into
base: master
Choose a base branch
from
+29 −16
Open

test(streams): harden take() ack tests against event-loop timing (fix PyPy flake) #709

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
3182eb1
test(streams): harden take() ack tests against event-loop timing
wbarnha Jul 19, 2026
2abee2c
Merge branch 'master' into claude/harden-take-tests
wbarnha Jul 19, 2026
03e5fde
test(streams): force gc in wait_for_stream_ack so take() acks land on…
wbarnha Jul 20, 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
45 changes: 29 additions & 16 deletions tests/functional/test_streams.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import asyncio
import gc
import platform
from copy import copy
from unittest.mock import Mock, patch
Expand Down Expand Up @@ -572,6 +573,30 @@ def mock_event_ack(event, return_value=False):
return event


async def wait_for_stream_ack(event, *, timeout=1.0):
"""Wait until a taken event has been acked.

Breaking out of ``async for value in s.take(...)`` leaves the ``take()``
async generator suspended; the ``finally:`` block that acks the buffered
events only runs once the generator is finalized. CPython's refcounting
finalizes it as soon as the loop exits (so the ack lands within a couple
of event-loop ticks), but PyPy has no refcounting -- without an explicit
``gc.collect()`` the finalizer (and thus the ack) may never run at all,
no matter how long we sleep. So poll for the ack, nudging the collector
each round to trigger generator finalization, exactly like
``Consumer.wait_empty()`` does in faust itself. If the ack never lands
the loop times out and returns, leaving the caller's assertions to report
the actual state.
"""
loop = asyncio.get_event_loop()
deadline = loop.time() + timeout
while loop.time() < deadline:
if event.ack.called or event.message.acked:
return
gc.collect()
await asyncio.sleep(0.05)


async def get_event_from_value(stream, value, key=None):
await stream.channel.send(key=key, value=value)
async for value in stream:
Expand Down Expand Up @@ -766,10 +791,7 @@ async def test_take(app):
break

assert event
# need one sleep on Python 3.6.0-3.6.6 + 3.7.0
# need two sleeps on Python 3.6.7 + 3.7.1 :-/
await asyncio.sleep(0)
await asyncio.sleep(0)
await wait_for_stream_ack(event)

if not event.ack.called:
assert event.message.acked
Expand Down Expand Up @@ -879,10 +901,7 @@ async def test_take_wit_timestamp(app):
break

assert event
# need one sleep on Python 3.6.0-3.6.6 + 3.7.0
# need two sleeps on Python 3.6.7 + 3.7.1 :-/
await asyncio.sleep(0)
await asyncio.sleep(0)
await wait_for_stream_ack(event)

if not event.ack.called:
assert event.message.acked
Expand All @@ -905,10 +924,7 @@ async def test_take_wit_timestamp_wit_simple_value(app):
break

assert event
# need one sleep on Python 3.6.0-3.6.6 + 3.7.0
# need two sleeps on Python 3.6.7 + 3.7.1 :-/
await asyncio.sleep(0)
await asyncio.sleep(0)
await wait_for_stream_ack(event)

if not event.ack.called:
assert event.message.acked
Expand All @@ -931,10 +947,7 @@ async def test_take_wit_timestamp_without_timestamp_field(app):
break

assert event
# need one sleep on Python 3.6.0-3.6.6 + 3.7.0
# need two sleeps on Python 3.6.7 + 3.7.1 :-/
await asyncio.sleep(0)
await asyncio.sleep(0)
await wait_for_stream_ack(event)

if not event.ack.called:
assert event.message.acked
Expand Down
Loading