(loop, lines_parser)
| 193 | |
| 194 | |
| 195 | def test_feed_parser(loop, lines_parser): |
| 196 | stream = parsers.StreamParser(loop=loop) |
| 197 | s = stream.set_parser(lines_parser) |
| 198 | |
| 199 | stream.feed_data(b'line1') |
| 200 | stream.feed_data(b'\r\nline2\r\ndata') |
| 201 | assert b'data' == bytes(stream._buffer) |
| 202 | |
| 203 | stream.feed_eof() |
| 204 | assert ([(bytearray(b'line1\r\n'), 7), (bytearray(b'line2\r\n'), 7)] == |
| 205 | list(s._buffer)) |
| 206 | assert b'data' == bytes(stream._buffer) |
| 207 | assert s.is_eof() |
| 208 | |
| 209 | |
| 210 | def test_feed_parser_exc(loop): |
nothing calls this directly
no test coverage detected