(loop, lines_parser)
| 158 | |
| 159 | |
| 160 | def test_set_parser_unset(loop, lines_parser): |
| 161 | stream = parsers.StreamParser(loop=loop) |
| 162 | s = stream.set_parser(lines_parser) |
| 163 | |
| 164 | stream.feed_data(b'line1\r\nline2\r\n') |
| 165 | assert ([(bytearray(b'line1\r\n'), 7), (bytearray(b'line2\r\n'), 7)] == |
| 166 | list(s._buffer)) |
| 167 | assert b'' == bytes(stream._buffer) |
| 168 | stream.unset_parser() |
| 169 | assert s._eof |
| 170 | assert b'' == bytes(stream._buffer) |
| 171 | |
| 172 | |
| 173 | def test_set_parser_feed_existing_stop(loop): |
nothing calls this directly
no test coverage detected