(self)
| 1368 | next(stream) |
| 1369 | |
| 1370 | def test_gzipped_streaming(self) -> None: |
| 1371 | compress = zlib.compressobj(6, zlib.DEFLATED, 16 + zlib.MAX_WBITS) |
| 1372 | data = compress.compress(b"foo") |
| 1373 | data += compress.flush() |
| 1374 | |
| 1375 | fp = BytesIO(data) |
| 1376 | resp = HTTPResponse( |
| 1377 | fp, headers={"content-encoding": "gzip"}, preload_content=False |
| 1378 | ) |
| 1379 | stream = resp.stream(2) |
| 1380 | |
| 1381 | assert next(stream) == b"fo" |
| 1382 | assert next(stream) == b"o" |
| 1383 | with pytest.raises(StopIteration): |
| 1384 | next(stream) |
| 1385 | |
| 1386 | def test_gzipped_streaming_tell(self) -> None: |
| 1387 | compress = zlib.compressobj(6, zlib.DEFLATED, 16 + zlib.MAX_WBITS) |
nothing calls this directly
no test coverage detected