(loop)
| 251 | |
| 252 | |
| 253 | def test_feed_eof_stop(loop): |
| 254 | def p(out, buf): |
| 255 | try: |
| 256 | while True: |
| 257 | yield # read chunk |
| 258 | except parsers.EofStream: |
| 259 | out.feed_eof() |
| 260 | |
| 261 | stream = parsers.StreamParser(loop=loop) |
| 262 | s = stream.set_parser(p) |
| 263 | |
| 264 | stream.feed_data(b'line1') |
| 265 | stream.feed_eof() |
| 266 | assert s._eof |
| 267 | |
| 268 | |
| 269 | def test_feed_eof_unhandled_eof(loop): |
nothing calls this directly
no test coverage detected