traverseda
/
cpython
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

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

This commit is contained in:
Antoine Pitrou 2013-10-20 01:51:25 +02:00
parent 0d9eefda34
commit d20afad7d4
2 changed files with 20 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
Lib/asyncio/test_utils.py
View File

@ -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.

7
Lib/test/test_asyncio/test_events.py
View File

@ -558,13 +558,14 @@ class EventLoopTestsMixin:
self.assertEqual(host, '0.0.0.0')
client = socket.socket()
client.connect(('127.0.0.1', port))
client.send(b'xxx')
client.sendall(b'xxx')
test_utils.run_briefly(self.loop)
self.assertIsInstance(proto, MyProto)
self.assertEqual('INITIAL', proto.state)
test_utils.run_briefly(self.loop)
self.assertEqual('CONNECTED', proto.state)
test_utils.run_briefly(self.loop) # windows iocp
test_utils.run_until(self.loop, lambda: proto.nbytes > 0,
timeout=10)
self.assertEqual(3, proto.nbytes)
# extra info is available
@ -623,6 +624,8 @@ class EventLoopTestsMixin:
self.assertIsInstance(proto, MyProto)
test_utils.run_briefly(self.loop)
self.assertEqual('CONNECTED', proto.state)
test_utils.run_until(self.loop, lambda: proto.nbytes > 0,
timeout=10)
self.assertEqual(3, proto.nbytes)
# extra info is available