(self)
| 365 | self.assertEqual(b'ine3\n', stream._buffer) |
| 366 | |
| 367 | def test_readline_exception(self): |
| 368 | stream = asyncio.StreamReader(loop=self.loop) |
| 369 | stream.feed_data(b'line\n') |
| 370 | |
| 371 | data = self.loop.run_until_complete(stream.readline()) |
| 372 | self.assertEqual(b'line\n', data) |
| 373 | |
| 374 | stream.set_exception(ValueError()) |
| 375 | self.assertRaises( |
| 376 | ValueError, self.loop.run_until_complete, stream.readline()) |
| 377 | self.assertEqual(b'', stream._buffer) |
| 378 | |
| 379 | def test_readuntil_separator(self): |
| 380 | stream = asyncio.StreamReader(loop=self.loop) |
nothing calls this directly
no test coverage detected