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