def test_pause_reading(self):
        limit = 10
        size = (limit * 2 + 1)

        @asyncio.coroutine
        def test_pause_reading():
            code = '\n'.join((
                'import sys',
                '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):
                connect = connect_read_pipe(*args, **kw)
                transport, protocol = yield From(connect)
                transport.pause_reading = mock.Mock()
                transport.resume_reading = mock.Mock()
                raise 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,
                                         stdout=asyncio.subprocess.PIPE,
                                         limit=limit,
                                         loop=self.loop))
            stdout_transport = proc._transport.get_pipe_transport(1)

            stdout, stderr = yield From(proc.communicate())

            # The child process produced more than limit bytes of output,
            # the stream reader transport should pause the protocol to not
            # allocate too much memory.
            raise Return(stdout, stdout_transport)

        # Issue #22685: Ensure that the stream reader pauses the protocol
        # when the child process produces too much data
        stdout, transport = self.loop.run_until_complete(test_pause_reading())

        self.assertEqual(stdout, b'x' * size)
        self.assertTrue(transport.pause_reading.called)
        self.assertTrue(transport.resume_reading.called)