(loop)
| 267 | |
| 268 | |
| 269 | def test_feed_eof_unhandled_eof(loop): |
| 270 | def p(out, buf): |
| 271 | while True: |
| 272 | yield # read chunk |
| 273 | |
| 274 | stream = parsers.StreamParser(loop=loop) |
| 275 | s = stream.set_parser(p) |
| 276 | |
| 277 | stream.feed_data(b'line1') |
| 278 | stream.feed_eof() |
| 279 | assert not s.is_eof() |
| 280 | assert isinstance(s.exception(), RuntimeError) |
| 281 | |
| 282 | |
| 283 | def test_feed_parser2(loop, lines_parser): |
nothing calls this directly
no test coverage detected