(self)
| 156 | self.assertEqual(b'', stream._buffer) |
| 157 | |
| 158 | def test_read_line_breaks(self): |
| 159 | # Read bytes without line breaks. |
| 160 | stream = asyncio.StreamReader(loop=self.loop) |
| 161 | stream.feed_data(b'line1') |
| 162 | stream.feed_data(b'line2') |
| 163 | |
| 164 | data = self.loop.run_until_complete(stream.read(5)) |
| 165 | |
| 166 | self.assertEqual(b'line1', data) |
| 167 | self.assertEqual(b'line2', stream._buffer) |
| 168 | |
| 169 | def test_read_eof(self): |
| 170 | # Read bytes, stop at eof. |
nothing calls this directly
no test coverage detected