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
|
import os
from boltons.jsonutils import (JSONLIterator,
DEFAULT_BLOCKSIZE,
reverse_iter_lines)
CUR_PATH = os.path.dirname(os.path.abspath(__file__))
NEWLINES_DATA_PATH = CUR_PATH + '/newlines_test_data.txt'
JSONL_DATA_PATH = CUR_PATH + '/jsonl_test_data.txt'
def _test_reverse_iter_lines(filename, blocksize=DEFAULT_BLOCKSIZE):
fo = open(filename)
reference = fo.read()
fo.seek(0, os.SEEK_SET)
rev_lines = list(reverse_iter_lines(fo, blocksize))
assert '\n'.join(rev_lines[::-1]) == reference
def _test_reverse_iter_lines_bytes(filename, blocksize=DEFAULT_BLOCKSIZE):
fo = open(filename, 'rb')
reference = fo.read()
fo.seek(0, os.SEEK_SET)
rev_lines = list(reverse_iter_lines(fo, blocksize))
assert os.linesep.encode('ascii').join(rev_lines[::-1]) == reference
def test_reverse_iter_lines():
for blocksize in (2, 4, 16, 4096):
_test_reverse_iter_lines(NEWLINES_DATA_PATH, blocksize)
_test_reverse_iter_lines_bytes(NEWLINES_DATA_PATH, blocksize)
def test_jsonl_iterator():
ref = [{'4': 4}, {'3': 3}, {'2': 2}, {'1': 1}, {}]
jsonl_iter = JSONLIterator(open(JSONL_DATA_PATH), reverse=True)
jsonl_list = list(jsonl_iter)
assert jsonl_list == ref
|
