Method test_read_limit

Lib/test/test_asyncio/test_streams.py:216–221

(self)

            asyncio.StreamReader(limit=-1, loop=self.loop)

    def test_read_limit(self):
        stream = asyncio.StreamReader(limit=3, loop=self.loop)
        stream.feed_data(b'chunk')
        data = self.loop.run_until_complete(stream.read(5))
        self.assertEqual(b'chunk', data)
        self.assertEqual(b'', stream._buffer)

    def test_readline(self):
        # Read one line. 'readline' will need to wait for the data