(loop)
| 221 | |
| 222 | |
| 223 | def test_feed_parser_stop(loop): |
| 224 | def p(out, buf): |
| 225 | yield # chunk |
| 226 | |
| 227 | stream = parsers.StreamParser(loop=loop) |
| 228 | stream.set_parser(p) |
| 229 | |
| 230 | stream.feed_data(b'line1') |
| 231 | assert stream._parser is None |
| 232 | assert b'' == bytes(stream._buffer) |
| 233 | |
| 234 | |
| 235 | def test_feed_eof_exc(loop): |
nothing calls this directly
no test coverage detected