(self)
| 530 | self.assertEqual(self.DATA, stream._buffer) |
| 531 | |
| 532 | def test_readexactly(self): |
| 533 | # Read exact number of bytes. |
| 534 | stream = asyncio.StreamReader(loop=self.loop) |
| 535 | |
| 536 | n = 2 * len(self.DATA) |
| 537 | read_task = self.loop.create_task(stream.readexactly(n)) |
| 538 | |
| 539 | def cb(): |
| 540 | stream.feed_data(self.DATA) |
| 541 | stream.feed_data(self.DATA) |
| 542 | stream.feed_data(self.DATA) |
| 543 | self.loop.call_soon(cb) |
| 544 | |
| 545 | data = self.loop.run_until_complete(read_task) |
| 546 | self.assertEqual(self.DATA + self.DATA, data) |
| 547 | self.assertEqual(self.DATA, stream._buffer) |
| 548 | |
| 549 | def test_readexactly_limit(self): |
| 550 | stream = asyncio.StreamReader(limit=3, loop=self.loop) |
nothing calls this directly
no test coverage detected