openstack/deb-python-kafka
Code Issues Proposed changes

Update xerial snappy tests for new xerial default

Browse Source
This commit is contained in:
Dana Powers
2016-01-25 16:28:23 -08:00
parent cd956d8d79
commit 7b5e42bc8b
1 changed files with 5 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
test/test_codec.py
View File

@@ -35,13 +35,15 @@ def test_snappy_detect_xerial():
header = b'\x82SNAPPY\x00\x00\x00\x00\x01\x00\x00\x00\x01Some extra bytes'
false_header = b'\x01SNAPPY\x00\x00\x00\x01\x00\x00\x00\x01'
random_snappy = snappy_encode(b'SNAPPY' * 50)
default_snappy = snappy_encode(b'foobar' * 50)
random_snappy = snappy_encode(b'SNAPPY' * 50, xerial_compatible=False)
short_data = b'\x01\x02\x03\x04'
assert _detect_xerial_stream(header) is True
assert _detect_xerial_stream(b'') is False
assert _detect_xerial_stream(b'\x00') is False
assert _detect_xerial_stream(false_header) is False
assert _detect_xerial_stream(default_snappy) is True
assert _detect_xerial_stream(random_snappy) is False
assert _detect_xerial_stream(short_data) is False
@@ -49,9 +51,9 @@ def test_snappy_detect_xerial():
@pytest.mark.skipif(not has_snappy(), reason="Snappy not available")
def test_snappy_decode_xerial():
header = b'\x82SNAPPY\x00\x00\x00\x00\x01\x00\x00\x00\x01'
random_snappy = snappy_encode(b'SNAPPY' * 50)
random_snappy = snappy_encode(b'SNAPPY' * 50, xerial_compatible=False)
block_len = len(random_snappy)
random_snappy2 = snappy_encode(b'XERIAL' * 50)
random_snappy2 = snappy_encode(b'XERIAL' * 50, xerial_compatible=False)
block_len2 = len(random_snappy2)
to_test = header \