mmap-queue/test.py at main · rowles/mmap-queue · GitHub

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
import mmap_queue

import pytest
import os
import queue as buildin_q


def test_put_get():
    q = mmap_queue.RingBuffer()

    q.put_bytes(b'the')
    q.put_bytes(b'sky')
    q.put_bytes(b'is')
    q.put_bytes(b'falling')

    assert q.size() == 47

    assert q.get_bytes().tobytes() == b'the'
    assert q.get_bytes().tobytes() == b'sky'
    assert q.get_bytes().tobytes() == b'is'
    assert q.get_bytes().tobytes() == b'falling'

    assert q.size() == 0


def test_fifo():
    q = mmap_queue.RingBuffer()

    q.put_bytes(b'1')
    q.put_bytes(b'2')
    q.put_bytes(b'3')

    assert q.size() == (8 + 1) * 3

    assert q.get_bytes() == b'1'
    assert q.get_bytes() == b'2'
    assert q.get_bytes() == b'3'

    assert q.size() == 0


def test_empty():
    q = mmap_queue.RingBuffer()

    with pytest.raises(buildin_q.Empty):
        q.get_bytes()

    assert q.size() == 0


def test_full():
    q = mmap_queue.RingBuffer(100)

    with pytest.raises(buildin_q.Full):
        q.put_bytes(b'a'*1000)

    assert q.size() == 0


def test_filebacked():
    testfile = '/tmp/test.buf'

    q = mmap_queue.RingBuffer(1000, file_path=testfile)

    assert os.path.isfile(testfile)

    q.put_bytes(b'123456')
    assert q.get_bytes() == b'123456'

    del q

    assert not os.path.isfile(testfile)