dataimport/test/test_pgstore.py
author Julien Cristau <julien.cristau@logilab.fr>
Mon, 09 Nov 2015 15:22:43 +0100
changeset 10869 575982c948a9
parent 10810 0768bf2333a7
child 10943 1079d68130e1
permissions -rw-r--r--
[dataimport] remove drop_index parameter from massive store "drop_index=False" also implied not dropping any constraints, in particular foreign keys, which meant any attempt to import entities would fail, because we only add metadata (the entities table) after the entity's insertion.

# coding: utf-8
# copyright 2003-2015 LOGILAB S.A. (Paris, FRANCE), all rights reserved.
# contact http://www.logilab.fr/ -- mailto:contact@logilab.fr
#
# This file is part of CubicWeb.
#
# CubicWeb is free software: you can redistribute it and/or modify it under the
# terms of the GNU Lesser General Public License as published by the Free
# Software Foundation, either version 2.1 of the License, or (at your option)
# any later version.
#
# CubicWeb is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more
# details.
#
# You should have received a copy of the GNU Lesser General Public License along
# with CubicWeb.  If not, see <http://www.gnu.org/licenses/>.
"""unittest for cubicweb.dataimport.pgstore"""

import datetime as DT

from six import PY2
from logilab.common.testlib import TestCase, unittest_main

from cubicweb.dataimport import pgstore


class CreateCopyFromBufferTC(TestCase):

    # test converters

    def test_convert_none(self):
        cnvt = pgstore._copyfrom_buffer_convert_None
        self.assertEqual(u'NULL', cnvt(None))

    def test_convert_number(self):
        cnvt = pgstore._copyfrom_buffer_convert_number
        self.assertEqual(u'42', cnvt(42))
        if PY2:
            self.assertEqual(u'42', cnvt(long(42)))
        self.assertEqual(u'42.42', cnvt(42.42))

    def test_convert_string(self):
        cnvt = pgstore._copyfrom_buffer_convert_string
        # simple
        self.assertEqual(u'babar', cnvt('babar'))
        # unicode
        self.assertEqual(u'éléphant', cnvt(u'éléphant'))
        # escaping
        self.assertEqual(u'babar\\tceleste\\n', cnvt(u'babar\tceleste\n'))
        self.assertEqual(u'C:\\\\new\\tC:\\\\test', cnvt(u'C:\\new\tC:\\test'))

    def test_convert_date(self):
        cnvt = pgstore._copyfrom_buffer_convert_date
        self.assertEqual('0666-01-13', cnvt(DT.date(666, 1, 13)))

    def test_convert_time(self):
        cnvt = pgstore._copyfrom_buffer_convert_time
        self.assertEqual('06:06:06.000100', cnvt(DT.time(6, 6, 6, 100)))

    def test_convert_datetime(self):
        cnvt = pgstore._copyfrom_buffer_convert_datetime
        self.assertEqual('0666-06-13 06:06:06.000000', cnvt(DT.datetime(666, 6, 13, 6, 6, 6)))

    # test buffer
    def test_create_copyfrom_buffer_tuple(self):
        l = long if PY2 else int
        data = ((42, l(42), 42.42, u'éléphant', DT.date(666, 1, 13), DT.time(6, 6, 6),
                 DT.datetime(666, 6, 13, 6, 6, 6)),
                (6, l(6), 6.6, u'babar', DT.date(2014, 1, 14), DT.time(4, 2, 1),
                 DT.datetime(2014, 1, 1, 0, 0, 0)))
        results = pgstore._create_copyfrom_buffer(data)
        # all columns
        expected = u'''42\t42\t42.42\téléphant\t0666-01-13\t06:06:06.000000\t0666-06-13 06:06:06.000000
6\t6\t6.6\tbabar\t2014-01-14\t04:02:01.000000\t2014-01-01 00:00:00.000000'''
        self.assertMultiLineEqual(expected, results.getvalue())
        # selected columns
        results = pgstore._create_copyfrom_buffer(data, columns=(1, 3, 6))
        expected = u'''42\téléphant\t0666-06-13 06:06:06.000000
6\tbabar\t2014-01-01 00:00:00.000000'''
        self.assertMultiLineEqual(expected, results.getvalue())

    def test_create_copyfrom_buffer_dict(self):
        data = (dict(integer=42, double=42.42, text=u'éléphant',
                     date=DT.datetime(666, 6, 13, 6, 6, 6)),
                dict(integer=6, double=6.6, text=u'babar',
                     date=DT.datetime(2014, 1, 1, 0, 0, 0)))
        results = pgstore._create_copyfrom_buffer(data, ('integer', 'text'))
        expected = u'''42\téléphant\n6\tbabar'''
        self.assertEqual(expected, results.getvalue())

if __name__ == '__main__':
    unittest_main()