Make the timing tests independent of sleep accuracy

_python_utils_tests/clock.py

@@ -0,0 +1,56 @@
+"""A fake clock for the tests and doctests of ``python_utils.time``."""
+
+import typing
+
+import pytest
+
+import python_utils.time
+
+#: The doctest that sleeps on the clock, by the name pytest gives it.
+TIMEOUT_GENERATOR_DOCTEST: str = 'python_utils.time.timeout_generator'
+
+
+class FakeClock:
+    """
+    A clock that only moves when something sleeps on it.
+
+    ``time.sleep`` promises to sleep at least as long as requested. A busy
+    machine sleeps tens of milliseconds longer, and that changes how many
+    items ``timeout_generator`` yields before its timeout. On this clock the
+    number of items depends on the arguments alone.
+
+    Attributes:
+        now (float): The current time in seconds.
+        sleeps (list[float]): Every requested sleep, in order.
+    """
+
+    def __init__(self) -> None:
+        """Start at zero without any recorded sleeps."""
+        self.now: float = 0.0
+        self.sleeps: list[float] = []
+
+    def perf_counter(self) -> float:
+        """Return the current time, like ``time.perf_counter``."""
+        return self.now
+
+    def sleep(self, seconds: float) -> None:
+        """Record the sleep and move the clock forward, without waiting."""
+        self.sleeps.append(seconds)
+        self.now += seconds
+
+
+@pytest.fixture
+def fake_clock(monkeypatch: pytest.MonkeyPatch) -> FakeClock:
+    """Replace the ``time`` module inside ``python_utils.time``."""
+    clock: FakeClock = FakeClock()
+    monkeypatch.setattr(python_utils.time, 'time', clock)
+    return clock
+
+
+@pytest.fixture(autouse=True)
+def fake_clock_in_doctest(request: pytest.FixtureRequest) -> None:
+    """Run the ``timeout_generator`` doctest on the fake clock."""
+    # pytest leaves `FixtureRequest.node` without a type.
+    node: pytest.Item = typing.cast(pytest.Item, request.node)
+    if node.name == TIMEOUT_GENERATOR_DOCTEST:
+        request.getfixturevalue('fake_clock')

_python_utils_tests/test_lazy_imports.py

@@ -11,6 +11,7 @@
 import pytest
 
 import python_utils
+from _python_utils_tests import clock
 
 
 def _run_clean(code: str) -> subprocess.CompletedProcess[str]:
@@ -94,10 +95,13 @@ def test_star_import_resolves_all_names() -> None:
 
 
 @pytest.mark.asyncio
-async def test_aio_timeout_generator_default_iterable() -> None:
+async def test_aio_timeout_generator_default_iterable(
+    fake_clock: clock.FakeClock,
+) -> None:
     """Default the iterable to ``aio.acount`` when omitted."""
     # With no iterable the generator defaults to ``aio.acount`` -- exercising
     # the lazy ``aio``/``asyncio`` import and the None-resolution branch.
+    # The fake clock stands still, so the timeout cannot end the loop early.
     count = 0
     generator: collections.abc.AsyncGenerator[object, None] = (
         python_utils.aio_timeout_generator(timeout=0.05, interval=0.0)

_python_utils_tests/test_time.py

@@ -3,12 +3,18 @@
 import asyncio
 import datetime
 import itertools
+import time
 
 import pytest
 
 import python_utils
+from _python_utils_tests import clock
 from python_utils import types
 
+#: Far longer than every timeout in this module, so a generator that sleeps
+#: this long is always interrupted first.
+STALL: float = 10.0
+
 
 @pytest.mark.parametrize(
     'timeout,interval,interval_multiplier,maximum_interval,iterable,result',
@@ -28,6 +34,8 @@
 )
 @pytest.mark.asyncio
 async def test_aio_timeout_generator(
+    fake_clock: clock.FakeClock,
+    monkeypatch: pytest.MonkeyPatch,
     timeout: float,
     interval: float,
     interval_multiplier: float,
@@ -36,6 +44,13 @@ async def test_aio_timeout_generator(
     result: int,
 ) -> None:
     """Stop the async generator near the configured timeout."""
+
+    async def sleep(delay: float) -> None:
+        """Let the fake clock pass the delay without waiting for it."""
+        fake_clock.sleep(delay)
+
+    monkeypatch.setattr(asyncio, 'sleep', sleep)
+
     i = None
     async for i in python_utils.aio_timeout_generator(
         timeout, interval, iterable, maximum_interval=maximum_interval
@@ -46,23 +61,26 @@ async def test_aio_timeout_generator(
 
 
 @pytest.mark.parametrize(
-    'timeout,interval,interval_multiplier,maximum_interval,iterable,result',
+    'timeout,interval,interval_multiplier,maximum_interval,iterable,result,'
+    'sleeps',
     [
-        (0.1, 0.06, 0.5, 0.1, 'abc', 'c'),
-        (0.1, 0.07, 0.5, 0.1, itertools.count, 2),
-        (0.1, 0.07, 0.5, 0.1, itertools.count(), 2),
-        (0.1, 0.06, 1.0, None, 'abc', 'c'),
+        (0.1, 0.06, 0.5, 0.1, 'abc', 'c', [0.06, 0.03, 0.015]),
+        (0.1, 0.07, 0.5, 0.1, itertools.count, 2, [0.07, 0.035]),
+        (0.1, 0.07, 0.5, 0.1, itertools.count(), 2, [0.07, 0.035]),
+        (0.1, 0.06, 1.0, None, 'abc', 'c', [0.06, 0.06]),
         (
             datetime.timedelta(seconds=0.1),
             datetime.timedelta(seconds=0.06),
             2.0,
             datetime.timedelta(seconds=0.1),
             itertools.count,
             2,
+            [0.06, 0.1],
         ),
     ],
 )
 def test_timeout_generator(
+    fake_clock: clock.FakeClock,
     timeout: float,
     interval: float,
     interval_multiplier: float,
@@ -73,8 +91,9 @@ def test_timeout_generator(
         types.Callable[..., types.Iterable[types.Any]],
     ],
     result: int,
+    sleeps: types.List[float],
 ) -> None:
-    """Stop the sync generator near the configured timeout."""
+    """Stop the sync generator at the timeout and scale the interval."""
     i = None
     for i in python_utils.timeout_generator(
         timeout=timeout,
@@ -86,59 +105,97 @@ def test_timeout_generator(
         assert i is not None
 
     assert i == result
+    assert fake_clock.sleeps == pytest.approx(sleeps)
 
 
-@pytest.mark.asyncio
-async def test_aio_generator_timeout_detector() -> None:
-    """Raise or exit on per-item and total timeouts."""
-    # Make pyright happy
-    i = None
+def test_timeout_generator_real_clock() -> None:
+    """Keep yielding on the real clock until the timeout has passed."""
+    timeout: float = 0.05
+    interval: float = 0.01
+    start: float = time.perf_counter()
+    items: types.List[int] = list(
+        python_utils.timeout_generator(timeout, interval, itertools.count())
+    )
+    elapsed: float = time.perf_counter() - start
+
+    # A sleep can take longer than requested but never shorter, so these
+    # hold on any machine. The exact number of items does not.
+    assert items == list(range(len(items)))
+    assert len(items) <= timeout / interval + 2
+    assert elapsed >= timeout
+
+
+async def stalling_generator() -> types.AsyncGenerator[int, None]:
+    """Yield 0-4 without waiting, then stall before the next item."""
+    for i in range(10):
+        if i == 5:
+            await asyncio.sleep(STALL)
+        yield i
+
+
+def ticking_generator(
+    fake_clock: clock.FakeClock,
+) -> types.AsyncGenerator[int, None]:
+    """Yield 0-9 and let 0.1 seconds pass on the fake clock for each item."""
 
     async def generator() -> types.AsyncGenerator[int, None]:
-        """Yield 0-9 with increasing sleeps between items."""
+        """Advance the fake clock before every item."""
         for i in range(10):
-            await asyncio.sleep(i / 20.0)
+            fake_clock.sleep(0.1)
             yield i
 
+    return generator()
+
+
+@pytest.mark.asyncio
+async def test_aio_generator_timeout_detector(
+    fake_clock: clock.FakeClock,
+) -> None:
+    """Raise or exit on per-item and total timeouts."""
+    # Make pyright happy
+    i = None
+
     detector = python_utils.aio_generator_timeout_detector
     # Test regular timeout with reraise
     with pytest.raises(asyncio.TimeoutError):
-        async for i in detector(generator(), 0.25):
+        async for i in detector(stalling_generator(), 0.05):
             pass
 
     # Test regular timeout with clean exit
-    async for i in detector(generator(), 0.25, on_timeout=None):
+    async for i in detector(stalling_generator(), 0.05, on_timeout=None):
         pass
 
     assert i == 4
 
     # Test total timeout with reraise
     with pytest.raises(asyncio.TimeoutError):
-        async for i in detector(generator(), total_timeout=0.5):
+        async for i in detector(
+            ticking_generator(fake_clock), total_timeout=0.45
+        ):
             pass
 
     # Test total timeout with clean exit
-    async for i in detector(generator(), total_timeout=0.5, on_timeout=None):
+    async for i in detector(
+        ticking_generator(fake_clock), total_timeout=0.45, on_timeout=None
+    ):
         pass
 
     assert i == 4
 
     # Test stop iteration
-    async for i in detector(generator(), on_timeout=None):
+    async for i in detector(ticking_generator(fake_clock), on_timeout=None):
         pass
 
+    assert i == 9
+
 
 @pytest.mark.asyncio
 async def test_aio_generator_timeout_detector_decorator_reraise() -> None:
     """Reraise ``TimeoutError`` on a per-item timeout."""
-
     # Test regular timeout with reraise
-    @python_utils.aio_generator_timeout_detector_decorator(timeout=0.05)
-    async def generator_timeout() -> types.AsyncGenerator[int, None]:
-        """Yield with increasing delays to trip the timeout."""
-        for i in range(10):
-            await asyncio.sleep(i / 100.0)
-            yield i
+    generator_timeout = python_utils.aio_generator_timeout_detector_decorator(
+        timeout=0.05
+    )(stalling_generator)
 
     with pytest.raises(asyncio.TimeoutError):
         async for _ in generator_timeout():
@@ -152,14 +209,9 @@ async def test_aio_generator_timeout_detector_decorator_clean_exit() -> None:
     i = None
 
     # Test regular timeout with clean exit
-    @python_utils.aio_generator_timeout_detector_decorator(
+    generator_clean = python_utils.aio_generator_timeout_detector_decorator(
         timeout=0.05, on_timeout=None
-    )
-    async def generator_clean() -> types.AsyncGenerator[int, None]:
-        """Yield with increasing delays to trip the timeout."""
-        for i in range(10):
-            await asyncio.sleep(i / 100.0)
-            yield i
+    )(stalling_generator)
 
     async for i in generator_clean():
         pass
@@ -168,17 +220,16 @@ async def generator_clean() -> types.AsyncGenerator[int, None]:
 
 
 @pytest.mark.asyncio
-async def test_aio_generator_timeout_detector_decorator_reraise_total() -> (
-    None
-):
+async def test_aio_generator_timeout_detector_decorator_reraise_total(
+    fake_clock: clock.FakeClock,
+) -> None:
     """Reraise ``TimeoutError`` on a total timeout."""
 
     # Test total timeout with reraise
-    @python_utils.aio_generator_timeout_detector_decorator(total_timeout=0.1)
+    @python_utils.aio_generator_timeout_detector_decorator(total_timeout=0.45)
     async def generator_reraise() -> types.AsyncGenerator[int, None]:
-        """Yield with increasing delays to trip the timeout."""
-        for i in range(10):
-            await asyncio.sleep(i / 100.0)
+        """Let the fake clock pass the total timeout while yielding."""
+        async for i in ticking_generator(fake_clock):
             yield i
 
     with pytest.raises(asyncio.TimeoutError):
@@ -187,19 +238,20 @@ async def generator_reraise() -> types.AsyncGenerator[int, None]:
 
 
 @pytest.mark.asyncio
-async def test_aio_generator_timeout_detector_decorator_clean_total() -> None:
+async def test_aio_generator_timeout_detector_decorator_clean_total(
+    fake_clock: clock.FakeClock,
+) -> None:
     """Exit cleanly on total timeout when ``on_timeout`` is ``None``."""
     # Make pyright happy
     i = None
 
     # Test total timeout with clean exit
     @python_utils.aio_generator_timeout_detector_decorator(
-        total_timeout=0.1, on_timeout=None
+        total_timeout=0.45, on_timeout=None
     )
     async def generator_clean_total() -> types.AsyncGenerator[int, None]:
-        """Yield with increasing delays to trip the timeout."""
-        for i in range(10):
-            await asyncio.sleep(i / 100.0)
+        """Let the fake clock pass the total timeout while yielding."""
+        async for i in ticking_generator(fake_clock):
             yield i
 
     async for i in generator_clean_total():

conftest.py

@@ -0,0 +1,8 @@
+"""
+Load the shared fixtures for the tests and the doctests.
+
+The fixtures live in the tests package. They are loaded from the repository
+root because the doctests in ``python_utils`` need them as well.
+"""
+
+pytest_plugins: tuple[str, ...] = ('_python_utils_tests.clock',)

pyproject.toml

@@ -8,7 +8,7 @@ module-root = ''
 module-name = 'python_utils'
 # Keep the tests and tox config in the sdist (parity with the old MANIFEST.in)
 # so downstream packagers can build and test from the source distribution.
-source-include = ['_python_utils_tests/**/*.py', 'tox.ini']
+source-include = ['_python_utils_tests/**/*.py', 'conftest.py', 'tox.ini']
 
 [project]
 name = 'python-utils'