use async_joinall

Author: sleepyStick

test/asynchronous/test_gridfs.py

@@ -28,7 +28,7 @@
 sys.path[0:0] = [""]
 
 from test.asynchronous import AsyncIntegrationTest, async_client_context, unittest
-from test.utils import joinall, one
+from test.utils import async_joinall, one
 
 import gridfs
 from bson.binary import Binary
@@ -230,10 +230,7 @@ async def test_threaded_reads(self):
             tasks.append(JustRead(self.fs, 10, results))
             await tasks[i].start()
 
-        if _IS_SYNC:
-            joinall(tasks)
-        else:
-            await asyncio.wait([t.task for t in tasks if t.task is not None])
+        await async_joinall(tasks)
 
         self.assertEqual(100 * [b"hello"], results)
 
@@ -243,10 +240,7 @@ async def test_threaded_writes(self):
             tasks.append(JustWrite(self.fs, 10))
             await tasks[i].start()
 
-        if _IS_SYNC:
-            joinall(tasks)
-        else:
-            await asyncio.wait([t.task for t in tasks if t.task is not None])
+        await async_joinall(tasks)
 
         f = await self.fs.get_last_version("test")
         self.assertEqual(await f.read(), b"hello")

test/test_gridfs.py

@@ -228,10 +228,7 @@ def test_threaded_reads(self):
             tasks.append(JustRead(self.fs, 10, results))
             tasks[i].start()
 
-        if _IS_SYNC:
-            joinall(tasks)
-        else:
-            asyncio.wait([t.task for t in tasks if t.task is not None])
+        joinall(tasks)
 
         self.assertEqual(100 * [b"hello"], results)
 
@@ -241,10 +238,7 @@ def test_threaded_writes(self):
             tasks.append(JustWrite(self.fs, 10))
             tasks[i].start()
 
-        if _IS_SYNC:
-            joinall(tasks)
-        else:
-            asyncio.wait([t.task for t in tasks if t.task is not None])
+        joinall(tasks)
 
         f = self.fs.get_last_version("test")
         self.assertEqual(f.read(), b"hello")

test/utils.py

@@ -666,6 +666,11 @@ def joinall(threads):
         assert not t.is_alive(), "Thread %s hung" % t
 
 
+async def async_joinall(tasks):
+    """Join threads with a 5-minute timeout, assert joins succeeded"""
+    await asyncio.wait([t.task for t in tasks if t is not None], timeout=300)
+
+
 def wait_until(predicate, success_description, timeout=10):
     """Wait up to 10 seconds (by default) for predicate to be true.
 

tools/synchro.py

@@ -124,6 +124,7 @@
     "AsyncMockPool": "MockPool",
     "StopAsyncIteration": "StopIteration",
     "create_async_event": "create_event",
+    "async_joinall": "joinall",
 }
 
 docstring_replacements: dict[tuple[str, str], str] = {