mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-08-31 14:07:50 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 9

Issue #22685: Fix test_pause_reading() of asyncio/test_subprocess

Browse source 
Override the connect_read_pipe() method of the loop to mock immediatly
pause_reading() and resume_reading() methods.

The test failed randomly on FreeBSD 9 buildbot and on Windows using trollius.
This commit is contained in:
Victor Stinner 2015-01-15 22:52:59 +01:00
parent 406204c8c2
commit f716d8b7a5
1 changed files with 12 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

14
Lib/test/test_asyncio/test_subprocess.py
View file

@ -179,6 +179,18 @@ class SubprocessMixin:
'sys.stdout.write("x" * %s)' % size,
'sys.stdout.flush()',
))
connect_read_pipe = self.loop.connect_read_pipe
@asyncio.coroutine
def connect_read_pipe_mock(*args, **kw):
transport, protocol = yield from connect_read_pipe(*args, **kw)
transport.pause_reading = mock.Mock()
transport.resume_reading = mock.Mock()
return (transport, protocol)
self.loop.connect_read_pipe = connect_read_pipe_mock
proc = yield from asyncio.create_subprocess_exec(
sys.executable, '-c', code,
stdin=asyncio.subprocess.PIPE,
@ -186,8 +198,6 @@ class SubprocessMixin:
limit=limit,
loop=self.loop)
stdout_transport = proc._transport.get_pipe_transport(1)
stdout_transport.pause_reading = mock.Mock()
stdout_transport.resume_reading = mock.Mock()
stdout, stderr = yield from proc.communicate()