test: deflake recurring task test and fix tests hanging on Python 3.10

pyproject.toml

@@ -130,7 +130,10 @@ dev = [
     "pre-commit<5.0.0",
     "proxy-py<3.0.0",
     "pydoc-markdown<5.0.0",
-    "pytest-asyncio<2.0.0",
+    # pytest-asyncio 1.1+ runs tests through `backports.asyncio.runner` on Python 3.10, whose process-wide
+    # `contextvars.copy_context` patch can leak the fixture context into the test server thread and hang the test.
+    "pytest-asyncio<2.0.0; python_version >= '3.11'",
+    "pytest-asyncio>=1.0.0,<1.1.0; python_version < '3.11'",
     "pytest-cov<8.0.0",
     "pytest-rerunfailures<17.0.0",
     "pytest-timeout<3.0.0",
@@ -267,6 +270,11 @@ addopts = "-r a --verbose --dist worksteal"
 asyncio_default_fixture_loop_scope = "function"
 asyncio_mode = "auto"
 timeout = 1800
+# A thread timer kills the stuck worker even when its main thread is blocked and a SIGALRM handler can never run.
+timeout_method = "thread"
+# pytest-timeout writes its stack dump to the xdist worker's stdout, which never reaches the console. faulthandler
+# writes straight to the stderr file descriptor, so a test stuck this long leaves its stacks in the log.
+faulthandler_timeout = 600
 markers = [
     "run_alone: marks tests that must run in isolation",
 ]

tests/unit/_utils/test_recurring_task.py

@@ -42,18 +42,17 @@ async def test_start_and_stop(function: AsyncMock, delay: timedelta) -> None:
     assert rt.task.done()
 
 
-@pytest.mark.run_alone
 async def test_execution(function: AsyncMock, delay: timedelta) -> None:
+    """The recurring task calls its function repeatedly."""
+    third_call_done = asyncio.Event()
+    function.side_effect = lambda: third_call_done.set() if function.call_count >= 3 else None
     task = RecurringTask(function, delay)
 
     task.start()
-    await asyncio.sleep(0.2)  # Wait enough for the task to execute a few times
+    await asyncio.wait_for(third_call_done.wait(), timeout=5)
     await task.stop()
 
-    assert isinstance(task.func, AsyncMock)  # To let type checker know that the function is a mock
-    assert task.func.call_count >= 3
-
-    await task.stop()
+    assert function.call_count >= 3
 
 
 async def test_execution_continues_after_exception(delay: timedelta, caplog: pytest.LogCaptureFixture) -> None:

tests/unit/storages/test_dataset.py

@@ -930,7 +930,7 @@ async def test_purge_on_start_enabled(storage_client: StorageClient) -> None:
 
     # Skip this test for memory storage since it doesn't persist data between client instances.
     if isinstance(storage_client, MemoryStorageClient):
-        pytest.skip('Memory storage does not persist data between client instances.')
+        raise pytest.skip.Exception('Memory storage does not persist data between client instances.')
 
     configuration = Configuration(purge_on_start=True)
 
@@ -1016,7 +1016,7 @@ async def test_purge_on_start_disabled(storage_client: StorageClient) -> None:
 
     # Skip this test for memory storage since it doesn't persist data between client instances.
     if isinstance(storage_client, MemoryStorageClient):
-        pytest.skip('Memory storage does not persist data between client instances.')
+        raise pytest.skip.Exception('Memory storage does not persist data between client instances.')
 
     configuration = Configuration(purge_on_start=False)
 

tests/unit/storages/test_key_value_store.py

@@ -994,7 +994,7 @@ async def test_purge_on_start_enabled(storage_client: StorageClient) -> None:
 
     # Skip this test for memory storage since it doesn't persist data between client instances.
     if isinstance(storage_client, MemoryStorageClient):
-        pytest.skip('Memory storage does not persist data between client instances.')
+        raise pytest.skip.Exception('Memory storage does not persist data between client instances.')
 
     configuration = Configuration(purge_on_start=True)
 
@@ -1080,7 +1080,7 @@ async def test_purge_on_start_disabled(storage_client: StorageClient) -> None:
 
     # Skip this test for memory storage since it doesn't persist data between client instances.
     if isinstance(storage_client, MemoryStorageClient):
-        pytest.skip('Memory storage does not persist data between client instances.')
+        raise pytest.skip.Exception('Memory storage does not persist data between client instances.')
 
     configuration = Configuration(purge_on_start=False)
 

tests/unit/storages/test_request_queue.py

@@ -1279,7 +1279,7 @@ async def test_purge_on_start_enabled(storage_client: StorageClient) -> None:
 
     # Skip this test for memory storage since it doesn't persist data between client instances.
     if isinstance(storage_client, MemoryStorageClient):
-        pytest.skip('Memory storage does not persist data between client instances.')
+        raise pytest.skip.Exception('Memory storage does not persist data between client instances.')
 
     configuration = Configuration(purge_on_start=True)
 
@@ -1405,7 +1405,7 @@ async def test_purge_on_start_disabled(storage_client: StorageClient) -> None:
 
     # Skip this test for memory storage since it doesn't persist data between client instances.
     if isinstance(storage_client, MemoryStorageClient):
-        pytest.skip('Memory storage does not persist data between client instances.')
+        raise pytest.skip.Exception('Memory storage does not persist data between client instances.')
 
     configuration = Configuration(purge_on_start=False)