(self)
| 439 | self.assertEqual(b'', stream._buffer) |
| 440 | |
| 441 | def test_readuntil_eof(self): |
| 442 | stream = asyncio.StreamReader(loop=self.loop) |
| 443 | data = b'some dataAA' |
| 444 | stream.feed_data(data) |
| 445 | stream.feed_eof() |
| 446 | |
| 447 | with self.assertRaisesRegex(asyncio.IncompleteReadError, |
| 448 | 'undefined expected bytes') as cm: |
| 449 | self.loop.run_until_complete(stream.readuntil(b'AAA')) |
| 450 | self.assertEqual(cm.exception.partial, data) |
| 451 | self.assertIsNone(cm.exception.expected) |
| 452 | self.assertEqual(b'', stream._buffer) |
| 453 | |
| 454 | def test_readuntil_limit_found_sep(self): |
| 455 | stream = asyncio.StreamReader(loop=self.loop, limit=3) |
nothing calls this directly
no test coverage detected