diff --git a/tests/functional/test_streams.py b/tests/functional/test_streams.py
index 851594155..521c99378 100644
--- a/tests/functional/test_streams.py
+++ b/tests/functional/test_streams.py
@@ -1,4 +1,5 @@
 import asyncio
+import gc
 import platform
 from copy import copy
 from unittest.mock import Mock, patch
@@ -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:
@@ -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
@@ -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
@@ -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
@@ -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