cramjam 2.0.2

Thin Python bindings to de/compression algorithms in Rust
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

import pytest
import numpy as np
import cramjam
import hashlib


def same_same(a, b):
    return hashlib.md5(a).hexdigest() == hashlib.md5(b).hexdigest()


@pytest.mark.parametrize("is_bytearray", (True, False))
@pytest.mark.parametrize(
    "variant_str", ("snappy", "brotli", "lz4", "gzip", "deflate", "zstd")
)
def test_variants_simple(variant_str, is_bytearray):

    variant = getattr(cramjam, variant_str)

    uncompressed = b"some bytes to compress 123" * 1000
    if is_bytearray:
        uncompressed = bytearray(uncompressed)

    compressed = variant.compress(uncompressed)
    assert compressed != uncompressed
    assert type(compressed) == type(uncompressed)

    decompressed = variant.decompress(compressed, output_len=len(uncompressed))
    assert decompressed == uncompressed
    assert type(decompressed) == type(uncompressed)


@pytest.mark.parametrize(
    "variant_str", ("snappy", "brotli", "lz4", "gzip", "deflate", "zstd")
)
def test_variants_raise_exception(variant_str):
    variant = getattr(cramjam, variant_str)
    with pytest.raises(cramjam.DecompressionError):
        variant.decompress(b"sknow")


@pytest.mark.parametrize("variant_str", ("snappy", "brotli", "gzip", "deflate", "zstd"))
def test_variants_de_compress_into(variant_str):

    # TODO: support lz4 de/compress_into

    variant = getattr(cramjam, variant_str)

    data = b"oh what a beautiful morning, oh what a beautiful day!!" * 1000000

    compressed_array = np.zeros(len(data), dtype=np.uint8)  # plenty of space
    compressed_size = variant.compress_into(data, compressed_array)
    decompressed = variant.decompress(compressed_array[:compressed_size].tobytes())
    assert same_same(decompressed, data)

    compressed = variant.compress(data)
    decompressed_array = np.zeros(len(data), np.uint8)
    decompressed_size = variant.decompress_into(compressed, decompressed_array)
    decompressed = decompressed_array[:decompressed_size].tobytes()
    assert same_same(decompressed, data)