MCPcopy Create free account
hub / github.com/dpkp/kafka-python / test_snappy_decode_xerial

Function test_snappy_decode_xerial

test/test_codec.py:56–67  ·  view source on GitHub ↗
()

Source from the content-addressed store, hash-verified

54
55@pytest.mark.skipif(not has_snappy(), reason="Snappy not available")
56def test_snappy_decode_xerial():
57 header = b'\x82SNAPPY\x00\x00\x00\x00\x01\x00\x00\x00\x01'
58 random_snappy = snappy_encode(b'SNAPPY' * 50, xerial_compatible=False)
59 block_len = len(random_snappy)
60 random_snappy2 = snappy_encode(b'XERIAL' * 50, xerial_compatible=False)
61 block_len2 = len(random_snappy2)
62
63 to_test = header \
64 + struct.pack('!i', block_len) + random_snappy \
65 + struct.pack('!i', block_len2) + random_snappy2 \
66
67 assert snappy_decode(to_test) == (b'SNAPPY' * 50) + (b'XERIAL' * 50)
68
69
70@pytest.mark.skipif(not has_snappy(), reason="Snappy not available")

Callers

nothing calls this directly

Calls 2

snappy_encodeFunction · 0.90
snappy_decodeFunction · 0.90

Tested by

no test coverage detected

Used in the wild real call sites across dependent graphs

searching dependent graphs…