dataimport/test/test_pgstore.py
changeset 11057 0b59724cb3f2
parent 11052 058bb3dc685f
child 11058 23eb30449fe5
equal deleted inserted replaced
11052:058bb3dc685f 11057:0b59724cb3f2
     1 # coding: utf-8
       
     2 # copyright 2003-2015 LOGILAB S.A. (Paris, FRANCE), all rights reserved.
       
     3 # contact http://www.logilab.fr/ -- mailto:contact@logilab.fr
       
     4 #
       
     5 # This file is part of CubicWeb.
       
     6 #
       
     7 # CubicWeb is free software: you can redistribute it and/or modify it under the
       
     8 # terms of the GNU Lesser General Public License as published by the Free
       
     9 # Software Foundation, either version 2.1 of the License, or (at your option)
       
    10 # any later version.
       
    11 #
       
    12 # CubicWeb is distributed in the hope that it will be useful, but WITHOUT
       
    13 # ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
       
    14 # FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more
       
    15 # details.
       
    16 #
       
    17 # You should have received a copy of the GNU Lesser General Public License along
       
    18 # with CubicWeb.  If not, see <http://www.gnu.org/licenses/>.
       
    19 """unittest for cubicweb.dataimport.pgstore"""
       
    20 
       
    21 import datetime as DT
       
    22 
       
    23 from six import PY2
       
    24 from logilab.common.testlib import TestCase, unittest_main
       
    25 
       
    26 from cubicweb.dataimport import pgstore
       
    27 from cubicweb.devtools import testlib
       
    28 
       
    29 
       
    30 class CreateCopyFromBufferTC(TestCase):
       
    31 
       
    32     # test converters
       
    33 
       
    34     def test_convert_none(self):
       
    35         cnvt = pgstore._copyfrom_buffer_convert_None
       
    36         self.assertEqual(u'NULL', cnvt(None))
       
    37 
       
    38     def test_convert_number(self):
       
    39         cnvt = pgstore._copyfrom_buffer_convert_number
       
    40         self.assertEqual(u'42', cnvt(42))
       
    41         if PY2:
       
    42             self.assertEqual(u'42', cnvt(long(42)))
       
    43         self.assertEqual(u'42.42', cnvt(42.42))
       
    44 
       
    45     def test_convert_string(self):
       
    46         cnvt = pgstore._copyfrom_buffer_convert_string
       
    47         # simple
       
    48         self.assertEqual(u'babar', cnvt('babar'))
       
    49         # unicode
       
    50         self.assertEqual(u'éléphant', cnvt(u'éléphant'))
       
    51         # escaping
       
    52         self.assertEqual(u'babar\\tceleste\\n', cnvt(u'babar\tceleste\n'))
       
    53         self.assertEqual(u'C:\\\\new\\tC:\\\\test', cnvt(u'C:\\new\tC:\\test'))
       
    54 
       
    55     def test_convert_date(self):
       
    56         cnvt = pgstore._copyfrom_buffer_convert_date
       
    57         self.assertEqual('0666-01-13', cnvt(DT.date(666, 1, 13)))
       
    58 
       
    59     def test_convert_time(self):
       
    60         cnvt = pgstore._copyfrom_buffer_convert_time
       
    61         self.assertEqual('06:06:06.000100', cnvt(DT.time(6, 6, 6, 100)))
       
    62 
       
    63     def test_convert_datetime(self):
       
    64         cnvt = pgstore._copyfrom_buffer_convert_datetime
       
    65         self.assertEqual('0666-06-13 06:06:06.000000', cnvt(DT.datetime(666, 6, 13, 6, 6, 6)))
       
    66 
       
    67     # test buffer
       
    68     def test_create_copyfrom_buffer_tuple(self):
       
    69         l = long if PY2 else int
       
    70         data = ((42, l(42), 42.42, u'éléphant', DT.date(666, 1, 13), DT.time(6, 6, 6),
       
    71                  DT.datetime(666, 6, 13, 6, 6, 6)),
       
    72                 (6, l(6), 6.6, u'babar', DT.date(2014, 1, 14), DT.time(4, 2, 1),
       
    73                  DT.datetime(2014, 1, 1, 0, 0, 0)))
       
    74         results = pgstore._create_copyfrom_buffer(data)
       
    75         # all columns
       
    76         expected = u'''42\t42\t42.42\téléphant\t0666-01-13\t06:06:06.000000\t0666-06-13 06:06:06.000000
       
    77 6\t6\t6.6\tbabar\t2014-01-14\t04:02:01.000000\t2014-01-01 00:00:00.000000'''
       
    78         self.assertMultiLineEqual(expected, results.getvalue())
       
    79         # selected columns
       
    80         results = pgstore._create_copyfrom_buffer(data, columns=(1, 3, 6))
       
    81         expected = u'''42\téléphant\t0666-06-13 06:06:06.000000
       
    82 6\tbabar\t2014-01-01 00:00:00.000000'''
       
    83         self.assertMultiLineEqual(expected, results.getvalue())
       
    84 
       
    85     def test_create_copyfrom_buffer_dict(self):
       
    86         data = (dict(integer=42, double=42.42, text=u'éléphant',
       
    87                      date=DT.datetime(666, 6, 13, 6, 6, 6)),
       
    88                 dict(integer=6, double=6.6, text=u'babar',
       
    89                      date=DT.datetime(2014, 1, 1, 0, 0, 0)))
       
    90         results = pgstore._create_copyfrom_buffer(data, ('integer', 'text'))
       
    91         expected = u'''42\téléphant\n6\tbabar'''
       
    92         self.assertEqual(expected, results.getvalue())
       
    93 
       
    94 
       
    95 class SQLGenObjectStoreTC(testlib.CubicWebTC):
       
    96 
       
    97     def test_prepare_insert_entity(self):
       
    98         with self.admin_access.repo_cnx() as cnx:
       
    99             store = pgstore.SQLGenObjectStore(cnx)
       
   100             eid = store.prepare_insert_entity('CWUser', login=u'toto',
       
   101                                               upassword=u'pwd')
       
   102             self.assertIsNotNone(eid)
       
   103 
       
   104 
       
   105 if __name__ == '__main__':
       
   106     unittest_main()