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

Function test_snappy_encode_xerial

test/test_codec.py:71–83  ·  view source on GitHub ↗
()

Source from the content-addressed store, hash-verified

69
70@pytest.mark.skipif(not has_snappy(), reason="Snappy not available")
71def test_snappy_encode_xerial():
72 to_ensure = (
73 b'\x82SNAPPY\x00\x00\x00\x00\x01\x00\x00\x00\x01'
74 b'\x00\x00\x00\x18'
75 b'\xac\x02\x14SNAPPY\xfe\x06\x00\xfe\x06\x00\xfe\x06\x00\xfe\x06\x00\x96\x06\x00'
76 b'\x00\x00\x00\x18'
77 b'\xac\x02\x14XERIAL\xfe\x06\x00\xfe\x06\x00\xfe\x06\x00\xfe\x06\x00\x96\x06\x00'
78 )
79
80 to_test = (b'SNAPPY' * 50) + (b'XERIAL' * 50)
81
82 compressed = snappy_encode(to_test, xerial_compatible=True, xerial_blocksize=300)
83 assert compressed == to_ensure
84
85
86@pytest.mark.skipif(not has_lz4() or platform.python_implementation() == 'PyPy',

Callers

nothing calls this directly

Calls 1

snappy_encodeFunction · 0.90

Tested by

no test coverage detected

Used in the wild real call sites across dependent graphs

searching dependent graphs…