Stowage/cpython
2
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-10-30 21:21:22 +00:00
Code Issues Projects Releases Packages Wiki Activity

Issue #22685: Debug test_pause_reading() on FreeBSD

Browse source
This commit is contained in:
Victor Stinner 2014-11-28 18:02:03 +01:00
parent e7182979d6
commit cbbd04d176
1 changed files with 10 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -163,13 +163,14 @@ def test_communicate_ignore_broken_pipe(self):
self.loop.run_until_complete(proc.wait()) self.loop.run_until_complete(proc.wait())
def test_pause_reading(self): def test_pause_reading(self):
limit = 10
size = (limit * 2 + 1)
@asyncio.coroutine @asyncio.coroutine
def test_pause_reading(): def test_pause_reading():
limit = 100
code = '\n'.join(( code = '\n'.join((
'import sys', 'import sys',
'sys.stdout.write("x" * %s)' % (limit * 2 + 1), 'sys.stdout.write("x" * %s)' % size,
'sys.stdout.flush()', 'sys.stdout.flush()',
)) ))
proc = yield from asyncio.create_subprocess_exec( proc = yield from asyncio.create_subprocess_exec(
@ -181,17 +182,19 @@ def test_pause_reading():
stdout_transport = proc._transport.get_pipe_transport(1) stdout_transport = proc._transport.get_pipe_transport(1)
stdout_transport.pause_reading = mock.Mock() stdout_transport.pause_reading = mock.Mock()
yield from proc.wait() stdout, stderr = yield from proc.communicate()
# The child process produced more than limit bytes of output, # The child process produced more than limit bytes of output,
# the stream reader transport should pause the protocol to not # the stream reader transport should pause the protocol to not
# allocate too much memory. # allocate too much memory.
return stdout_transport.pause_reading.called return (stdout, stdout_transport)
# Issue #22685: Ensure that the stream reader pauses the protocol # Issue #22685: Ensure that the stream reader pauses the protocol
# when the child process produces too much data # when the child process produces too much data
called = self.loop.run_until_complete(test_pause_reading()) stdout, transport = self.loop.run_until_complete(test_pause_reading())
self.assertTrue(called)
self.assertEqual(stdout, b'x' * size)
self.assertTrue(transport.pause_reading.called)
if sys.platform != 'win32': if sys.platform != 'win32':