Issue #19305: try to fix sporadic test_asyncio failure on FreeBSD 10.0

1 files changed, 15 insertions, 0 deletions

diff --git a/Lib/asyncio/test_utils.py b/Lib/asyncio/test_utils.py
index 91bbedb..d650c44 100644
--- a/Lib/asyncio/test_utils.py
+++ b/Lib/asyncio/test_utils.py
@@ -7,6 +7,7 @@ import unittest.mock
 import os
 import sys
 import threading
+import time
 import unittest
 import unittest.mock
 from wsgiref.simple_server import make_server, WSGIRequestHandler, WSGIServer
@@ -46,6 +47,20 @@ def run_briefly(loop):
         gen.close()
 
 
+def run_until(loop, pred, timeout=None):
+    if timeout is not None:
+        deadline = time.time() + timeout
+    while not pred():
+        if timeout is not None:
+            timeout = deadline - time.time()
+            if timeout <= 0:
+                return False
+            loop.run_until_complete(tasks.sleep(timeout, loop=loop))
+        else:
+            run_briefly(loop)
+    return True
+
+
 def run_once(loop):
     """loop.stop() schedules _raise_stop_error()
     and run_forever() runs until _raise_stop_error() callback.