forked from wireservice/csvkit
/
test_fixed.py
91 lines (73 loc) · 3.27 KB
/
test_fixed.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
78
79
80
81
82
83
84
85
86
87
88
89
90
91
import unittest
from cStringIO import StringIO
from csvkit.convert import fixed
class TestFixed(unittest.TestCase):
def test_fixed(self):
with open('examples/testfixed', 'r') as f:
with open('examples/testfixed_schema.csv', 'r') as schema:
output = fixed.fixed2csv(f, schema)
with open('examples/testfixed_converted.csv', 'r') as f:
self.assertEqual(f.read(), output)
def test_fixed_streaming(self):
with open('examples/testfixed', 'r') as f:
with open('examples/testfixed_schema.csv', 'r') as schema:
output = StringIO()
fixed.fixed2csv(f, schema, output=output)
output = output.getvalue()
with open('examples/testfixed_converted.csv', 'r') as f:
self.assertEqual(f.read(), output)
def test_schema_decoder_init(self):
rd = fixed.SchemaDecoder(['column', 'start', 'length'])
self.assertEqual(1,rd.start)
self.assertEqual(2,rd.length)
self.assertEqual(0,rd.column)
def test_schema_decoder_in_action(self):
rd = fixed.SchemaDecoder(['comment', 'start', 'length', 'column'])
(column, start, length) = rd(['This is a comment','0','1','column_name'])
self.assertEqual(False,rd.one_based)
self.assertEqual('column_name',column)
self.assertEqual(0, start)
self.assertEqual(1, length)
(column, start, length) = rd(['This is another comment','1','5','column_name2'])
self.assertEqual(False,rd.one_based)
self.assertEqual('column_name2',column)
self.assertEqual(1, start)
self.assertEqual(5, length)
(column, start, length) = rd(['yet another comment','9','14','column_name3'])
self.assertEqual(False,rd.one_based)
self.assertEqual('column_name3',column)
self.assertEqual(9, start)
self.assertEqual(14, length)
def test_one_based_row_decoder(self):
rd = fixed.SchemaDecoder(['column','start','length'])
(column, start, length) = rd(['LABEL', '1', '5' ])
self.assertEqual(True,rd.one_based)
self.assertEqual('LABEL',column)
self.assertEqual(0, start)
self.assertEqual(5, length)
(column, start, length) = rd(['LABEL2', '6', '15' ])
self.assertEqual('LABEL2',column)
self.assertEqual(5, start)
self.assertEqual(15, length)
def test_schematic_line_parser(self):
schema = """column,start,length
foo,1,5
bar,6,2
baz,8,5"""
f = StringIO(schema)
parser = fixed.FixedWidthRowParser(f)
self.assertEqual('foo',parser.headers[0])
self.assertEqual('bar',parser.headers[1])
self.assertEqual('baz',parser.headers[2])
parsed = parser.parse("111112233333")
self.assertEqual('11111',parsed[0])
self.assertEqual('22',parsed[1])
self.assertEqual('33333',parsed[2])
parsed = parser.parse(" 1 2 3")
self.assertEqual('1',parsed[0])
self.assertEqual('2',parsed[1])
self.assertEqual('3',parsed[2])
parsed = parser.parse("1 1 233 3")
self.assertEqual('1 1',parsed[0])
self.assertEqual('2',parsed[1])
self.assertEqual('33 3',parsed[2])