From 7a2117e2a8d80fe35bbad2c25e2dd4be9af0c903 Mon Sep 17 00:00:00 2001 From: Thor Whalen <1906276+thorwhalen@users.noreply.github.com> Date: Tue, 20 Oct 2026 07:22:17 +0000 Subject: [PATCH] Tests: poll for data with a deadline instead of fixed sleeps (#28) Co-Authored-By: Claude Sonnet 5.5 --- stream2py/tests/test_buffer_reader_blocking.py | 6 +++--- .../tests/test_buffer_reader_stopiteration.py | 15 ++++----------- stream2py/tests/test_reader_without_context.py | 8 +++----- stream2py/tests/test_stream_buffer.py | 9 ++++++--- stream2py/tests/test_stream_source.py | 9 ++++++--- stream2py/tests/utils_for_testing.py | 15 +++++++++++++++ 6 files changed, 37 insertions(+), 25 deletions(-) diff --git a/stream2py/tests/test_buffer_reader_blocking.py b/stream2py/tests/test_buffer_reader_blocking.py index a9771de..f572994 100644 --- a/stream2py/tests/test_buffer_reader_blocking.py +++ b/stream2py/tests/test_buffer_reader_blocking.py @@ -2,7 +2,7 @@ import time import threading from stream2py import StreamBuffer -from stream2py.tests.utils_for_testing import SimpleSourceReader +from stream2py.tests.utils_for_testing import SimpleSourceReader, wait_until from itertools import count @@ -15,7 +15,7 @@ def test_buffer_reader_blocking(): reader = buffer.mk_reader() # Wait a moment for some data to be available - time.sleep(0.2) + assert wait_until(lambda: len(buffer.source_buffer) >= 6) # Non-blocking read should return data immediately when available result = reader.read(blocking=False, ignore_no_item_found=True) @@ -47,7 +47,7 @@ def test_buffer_reader_blocking_stops_on_stop_event(): reader = buffer.mk_reader() # Read all available data - time.sleep(0.1) + assert wait_until(lambda: len(buffer.source_buffer) >= 5) while reader.read(blocking=False, ignore_no_item_found=True) is not None: pass diff --git a/stream2py/tests/test_buffer_reader_stopiteration.py b/stream2py/tests/test_buffer_reader_stopiteration.py index 163ff3a..d06b5e8 100644 --- a/stream2py/tests/test_buffer_reader_stopiteration.py +++ b/stream2py/tests/test_buffer_reader_stopiteration.py @@ -1,7 +1,7 @@ """Tests for BufferReader StopIteration behavior (Issue #10)""" import pytest from stream2py import StreamBuffer -from stream2py.tests.utils_for_testing import SimpleSourceReader +from stream2py.tests.utils_for_testing import SimpleSourceReader, wait_until def test_next_raises_stopiteration_when_stopped(): @@ -38,10 +38,7 @@ def test_next_returns_none_when_no_data_but_running(): reader = buffer.mk_reader() # Read all available data quickly - import time - deadline = time.time() + 5 # wait for data rather than hope 0.1s is enough on a loaded runner - while not len(buffer.source_buffer) and time.time() < deadline: - time.sleep(0.01) + assert wait_until(lambda: len(buffer.source_buffer) >= 1) # Read until we get None (no more data available yet) items = [] @@ -65,8 +62,7 @@ def test_builtin_next_with_default(): buffer.start() reader = buffer.mk_reader() - import time - time.sleep(0.1) + assert wait_until(lambda: len(buffer.source_buffer) >= 2) # Read items item1 = next(reader, 'default') @@ -80,8 +76,6 @@ def test_builtin_next_with_default(): # When stopped and no data, should raise StopIteration # and the default should be returned by builtin next() - time.sleep(0.1) - # Consume any remaining data while True: try: @@ -104,8 +98,7 @@ def test_for_loop_stops_correctly(): buffer = StreamBuffer(source, maxlen=100) buffer.start() - import time - time.sleep(0.2) # Let all data load + assert wait_until(lambda: len(buffer.source_buffer) >= 10) # all data loaded buffer.stop() # Stop before iterating diff --git a/stream2py/tests/test_reader_without_context.py b/stream2py/tests/test_reader_without_context.py index 24971cb..30e1e2b 100644 --- a/stream2py/tests/test_reader_without_context.py +++ b/stream2py/tests/test_reader_without_context.py @@ -1,7 +1,7 @@ """Tests for using readers without context manager (Issue #18)""" import pytest from stream2py import StreamBuffer -from stream2py.tests.utils_for_testing import SimpleSourceReader +from stream2py.tests.utils_for_testing import SimpleSourceReader, wait_until from stream2py.exceptions import StreamNotStartedError @@ -26,8 +26,7 @@ def test_reader_works_without_context_manager(): reader = buffer.mk_reader() # Reading should work fine without context manager - import time - time.sleep(0.1) # Let some data accumulate + assert wait_until(lambda: len(buffer.source_buffer) >= 2) # some data accumulated item = reader.read(ignore_no_item_found=True) assert item is not None @@ -84,8 +83,7 @@ def test_stream_buffer_with_context(): with StreamBuffer(source, maxlen=100) as buffer: reader = buffer.mk_reader() - import time - time.sleep(0.1) + assert wait_until(lambda: len(buffer.source_buffer) >= 1) item = reader.read(ignore_no_item_found=True) assert item is not None diff --git a/stream2py/tests/test_stream_buffer.py b/stream2py/tests/test_stream_buffer.py index 671c46c..1628377 100644 --- a/stream2py/tests/test_stream_buffer.py +++ b/stream2py/tests/test_stream_buffer.py @@ -1,4 +1,7 @@ -from stream2py.tests.utils_for_testing import TenthSecondCounterStreamSource +from stream2py.tests.utils_for_testing import ( + TenthSecondCounterStreamSource, + wait_until, +) from stream2py.stream_buffer import StreamBuffer import time @@ -12,7 +15,7 @@ def test_stream_buffer(): sc_reader = ( sc_buf.mk_reader() ) # reader must be made after start to have data related to said start - time.sleep(2) + assert wait_until(lambda: len(sc_buf.source_buffer) >= 20) # ~2s of data assert sc_reader.source_reader_info['bt'] != 0, sc_reader.source_reader_info print('data as shown is (timestamp, count) at every tenth of a second') while True: @@ -87,7 +90,7 @@ def test_stream_buffer(): ), 'new readers should now have the same last_item cursor position' # print some data from old and new reader and see they are different - time.sleep(1.5) + assert wait_until(lambda: len(sc_buf.source_buffer) >= 3) print('comparing first start reader values to second start') while True: r22 = sc_reader22.read(n=1, ignore_no_item_found=True) diff --git a/stream2py/tests/test_stream_source.py b/stream2py/tests/test_stream_source.py index 9451f3d..7f076e2 100644 --- a/stream2py/tests/test_stream_source.py +++ b/stream2py/tests/test_stream_source.py @@ -1,4 +1,7 @@ -from stream2py.tests.utils_for_testing import TenthSecondCounterStreamSource +from stream2py.tests.utils_for_testing import ( + TenthSecondCounterStreamSource, + wait_until, +) import time @@ -6,7 +9,7 @@ def test_stream_source(): source = TenthSecondCounterStreamSource() # source with source.open_reader() as sc_reader: sc_buf = source._stream_buffer - time.sleep(2) + assert wait_until(lambda: len(sc_buf.source_buffer) >= 20) # ~2s of data assert source.info['bt'] != 0, source.info print('data as shown is (timestamp, count) at every tenth of a second') while True: @@ -85,7 +88,7 @@ def test_stream_source(): ), 'new readers should now have the same last_item cursor position' # print some data from old and new reader and see they are different - time.sleep(1.5) + assert wait_until(lambda: len(sc_buf.source_buffer) >= 3) print('comparing first start reader values to second start') while True: r22 = sc_reader22.read(n=1, ignore_no_item_found=True) diff --git a/stream2py/tests/utils_for_testing.py b/stream2py/tests/utils_for_testing.py index c16a42a..40f1c8d 100644 --- a/stream2py/tests/utils_for_testing.py +++ b/stream2py/tests/utils_for_testing.py @@ -271,3 +271,18 @@ def read(self): return None key = operator.itemgetter(0) # (timestamp_us, count) -> timestamp_us + + +def wait_until(condition, *, timeout=10.0, interval=0.01): + """Poll ``condition()`` until it is truthy or ``timeout`` seconds pass. + + Use instead of a fixed ``time.sleep`` before asserting on data produced by a + background thread: a fixed sleep is too short on a loaded (or single-core) runner. + Returns the final truthiness of ``condition()`` so callers can ``assert`` it. + """ + deadline = time.monotonic() + timeout + while not condition(): + if time.monotonic() >= deadline: + return bool(condition()) + time.sleep(interval) + return True