Fix the asyncio guards to expect changes during the task lising

The WeakSet can change while it is iterated and being converted to a list (this seems fast, but it is not always the case and tends to be reproducible in the same setup). This fix guards against that and tries to convert the WeakSet to a list 1000 times before giving up — the same as Python's asyncio does.

Signed-off-by: Sergey Vasilyev <[email protected]>

tests/conftest.py

@@ -7,6 +7,7 @@
 import re
 import sys
 import time
+from typing import Set
 from unittest.mock import Mock
 
 import aiohttp.web
@@ -700,9 +701,7 @@ def _no_asyncio_pending_tasks(event_loop):
     collection after every test, and check messages from `asyncio.Task.__del__`.
     This, however, requires intercepting all event-loop creation in the code.
     """
-
-    # See `asyncio.all_tasks()` implementation for reference.
-    before = {t for t in list(asyncio.tasks._all_tasks) if not t.done()}
+    before = _get_all_tasks()
 
     # Run the test.
     yield
@@ -712,7 +711,22 @@ def _no_asyncio_pending_tasks(event_loop):
     event_loop.run_until_complete(asyncio.sleep(0))
 
     # Detect all leftover tasks.
-    after = {t for t in list(asyncio.tasks._all_tasks) if not t.done()}
+    after = _get_all_tasks()
     remains = after - before
     if remains:
         pytest.fail(f"Unattended asyncio tasks detected: {remains!r}")
+
+
+def _get_all_tasks() -> Set[asyncio.Task]:
+    """Similar to `asyncio.all_tasks`, but for all event loops at once."""
+    i = 0
+    while True:
+        try:
+            tasks = list(asyncio.tasks._all_tasks)
+        except RuntimeError:
+            i += 1
+            if i >= 1000:
+                raise  # we are truly unlucky today; try again tomorrow
+        else:
+            break
+    return {t for t in tasks if not t.done()}