File: test_frame_5.py

package info (click to toggle)
python-lz4 1.1.0%2Bdfsg-1
  • links: PTS, VCS
  • area: main
  • in suites: buster
  • size: 424 kB
  • sloc: ansic: 2,222; python: 1,627; makefile: 147
file content (92 lines) | stat: -rw-r--r-- 2,319 bytes parent folder | download
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
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
import lz4.frame
import time
import pytest

test_data=[
    (b'a' * 1024 * 1024),
]

@pytest.fixture(
    params=test_data,
    ids=[
        'data' + str(i) for i in range(len(test_data))
    ]
)
def data(request):
    return request.param


def test_frame_decompress_mem_usage(data):
    tracemalloc = pytest.importorskip('tracemalloc')

    tracemalloc.start()

    compressed = lz4.frame.compress(data)
    prev_snapshot = None

    for i in range(1000):
        decompressed = lz4.frame.decompress(compressed)

        if i % 100 == 0:
            snapshot = tracemalloc.take_snapshot()

            if prev_snapshot:
                stats = snapshot.compare_to(prev_snapshot, 'lineno')
                assert stats[0].size_diff < (1024 * 4)

            prev_snapshot = snapshot


def test_frame_decompress_chunk_mem_usage(data):
    tracemalloc = pytest.importorskip('tracemalloc')
    tracemalloc.start()

    compressed = lz4.frame.compress(data)

    prev_snapshot = None

    for i in range(1000):
        context = lz4.frame.create_decompression_context()
        decompressed = lz4.frame.decompress_chunk(context, compressed)

        if i % 100 == 0:
            snapshot = tracemalloc.take_snapshot()

            if prev_snapshot:
                stats = snapshot.compare_to(prev_snapshot, 'lineno')
                assert stats[0].size_diff < (1024 * 10)

            prev_snapshot = snapshot


def test_frame_open_decompress_mem_usage(data):
    tracemalloc = pytest.importorskip('tracemalloc')
    tracemalloc.start()

    with lz4.frame.open('test.lz4', 'w') as f:
        f.write(data)

    prev_snapshot = None

    for i in range(1000):
        with lz4.frame.open('test.lz4', 'r') as f:
            decompressed = f.read()

        if i % 100 == 0:
            snapshot = tracemalloc.take_snapshot()

            if prev_snapshot:
                stats = snapshot.compare_to(prev_snapshot, 'lineno')
                assert stats[0].size_diff < (1024 * 10)

            prev_snapshot = snapshot


# TODO: add many more memory usage tests along the lines of this one for other funcs

def test_dummy_always_pass():
    # If pytest finds all tests are skipped, then it exits with code 5 rather
    # than 0, which tox sees as an error. Here we add a dummy test that always passes.
    assert True