(self)
| 554 | self.assertEqual(b'', stream._buffer) |
| 555 | |
| 556 | def test_readexactly_eof(self): |
| 557 | # Read exact number of bytes (eof). |
| 558 | stream = asyncio.StreamReader(loop=self.loop) |
| 559 | n = 2 * len(self.DATA) |
| 560 | read_task = self.loop.create_task(stream.readexactly(n)) |
| 561 | |
| 562 | def cb(): |
| 563 | stream.feed_data(self.DATA) |
| 564 | stream.feed_eof() |
| 565 | self.loop.call_soon(cb) |
| 566 | |
| 567 | with self.assertRaises(asyncio.IncompleteReadError) as cm: |
| 568 | self.loop.run_until_complete(read_task) |
| 569 | self.assertEqual(cm.exception.partial, self.DATA) |
| 570 | self.assertEqual(cm.exception.expected, n) |
| 571 | self.assertEqual(str(cm.exception), |
| 572 | '18 bytes read on a total of 36 expected bytes') |
| 573 | self.assertEqual(b'', stream._buffer) |
| 574 | |
| 575 | def test_readexactly_exception(self): |
| 576 | stream = asyncio.StreamReader(loop=self.loop) |
nothing calls this directly
no test coverage detected