|
1 # copyright 2003-2015 LOGILAB S.A. (Paris, FRANCE), all rights reserved. |
|
2 # contact http://www.logilab.fr/ -- mailto:contact@logilab.fr |
|
3 # |
|
4 # This file is part of CubicWeb. |
|
5 # |
|
6 # CubicWeb is free software: you can redistribute it and/or modify it under the |
|
7 # terms of the GNU Lesser General Public License as published by the Free |
|
8 # Software Foundation, either version 2.1 of the License, or (at your option) |
|
9 # any later version. |
|
10 # |
|
11 # CubicWeb is distributed in the hope that it will be useful, but WITHOUT |
|
12 # ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS |
|
13 # FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more |
|
14 # details. |
|
15 # |
|
16 # You should have received a copy of the GNU Lesser General Public License along |
|
17 # with CubicWeb. If not, see <http://www.gnu.org/licenses/>. |
|
18 """unittest for cubicweb.dataimport.csv""" |
|
19 |
|
20 from io import BytesIO |
|
21 |
|
22 from logilab.common.testlib import TestCase, unittest_main |
|
23 |
|
24 from cubicweb.dataimport import csv |
|
25 |
|
26 |
|
27 class UcsvreaderTC(TestCase): |
|
28 |
|
29 def test_empty_lines_skipped(self): |
|
30 stream = BytesIO(b'''a,b,c,d, |
|
31 1,2,3,4, |
|
32 ,,,, |
|
33 ,,,, |
|
34 ''') |
|
35 self.assertEqual([[u'a', u'b', u'c', u'd', u''], |
|
36 [u'1', u'2', u'3', u'4', u''], |
|
37 ], |
|
38 list(csv.ucsvreader(stream))) |
|
39 stream.seek(0) |
|
40 self.assertEqual([[u'a', u'b', u'c', u'd', u''], |
|
41 [u'1', u'2', u'3', u'4', u''], |
|
42 [u'', u'', u'', u'', u''], |
|
43 [u'', u'', u'', u'', u''] |
|
44 ], |
|
45 list(csv.ucsvreader(stream, skip_empty=False))) |
|
46 |
|
47 def test_skip_first(self): |
|
48 stream = BytesIO(b'a,b,c,d,\n1,2,3,4,\n') |
|
49 reader = csv.ucsvreader(stream, skipfirst=True, ignore_errors=True) |
|
50 self.assertEqual(list(reader), |
|
51 [[u'1', u'2', u'3', u'4', u'']]) |
|
52 |
|
53 stream.seek(0) |
|
54 reader = csv.ucsvreader(stream, skipfirst=True, ignore_errors=False) |
|
55 self.assertEqual(list(reader), |
|
56 [[u'1', u'2', u'3', u'4', u'']]) |
|
57 |
|
58 stream.seek(0) |
|
59 reader = csv.ucsvreader(stream, skipfirst=False, ignore_errors=True) |
|
60 self.assertEqual(list(reader), |
|
61 [[u'a', u'b', u'c', u'd', u''], |
|
62 [u'1', u'2', u'3', u'4', u'']]) |
|
63 |
|
64 stream.seek(0) |
|
65 reader = csv.ucsvreader(stream, skipfirst=False, ignore_errors=False) |
|
66 self.assertEqual(list(reader), |
|
67 [[u'a', u'b', u'c', u'd', u''], |
|
68 [u'1', u'2', u'3', u'4', u'']]) |
|
69 |
|
70 |
|
71 if __name__ == '__main__': |
|
72 unittest_main() |