(self)
| 1384 | next(stream) |
| 1385 | |
| 1386 | def test_gzipped_streaming_tell(self) -> None: |
| 1387 | compress = zlib.compressobj(6, zlib.DEFLATED, 16 + zlib.MAX_WBITS) |
| 1388 | uncompressed_data = b"foo" |
| 1389 | data = compress.compress(uncompressed_data) |
| 1390 | data += compress.flush() |
| 1391 | |
| 1392 | fp = BytesIO(data) |
| 1393 | resp = HTTPResponse( |
| 1394 | fp, headers={"content-encoding": "gzip"}, preload_content=False |
| 1395 | ) |
| 1396 | stream = resp.stream() |
| 1397 | |
| 1398 | # Read everything |
| 1399 | payload = next(stream) |
| 1400 | assert payload == uncompressed_data |
| 1401 | |
| 1402 | assert len(data) == resp.tell() |
| 1403 | |
| 1404 | with pytest.raises(StopIteration): |
| 1405 | next(stream) |
| 1406 | |
| 1407 | def test_deflate_streaming_tell_intermediate_point(self) -> None: |
| 1408 | # Ensure that ``tell()`` returns the correct number of bytes when |
nothing calls this directly
no test coverage detected