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
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
|
import array
import os
import io
import pickle
import sys
import threading
import lz4.frame
import pytest
def test_issue_172_1(tmp_path):
"""Test reproducer for issue 172
Issue 172 is a reported failure occurring on Windows 10 only. This bug was
due to incorrect handling of Py_ssize_t types when doing comparisons and
using them as a size when allocating memory.
"""
input_data = 8 * os.urandom(1024)
thread_id = threading.get_native_id()
with lz4.frame.open(tmp_path / f'testfile_small_{thread_id}', 'wb') as fp:
bytes_written = fp.write(input_data) # noqa: F841
with lz4.frame.open(tmp_path / f'testfile_small_{thread_id}', 'rb') as fp:
data = fp.read(10)
assert len(data) == 10
def test_issue_172_2(tmp_path):
input_data = 9 * os.urandom(1024)
thread_id = threading.get_native_id()
with lz4.frame.open(tmp_path / f'testfile_small_{thread_id}', 'w') as fp:
bytes_written = fp.write(input_data) # noqa: F841
with lz4.frame.open(tmp_path / f'testfile_small_{thread_id}', 'r') as fp:
data = fp.read(10)
assert len(data) == 10
def test_issue_172_3(tmp_path):
input_data = 9 * os.urandom(1024)
thread_id = threading.get_native_id()
with lz4.frame.open(tmp_path / f'testfile_small_{thread_id}', 'wb') as fp:
bytes_written = fp.write(input_data) # noqa: F841
with lz4.frame.open(tmp_path / f'testfile_small_{thread_id}', 'rb') as fp:
data = fp.read(10)
assert len(data) == 10
with lz4.frame.open(tmp_path / f'testfile_small_{thread_id}', 'rb') as fp:
data = fp.read(16 * 1024 - 1)
assert len(data) == 9 * 1024
assert data == input_data
def test_issue_227_1():
q = array.array('Q', [1, 2, 3, 4, 5])
LENGTH = len(q) * q.itemsize
with lz4.frame.open(io.BytesIO(), 'w') as f:
assert f.write(q) == LENGTH
assert f.tell() == LENGTH
@pytest.mark.skipif(
sys.version_info < (3, 8),
reason="PickleBuffer only available in Python 3.8 or greater"
)
def test_issue_227_2():
q = array.array('Q', [1, 2, 3, 4, 5])
c = lz4.frame.compress(q)
d = lz4.frame.LZ4FrameDecompressor().decompress(pickle.PickleBuffer(c))
assert memoryview(q).tobytes() == d
|
