Make tests more stable

tests/test_mixed.py

@@ -1,6 +1,8 @@
 import asyncio
 import threading
 
+from concurrent.futures import ThreadPoolExecutor
+
 import pytest
 
 import janus
@@ -327,20 +329,18 @@ async def test_put_notifies_sync_not_empty(self):
         loop = asyncio.get_running_loop()
         q = janus.Queue()
 
-        futures = [
-            loop.run_in_executor(None, q.sync_q.get)
-            for _ in range(4)
-        ]
+        with ThreadPoolExecutor(4) as executor:
+            for _ in range(4):
+                executor.submit(q.sync_q.get)
 
-        while q._sync_not_empty_waiting != 4:
-            await asyncio.sleep(0.001)
+            while q._sync_not_empty_waiting != 4:
+                await asyncio.sleep(0.001)
 
-        q.sync_q.put_nowait(1)
-        q.async_q.put_nowait(2)
-        await loop.run_in_executor(None, q.sync_q.put, 3)
-        await q.async_q.put(4)
+            q.sync_q.put_nowait(1)
+            q.async_q.put_nowait(2)
+            await loop.run_in_executor(executor, q.sync_q.put, 3)
+            await q.async_q.put(4)
 
-        await asyncio.gather(*futures)
         assert q.sync_q.empty()
         await q.aclose()
 
@@ -354,7 +354,8 @@ async def test_put_notifies_async_not_empty(self):
             for _ in range(4)
         ]
 
-        await asyncio.sleep(0)
+        while q._async_not_empty_waiting != 4:
+            await asyncio.sleep(0)
 
         q.sync_q.put_nowait(1)
         q.async_q.put_nowait(2)
@@ -372,20 +373,18 @@ async def test_get_notifies_sync_not_full(self):
         q.sync_q.put_nowait(1)
         q.sync_q.put_nowait(2)
 
-        futures = [
-            loop.run_in_executor(None, q.sync_q.put, object())
-            for _ in range(4)
-        ]
+        with ThreadPoolExecutor(4) as executor:
+            for _ in range(4):
+                executor.submit(q.sync_q.put, object())
 
-        while q._sync_not_full_waiting != 4:
-            await asyncio.sleep(0.001)
+            while q._sync_not_full_waiting != 4:
+                await asyncio.sleep(0.001)
 
-        q.sync_q.get_nowait()
-        q.async_q.get_nowait()
-        await loop.run_in_executor(None, q.sync_q.get)
-        await q.async_q.get()
+            q.sync_q.get_nowait()
+            q.async_q.get_nowait()
+            await loop.run_in_executor(executor, q.sync_q.get)
+            await q.async_q.get()
 
-        await asyncio.gather(*futures)
         assert q.sync_q.qsize() == 2
         await q.aclose()
 
@@ -401,7 +400,8 @@ async def test_get_notifies_async_not_full(self):
             for _ in range(4)
         ]
 
-        await asyncio.sleep(0)
+        while q._async_not_full_waiting != 4:
+            await asyncio.sleep(0)
 
         q.sync_q.get_nowait()
         q.async_q.get_nowait()