(self, data: bytes)
| 663 | @onlyZstd() |
| 664 | @pytest.mark.parametrize("data", decode_param_set) |
| 665 | def test_decode_zstd_read1(self, data: bytes) -> None: |
| 666 | encoded_data = zstd_compress(data) |
| 667 | fp = BytesIO(encoded_data) |
| 668 | |
| 669 | r = HTTPResponse( |
| 670 | fp, headers={"content-encoding": "zstd"}, preload_content=False |
| 671 | ) |
| 672 | |
| 673 | amt_decoded = 0 |
| 674 | decoded_data = b"" |
| 675 | # loop, as read1() may return just partial data |
| 676 | while amt_decoded < len(data): |
| 677 | part = r.read1(decode_content=True) |
| 678 | amt_decoded += len(part) |
| 679 | decoded_data += part |
| 680 | assert decoded_data == data |
| 681 | |
| 682 | _test_compressor_params: list[ |
| 683 | tuple[str, tuple[str, typing.Callable[[bytes], bytes]] | None] |
nothing calls this directly
no test coverage detected