Refactor time-based tests to use fake and sharp loop time

Utilise a new library — https://github.com/nolar/looptime — to have a fake time in event loops. The library does a lot of things, but the most important one is the sharp time with predictable steps with **no code overhead included**, which is typically random.

Everything else goes as a side-effect: e.g. the fast execution of such tests with zero real-time for any fake duration of loop time, convenient `looptime` fixture for assertions, etc.

Signed-off-by: Sergey Vasilyev <nolar@nolar.info>

Author: nolar

pyproject.toml

@@ -77,6 +77,7 @@ test = [
     "codecov",
     "coverage>=7.12.0",
     "freezegun",
+    "looptime>=0.7",
     "lxml",
     "pyngrok",
     "pytest>=9.0.0",
@@ -116,7 +117,7 @@ ignore_missing_imports = true
 minversion = "9.0"
 asyncio_mode = "auto"
 asyncio_default_fixture_loop_scope = "function"
-addopts = ["--strict-markers"]
+addopts = ["--strict-markers", "--looptime"]
 
 [tool.isort]
 line_length = 100

tests/apis/test_api_requests.py

@@ -138,23 +138,23 @@ async def test_parsing_in_streams(
     (delete, 'delete'),
 ])
 async def test_direct_timeout_in_requests(
-        resp_mocker, aresponses, hostname, fn, method, settings, logger, timer):
+        resp_mocker, aresponses, hostname, fn, method, settings, logger, looptime):
 
     async def serve_slowly():
-        await asyncio.sleep(1.0)
+        await asyncio.sleep(10)
         return aiohttp.web.json_response({})
 
     mock = resp_mocker(side_effect=serve_slowly)
     aresponses.add(hostname, '/url', method, mock)
 
-    with timer, pytest.raises(asyncio.TimeoutError):
-        timeout = aiohttp.ClientTimeout(total=0.1)
+    with pytest.raises(asyncio.TimeoutError):
+        timeout = aiohttp.ClientTimeout(total=1.23)
         # aiohttp raises an asyncio.TimeoutError which is automatically retried.
         # To reduce the test duration we disable retries for this test.
         settings.networking.error_backoffs = None
         await fn('/url', timeout=timeout, settings=settings, logger=logger)
 
-    assert 0.1 < timer.seconds < 0.2
+    assert looptime == 1.23
 
     # Let the server request finish and release all resources (tasks).
     # TODO: Remove when fixed: https://github.com/aio-libs/aiohttp/issues/7551
@@ -168,23 +168,23 @@ async def serve_slowly():
     (delete, 'delete'),
 ])
 async def test_settings_timeout_in_requests(
-        resp_mocker, aresponses, hostname, fn, method, settings, logger, timer):
+        resp_mocker, aresponses, hostname, fn, method, settings, logger, looptime):
 
     async def serve_slowly():
-        await asyncio.sleep(1.0)
+        await asyncio.sleep(10)
         return aiohttp.web.json_response({})
 
     mock = resp_mocker(side_effect=serve_slowly)
     aresponses.add(hostname, '/url', method, mock)
 
-    with timer, pytest.raises(asyncio.TimeoutError):
-        settings.networking.request_timeout = 0.1
+    with pytest.raises(asyncio.TimeoutError):
+        settings.networking.request_timeout = 1.23
         # aiohttp raises an asyncio.TimeoutError which is automatically retried.
         # To reduce the test duration we disable retries for this test.
         settings.networking.error_backoffs = None
         await fn('/url', settings=settings, logger=logger)
 
-    assert 0.1 < timer.seconds < 0.2
+    assert looptime == 1.23
 
     # Let the server request finish and release all resources (tasks).
     # TODO: Remove when fixed: https://github.com/aio-libs/aiohttp/issues/7551
@@ -193,24 +193,24 @@ async def serve_slowly():
 
 @pytest.mark.parametrize('method', ['get'])  # the only supported method at the moment
 async def test_direct_timeout_in_streams(
-        resp_mocker, aresponses, hostname, method, settings, logger, timer):
+        resp_mocker, aresponses, hostname, method, settings, logger, looptime):
 
     async def serve_slowly():
-        await asyncio.sleep(1.0)
+        await asyncio.sleep(10)
         return "{}"
 
     mock = resp_mocker(side_effect=serve_slowly)
     aresponses.add(hostname, '/url', method, mock)
 
-    with timer, pytest.raises(asyncio.TimeoutError):
-        timeout = aiohttp.ClientTimeout(total=0.1)
+    with pytest.raises(asyncio.TimeoutError):
+        timeout = aiohttp.ClientTimeout(total=1.23)
         # aiohttp raises an asyncio.TimeoutError which is automatically retried.
         # To reduce the test duration we disable retries for this test.
         settings.networking.error_backoffs = None
         async for _ in stream('/url', timeout=timeout, settings=settings, logger=logger):
             pass
 
-    assert 0.1 < timer.seconds < 0.2
+    assert looptime == 1.23
 
     # Let the server request finish and release all resources (tasks).
     # TODO: Remove when fixed: https://github.com/aio-libs/aiohttp/issues/7551
@@ -219,46 +219,47 @@ async def serve_slowly():
 
 @pytest.mark.parametrize('method', ['get'])  # the only supported method at the moment
 async def test_settings_timeout_in_streams(
-        resp_mocker, aresponses, hostname, method, settings, logger, timer):
+        resp_mocker, aresponses, hostname, method, settings, logger, looptime):
 
     async def serve_slowly():
-        await asyncio.sleep(1.0)
+        await asyncio.sleep(10)
         return "{}"
 
     mock = resp_mocker(side_effect=serve_slowly)
     aresponses.add(hostname, '/url', method, mock)
 
-    with timer, pytest.raises(asyncio.TimeoutError):
-        settings.networking.request_timeout = 0.1
+    with pytest.raises(asyncio.TimeoutError):
+        settings.networking.request_timeout = 1.23
         # aiohttp raises an asyncio.TimeoutError which is automatically retried.
         # To reduce the test duration we disable retries for this test.
         settings.networking.error_backoffs = None
         async for _ in stream('/url', settings=settings, logger=logger):
             pass
 
-    assert 0.1 < timer.seconds < 0.2
+    assert looptime == 1.23
 
     # Let the server request finish and release all resources (tasks).
     # TODO: Remove when fixed: https://github.com/aio-libs/aiohttp/issues/7551
     await asyncio.sleep(1.0)
 
 
-@pytest.mark.parametrize('delay, expected', [
-    pytest.param(0.0, [], id='instant-none'),
-    pytest.param(0.1, [{'fake': 'result1'}], id='fast-single'),
-    pytest.param(9.9, [{'fake': 'result1'}, {'fake': 'result2'}], id='inf-double'),
+@pytest.mark.parametrize('delay, expected_times, expected_items', [
+    pytest.param(0, [], [], id='instant-none'),
+    pytest.param(2, [1], [{'fake': 'result1'}], id='fast-single'),
+    pytest.param(9, [1, 4], [{'fake': 'result1'}, {'fake': 'result2'}], id='inf-double'),
 ])
 @pytest.mark.parametrize('method', ['get'])  # the only supported method at the moment
 async def test_stopper_in_streams(
-        resp_mocker, aresponses, hostname, method, delay, expected, settings, logger):
+        resp_mocker, aresponses, hostname, method, delay, settings, logger, looptime,
+        expected_items, expected_times):
 
     async def stream_slowly(request: aiohttp.web.Request) -> aiohttp.web.StreamResponse:
         response = aiohttp.web.StreamResponse()
         await response.prepare(request)
         try:
-            await asyncio.sleep(0.05)
+            await asyncio.sleep(1)
             await response.write(b'{"fake": "result1"}\n')
-            await asyncio.sleep(0.15)
+            await asyncio.sleep(3)
             await response.write(b'{"fake": "result2"}\n')
             await response.write_eof()
         except ConnectionError:
@@ -271,9 +272,13 @@ async def stream_slowly(request: aiohttp.web.Request) -> aiohttp.web.StreamRespo
     asyncio.get_running_loop().call_later(delay, stopper.set_result, None)
 
     items = []
+    times = []
     async for item in stream('/url', stopper=stopper, settings=settings, logger=logger):
         items.append(item)
+        times.append(float(looptime))
 
-    assert items == expected
+    assert items == expected_items
+    assert times == expected_times
 
-    await asyncio.sleep(0.2)  # give the response some time to be cancelled and its tasks closed
+    # Give the response some time to be cancelled and its tasks closed. That is aiohttp's issue.
+    await asyncio.sleep(30)

tests/conftest.py

@@ -12,6 +12,7 @@
 
 import aiohttp.web
 import pytest
+import pytest_asyncio
 
 import kopf
 from kopf._cogs.clients.auth import APIContext
@@ -207,7 +208,6 @@ class K8sMocks:
     patch: Mock
     delete: Mock
     stream: Mock
-    sleep: Mock
 
 
 @pytest.fixture()
@@ -226,7 +226,6 @@ async def itr(*_, **__):
         patch=mocker.patch('kopf._cogs.clients.api.patch', return_value={}),
         delete=mocker.patch('kopf._cogs.clients.api.delete', return_value={}),
         stream=mocker.patch('kopf._cogs.clients.api.stream', side_effect=itr),
-        sleep=mocker.patch('kopf._cogs.aiokits.aiotime.sleep', return_value=None),
     )
 
 
@@ -566,69 +565,6 @@ def no_certvalidator():
     yield from _with_module_absent('certvalidator')
 
 
-#
-# Helpers for the timing checks.
-#
-
-@pytest.fixture()
-def timer():
-    return Timer()
-
-
-class Timer:
-    """
-    A helper context manager to measure the time of the code-blocks.
-    Also, supports direct comparison with time-deltas and the numbers of seconds.
-
-    Usage:
-
-        with Timer() as timer:
-            do_something()
-            print(f"Executing for {timer.seconds}s already.")
-            do_something_else()
-
-        print(f"Executed in {timer.seconds}s.")
-        assert timer < 5.0
-    """
-
-    def __init__(self):
-        super().__init__()
-        self._ts = None
-        self._te = None
-
-    @property
-    def seconds(self):
-        if self._ts is None:
-            return None
-        elif self._te is None:
-            return time.perf_counter() - self._ts
-        else:
-            return self._te - self._ts
-
-    def __repr__(self):
-        status = 'new' if self._ts is None else 'running' if self._te is None else 'finished'
-        return f'<Timer: {self.seconds}s ({status})>'
-
-    def __enter__(self):
-        self._ts = time.perf_counter()
-        self._te = None
-        return self
-
-    def __exit__(self, exc_type, exc_val, exc_tb):
-        self._te = time.perf_counter()
-
-    async def __aenter__(self):
-        return self.__enter__()
-
-    async def __aexit__(self, exc_type, exc_val, exc_tb):
-        return self.__exit__(exc_type, exc_val, exc_tb)
-
-    def __int__(self):
-        return int(self.seconds)
-
-    def __float__(self):
-        return float(self.seconds)
-
 #
 # Helpers for the logging checks.
 #
@@ -708,13 +644,8 @@ def assert_logs_fn(patterns, prohibited=[], strict=False):
 #
 # Helpers for asyncio checks.
 #
-@pytest.fixture()
-async def loop():
-    yield asyncio.get_running_loop()
-
-
-@pytest.fixture(autouse=True)
-def _no_asyncio_pending_tasks(loop: asyncio.AbstractEventLoop):
+@pytest_asyncio.fixture(autouse=True)
+def _no_asyncio_pending_tasks(request: pytest.FixtureRequest):
     """
     Ensure there are no unattended asyncio tasks after the test.
 
@@ -735,7 +666,28 @@ def _no_asyncio_pending_tasks(loop: asyncio.AbstractEventLoop):
 
     # Let the pytest-asyncio's async2sync wrapper to finish all callbacks. Otherwise, it raises:
     #   <Task pending name='Task-2' coro=<<async_generator_athrow without __name__>()>>
-    loop.run_until_complete(asyncio.sleep(0))
+    # We don't know which loops were used in the test & fixtures, so we wait on all of them.
+    for fixture_name, fixture_value in request.node.funcargs.items():
+        if isinstance(fixture_value, asyncio.BaseEventLoop):
+            fixture_value.run_until_complete(asyncio.sleep(0))
+
+        # Safe-guards for Python 3.10 until deprecated in ≈Oct'2026 (not needed for 3.11+).
+        try:
+            from asyncio import Runner as stdlib_Runner  # python >= 3.11 (absent in 3.10)
+        except ImportError:
+            pass
+        else:
+            if isinstance(fixture_value, stdlib_Runner):
+                fixture_value.get_loop().run_until_complete(asyncio.sleep(0))
+
+        # In case pytest's asyncio libraries use the backported runners in Python 3.10.
+        try:
+            from backports.asyncio.runner import Runner as backported_Runner
+        except ImportError:
+            pass
+        else:
+            if isinstance(fixture_value, backported_Runner):
+                fixture_value.get_loop().run_until_complete(asyncio.sleep(0))
 
     # Detect all leftover tasks.
     after = _get_all_tasks()
@@ -760,3 +712,9 @@ def _get_all_tasks() -> set[asyncio.Task]:
         else:
             break
     return {t for t in tasks if not t.done()}
+
+
+@pytest.fixture()
+def loop():
+    """Sync aiohttp's server-side timeline with kopf's client-side timeline."""
+    return asyncio.get_running_loop()