This repository has been archived by the owner on Jul 11, 2023. It is now read-only.
/
test_ndjson.py
77 lines (58 loc) · 1.78 KB
/
test_ndjson.py
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
# -*- coding: utf-8 -*-
from __future__ import division
from __future__ import print_function
from __future__ import absolute_import
from __future__ import unicode_literals
import io
import pytest
from mock import Mock
from six import StringIO
from tabulator import exceptions, Stream
from tabulator.parsers.ndjson import NDJSONParser
# Tests
def test_ndjson_parser():
source = 'data/table.ndjson'
encoding = None
loader = Mock()
loader.load = Mock(return_value=io.open(source, encoding='utf-8'))
parser = NDJSONParser(loader)
assert parser.closed is True
parser.open(source, encoding=encoding)
assert parser.closed is False
assert list(parser.extended_rows) == [
(1, ['id', 'name'], [1, 'english']),
(2, ['id', 'name'], [2, '中国人']),
]
assert len(list(parser.extended_rows)) == 0
parser.reset()
assert len(list(parser.extended_rows)) == 2
parser.close()
assert parser.closed
def test_stream_ndjson():
with Stream('data/table.ndjson', headers=1) as stream:
assert stream.headers == ['id', 'name']
assert stream.read(keyed=True) == [
{'id': 1, 'name': 'english'},
{'id': 2, 'name': '中国人'}]
def test_ndjson_list():
stream = StringIO(
'[1, 2, 3]\n'
'[4, 5, 6]\n'
)
loader = Mock(load=Mock(return_value=stream))
parser = NDJSONParser(loader)
parser.open(None)
assert list(parser.extended_rows) == [
(1, None, [1, 2, 3]),
(2, None, [4, 5, 6]),
]
def test_ndjson_scalar():
stream = StringIO(
'1\n'
'2\n'
)
loader = Mock(load=Mock(return_value=stream))
parser = NDJSONParser(loader)
parser.open(None)
with pytest.raises(exceptions.SourceError):
list(parser.extended_rows)