(compression)
| 862 | "snappy" |
| 863 | ]) |
| 864 | def test_compression_level(compression): |
| 865 | if not Codec.is_available(compression): |
| 866 | pytest.skip(f"{compression} support is not built") |
| 867 | |
| 868 | codec = Codec(compression) |
| 869 | if codec.name == "snappy": |
| 870 | assert codec.compression_level is None |
| 871 | else: |
| 872 | assert isinstance(codec.compression_level, int) |
| 873 | |
| 874 | # These codecs do not support a compression level |
| 875 | no_level = ['snappy'] |
| 876 | if compression in no_level: |
| 877 | assert not Codec.supports_compression_level(compression) |
| 878 | with pytest.raises(ValueError): |
| 879 | Codec(compression, 0) |
| 880 | with pytest.raises(ValueError): |
| 881 | Codec.minimum_compression_level(compression) |
| 882 | with pytest.raises(ValueError): |
| 883 | Codec.maximum_compression_level(compression) |
| 884 | with pytest.raises(ValueError): |
| 885 | Codec.default_compression_level(compression) |
| 886 | return |
| 887 | |
| 888 | INPUT_SIZE = 10000 |
| 889 | test_data = (np.random.randint(0, 255, size=INPUT_SIZE) |
| 890 | .astype(np.uint8) |
| 891 | .tobytes()) |
| 892 | test_buf = pa.py_buffer(test_data) |
| 893 | |
| 894 | min_level = Codec.minimum_compression_level(compression) |
| 895 | max_level = Codec.maximum_compression_level(compression) |
| 896 | default_level = Codec.default_compression_level(compression) |
| 897 | |
| 898 | assert min_level < max_level |
| 899 | assert default_level >= min_level |
| 900 | assert default_level <= max_level |
| 901 | |
| 902 | for compression_level in range(min_level, max_level+1): |
| 903 | codec = Codec(compression, compression_level) |
| 904 | compressed_buf = codec.compress(test_buf) |
| 905 | compressed_bytes = codec.compress(test_data, asbytes=True) |
| 906 | assert isinstance(compressed_bytes, bytes) |
| 907 | decompressed_buf = codec.decompress(compressed_buf, INPUT_SIZE) |
| 908 | decompressed_bytes = codec.decompress(compressed_bytes, INPUT_SIZE, |
| 909 | asbytes=True) |
| 910 | |
| 911 | assert isinstance(decompressed_bytes, bytes) |
| 912 | |
| 913 | assert decompressed_buf.equals(test_buf) |
| 914 | assert decompressed_bytes == test_data |
| 915 | |
| 916 | with pytest.raises(ValueError): |
| 917 | codec.decompress(compressed_bytes) |
| 918 | |
| 919 | # The ability to set a seed this way is not present on older versions of |
| 920 | # numpy (currently in our python 3.6 CI build). Some inputs might just |
| 921 | # happen to compress the same between the two levels so using seeded |
nothing calls this directly
no test coverage detected