Tests: poll with deadline instead of fixed sleeps (#28)

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
 

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
 

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

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)

stream2py/tests/test_stream_source.py

@@ -1,12 +1,15 @@
-from stream2py.tests.utils_for_testing import TenthSecondCounterStreamSource
+from stream2py.tests.utils_for_testing import (
+    TenthSecondCounterStreamSource,
+    wait_until,
+)
 import time
 
 
 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)

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