(self)
| 434 | HTTPResponse(fp, headers={"content-encoding": "gzip"}) |
| 435 | |
| 436 | def test_decode_gzip_swallow_garbage(self) -> None: |
| 437 | # When data comes from multiple calls to read(), data after |
| 438 | # the first zlib error (here triggered by garbage) should be |
| 439 | # ignored. |
| 440 | compress = zlib.compressobj(6, zlib.DEFLATED, 16 + zlib.MAX_WBITS) |
| 441 | data = compress.compress(b"foo") |
| 442 | data += compress.flush() |
| 443 | data = data * 3 + b"foo" |
| 444 | |
| 445 | fp = BytesIO(data) |
| 446 | r = HTTPResponse( |
| 447 | fp, headers={"content-encoding": "gzip"}, preload_content=False |
| 448 | ) |
| 449 | ret = b"" |
| 450 | for _ in range(100): |
| 451 | ret += r.read(1) |
| 452 | if r.closed: |
| 453 | break |
| 454 | |
| 455 | assert ret == b"foofoofoo" |
| 456 | |
| 457 | def test_chunked_decoding_gzip_swallow_garbage(self) -> None: |
| 458 | compress = zlib.compressobj(6, zlib.DEFLATED, 16 + zlib.MAX_WBITS) |
nothing calls this directly
no test coverage detected