test/unittest_dataimport.py
author Christophe de Vienne <christophe@unlish.com>
Mon, 29 Dec 2014 12:35:34 +0100
changeset 10178 4b33246cb43c
parent 9695 aa982b7c3f2a
child 9903 d1fdbdbab194
permissions -rw-r--r--
[native] Change SQL errors log level. - The SQL error itself is logged as 'info'. - The 'transaction has been rolled back' message is logged as 'debug'. These log entries should not be critical: - If an error occur at the sql level, it will be transmitted as a CW error. The exception is raised and handled in a higher level function, which will decide if it is critical or not. - Admitting that rolling-back automatically on errors is wanted in some cases, it cannot be considered 'critical'. The caller will receive the original exception anyway, it is its job to decide if it is critical or not. Closes #4801120

from StringIO import StringIO
from logilab.common.testlib import TestCase, unittest_main
from cubicweb import dataimport


class UcsvreaderTC(TestCase):

    def test_empty_lines_skipped(self):
        stream = StringIO('''a,b,c,d,
1,2,3,4,
,,,,
,,,,
''')
        self.assertEqual([[u'a', u'b', u'c', u'd', u''],
                          [u'1', u'2', u'3', u'4', u''],
                          ],
                         list(dataimport.ucsvreader(stream)))
        stream.seek(0)
        self.assertEqual([[u'a', u'b', u'c', u'd', u''],
                          [u'1', u'2', u'3', u'4', u''],
                          [u'', u'', u'', u'', u''],
                          [u'', u'', u'', u'', u'']
                          ],
                         list(dataimport.ucsvreader(stream, skip_empty=False)))

    def test_skip_first(self):
        stream = StringIO('a,b,c,d,\n'
                          '1,2,3,4,\n')
        reader = dataimport.ucsvreader(stream, skipfirst=True,
                                       ignore_errors=True)
        self.assertEqual(list(reader),
                         [[u'1', u'2', u'3', u'4', u'']])

        stream.seek(0)
        reader = dataimport.ucsvreader(stream, skipfirst=True,
                                       ignore_errors=False)
        self.assertEqual(list(reader),
                         [[u'1', u'2', u'3', u'4', u'']])

        stream.seek(0)
        reader = dataimport.ucsvreader(stream, skipfirst=False,
                                       ignore_errors=True)
        self.assertEqual(list(reader),
                         [[u'a', u'b', u'c', u'd', u''],
                          [u'1', u'2', u'3', u'4', u'']])

        stream.seek(0)
        reader = dataimport.ucsvreader(stream, skipfirst=False,
                                       ignore_errors=False)
        self.assertEqual(list(reader),
                         [[u'a', u'b', u'c', u'd', u''],
                          [u'1', u'2', u'3', u'4', u'']])


if __name__ == '__main__':
    unittest_main()