1 from StringIO import StringIO |
1 from StringIO import StringIO |
2 from logilab.common.testlib import TestCase, unittest_main |
2 from logilab.common.testlib import TestCase, unittest_main |
3 from cubicweb import dataimport |
3 from cubicweb import dataimport |
|
4 |
|
5 |
4 class UcsvreaderTC(TestCase): |
6 class UcsvreaderTC(TestCase): |
5 |
7 |
6 def test_empty_lines_skipped(self): |
8 def test_empty_lines_skipped(self): |
7 stream = StringIO('''a,b,c,d, |
9 stream = StringIO('''a,b,c,d, |
8 1,2,3,4, |
10 1,2,3,4, |
19 [u'', u'', u'', u'', u''], |
21 [u'', u'', u'', u'', u''], |
20 [u'', u'', u'', u'', u''] |
22 [u'', u'', u'', u'', u''] |
21 ], |
23 ], |
22 list(dataimport.ucsvreader(stream, skip_empty=False))) |
24 list(dataimport.ucsvreader(stream, skip_empty=False))) |
23 |
25 |
|
26 def test_skip_first(self): |
|
27 stream = StringIO('a,b,c,d,\n' |
|
28 '1,2,3,4,\n') |
|
29 reader = dataimport.ucsvreader(stream, skipfirst=True, |
|
30 ignore_errors=True) |
|
31 self.assertEqual(list(reader), |
|
32 [[u'1', u'2', u'3', u'4', u'']]) |
|
33 |
|
34 stream.seek(0) |
|
35 reader = dataimport.ucsvreader(stream, skipfirst=True, |
|
36 ignore_errors=False) |
|
37 self.assertEqual(list(reader), |
|
38 [[u'1', u'2', u'3', u'4', u'']]) |
|
39 |
|
40 stream.seek(0) |
|
41 reader = dataimport.ucsvreader(stream, skipfirst=False, |
|
42 ignore_errors=True) |
|
43 self.assertEqual(list(reader), |
|
44 [[u'a', u'b', u'c', u'd', u''], |
|
45 [u'1', u'2', u'3', u'4', u'']]) |
|
46 |
|
47 stream.seek(0) |
|
48 reader = dataimport.ucsvreader(stream, skipfirst=False, |
|
49 ignore_errors=False) |
|
50 self.assertEqual(list(reader), |
|
51 [[u'a', u'b', u'c', u'd', u''], |
|
52 [u'1', u'2', u'3', u'4', u'']]) |
|
53 |
24 |
54 |
25 if __name__ == '__main__': |
55 if __name__ == '__main__': |
26 unittest_main() |
56 unittest_main() |