Package logilab :: Package common :: Package test :: Module unittest_db
[frames] | no frames]

Source Code for Module logilab.common.test.unittest_db

  1  """ 
  2  unit tests for module logilab.common.db 
  3  """ 
  4  import socket 
  5   
  6  from logilab.common.testlib import TestCase, unittest_main 
  7  from logilab.common.shellutils import getlogin 
  8  from logilab.common.db import * 
  9  from logilab.common.db import PREFERED_DRIVERS 
 10  from logilab.common.adbh import (_GenericAdvFuncHelper, _SqliteAdvFuncHelper, 
 11                                   _PGAdvFuncHelper, _MyAdvFuncHelper, 
 12                                   FunctionDescr, get_adv_func_helper, 
 13                                   auto_register_function) 
 14   
 15   
16 -class PreferedDriverTC(TestCase):
17 - def setUp(self):
18 self.drivers = {"pg":[('foo', None), ('bar', None)]} 19 self.drivers = {'pg' : ["foo", "bar"]}
20
21 - def testNormal(self):
22 set_prefered_driver('pg','bar', self.drivers) 23 self.assertEquals('bar', self.drivers['pg'][0])
24
25 - def testFailuresDb(self):
26 try: 27 set_prefered_driver('oracle','bar', self.drivers) 28 self.fail() 29 except UnknownDriver, exc: 30 self.assertEquals(exc.args[0], 'Unknown database oracle')
31
32 - def testFailuresDriver(self):
33 try: 34 set_prefered_driver('pg','baz', self.drivers) 35 self.fail() 36 except UnknownDriver, exc: 37 self.assertEquals(exc.args[0], 'Unknown module baz for pg')
38
39 - def testGlobalVar(self):
40 # XXX: Is this test supposed to be useful ? Is it supposed to test 41 # set_prefered_driver ? 42 old_drivers = PREFERED_DRIVERS['postgres'][:] 43 expected = old_drivers[:] 44 expected.insert(0, expected.pop(expected.index('pgdb'))) 45 set_prefered_driver('postgres', 'pgdb') 46 self.assertEquals(PREFERED_DRIVERS['postgres'], expected) 47 set_prefered_driver('postgres', 'psycopg') 48 # self.assertEquals(PREFERED_DRIVERS['postgres'], old_drivers) 49 expected.insert(0, expected.pop(expected.index('psycopg'))) 50 self.assertEquals(PREFERED_DRIVERS['postgres'], expected)
51 52
53 -class GetCnxTC(TestCase):
54 - def setUp(self):
55 try: 56 socket.gethostbyname('hercules') 57 except: 58 self.skip("those tests require specific DB configuration") 59 self.host = 'hercules' 60 self.db = 'template1' 61 self.user = getlogin() 62 self.passwd = getlogin()
63
64 - def testPsyco(self):
65 set_prefered_driver('postgres', 'psycopg') 66 try: 67 cnx = get_connection('postgres', 68 self.host, self.db, self.user, self.passwd, 69 quiet=1) 70 except ImportError: 71 self.skip('python-psycopg is not installed')
72
73 - def testPgdb(self):
74 set_prefered_driver('postgres', 'pgdb') 75 try: 76 cnx = get_connection('postgres', 77 self.host, self.db, self.user, self.passwd, 78 quiet=1) 79 except ImportError: 80 self.skip('python-pgsql is not installed')
81
82 - def testPgsql(self):
83 set_prefered_driver('postgres', 'pyPgSQL.PgSQL') 84 try: 85 cnx = get_connection('postgres', 86 self.host, self.db, self.user, self.passwd, 87 quiet=1) 88 except ImportError: 89 self.skip('python-pygresql is not installed')
90
91 - def testMysql(self):
92 set_prefered_driver('mysql', 'MySQLdb') 93 try: 94 cnx = get_connection('mysql', self.host, database='', user='root', 95 quiet=1) 96 except ImportError: 97 self.skip('python-mysqldb is not installed') 98 except Exception, ex: 99 # no mysql running ? 100 import MySQLdb 101 if isinstance(ex, MySQLdb.OperationalError): 102 if ex.args[0] == 1045: # find MysqlDb 103 self.skip('mysql test requires a specific configuration') 104 elif ex.args[0] != 2003: 105 raise 106 raise
107
108 - def test_connection_wrap(self):
109 """Tests the connection wrapping""" 110 try: 111 cnx = get_connection('postgres', 112 self.host, self.db, self.user, self.passwd, 113 quiet=1) 114 except ImportError: 115 self.skip('postgresql dbapi module not installed') 116 self.failIf(isinstance(cnx, PyConnection), 117 'cnx should *not* be a PyConnection instance') 118 cnx = get_connection('postgres', 119 self.host, self.db, self.user, self.passwd, 120 quiet=1, pywrap = True) 121 self.failUnless(isinstance(cnx, PyConnection), 122 'cnx should be a PyConnection instance')
123 124
125 - def test_cursor_wrap(self):
126 """Tests cursor wrapping""" 127 try: 128 cnx = get_connection('postgres', 129 self.host, self.db, self.user, self.passwd, 130 quiet=1, pywrap = True) 131 except ImportError: 132 self.skip('postgresql dbapi module not installed') 133 cursor = cnx.cursor() 134 self.failUnless(isinstance(cursor, PyCursor), 135 'cnx should be a PyCursor instance')
136 137
138 -class DBAPIAdaptersTC(TestCase):
139 """Tests DbApi adapters management""" 140
141 - def setUp(self):
142 """Memorize original PREFERED_DRIVERS""" 143 self.old_drivers = PREFERED_DRIVERS['postgres'][:] 144 self.base_functions = dict(_GenericAdvFuncHelper.FUNCTIONS) 145 self.host = 'crater.logilab.fr' 146 self.db = 'gincotest2' 147 self.user = 'adim' 148 self.passwd = 'adim'
149
150 - def tearDown(self):
151 """Reset PREFERED_DRIVERS as it was""" 152 PREFERED_DRIVERS['postgres'] = self.old_drivers 153 _GenericAdvFuncHelper.FUNCTIONS = self.base_functions 154 _PGAdvFuncHelper.FUNCTIONS = dict(self.base_functions) 155 _MyAdvFuncHelper.FUNCTIONS = dict(self.base_functions) 156 _SqliteAdvFuncHelper.FUNCTIONS = dict(self.base_functions)
157
158 - def test_raise(self):
159 self.assertRaises(UnknownDriver, get_dbapi_compliant_module, 'pougloup')
160
161 - def test_pgdb_types(self):
162 """Tests that NUMBER really wraps all number types""" 163 PREFERED_DRIVERS['postgres'] = ['pgdb'] 164 #set_prefered_driver('postgres', 'pgdb') 165 try: 166 module = get_dbapi_compliant_module('postgres') 167 except ImportError: 168 self.skip('postgresql pgdb module not installed') 169 number_types = ('int2', 'int4', 'serial', 170 'int8', 'float4', 'float8', 171 'numeric', 'bool', 'money', 'decimal') 172 for num_type in number_types: 173 yield self.assertEquals, num_type, module.NUMBER 174 yield self.assertNotEquals, 'char', module.NUMBER
175
176 - def test_pypgsql_getattr(self):
177 """Tests the getattr() delegation for pyPgSQL""" 178 set_prefered_driver('postgres', 'pyPgSQL.PgSQL') 179 try: 180 module = get_dbapi_compliant_module('postgres') 181 except ImportError: 182 self.skip('postgresql dbapi module not installed') 183 try: 184 binary = module.BINARY 185 except AttributeError, err: 186 raise 187 self.fail(str(err))
188
189 - def test_adv_func_helper(self):
190 try: 191 module = get_dbapi_compliant_module('postgres') 192 except ImportError: 193 self.skip('postgresql dbapi module not installed') 194 self.failUnless(isinstance(module.adv_func_helper, _PGAdvFuncHelper)) 195 try: 196 module = get_dbapi_compliant_module('sqlite') 197 except ImportError: 198 self.skip('sqlite dbapi module not installed') 199 self.failUnless(isinstance(module.adv_func_helper, _SqliteAdvFuncHelper))
200 201
203 class MYFUNC(FunctionDescr): 204 supported_backends = ('postgres', 'sqlite',) 205 name_mapping = {'postgres': 'MYFUNC', 206 'mysql': 'MYF', 207 'sqlite': 'SQLITE_MYFUNC'}
208 auto_register_function(MYFUNC) 209 210 pghelper = get_adv_func_helper('postgres') 211 mshelper = get_adv_func_helper('mysql') 212 slhelper = get_adv_func_helper('sqlite') 213 self.failUnless('MYFUNC' in pghelper.FUNCTIONS) 214 self.failUnless('MYFUNC' in slhelper.FUNCTIONS) 215 self.failIf('MYFUNC' in mshelper.FUNCTIONS)
216 217
218 - def test_funcname_with_different_backend_names(self):
219 class MYFUNC(FunctionDescr): 220 supported_backends = ('postgres', 'mysql', 'sqlite') 221 name_mapping = {'postgres': 'MYFUNC', 222 'mysql': 'MYF', 223 'sqlite': 'SQLITE_MYFUNC'}
224 auto_register_function(MYFUNC) 225 226 pghelper = get_adv_func_helper('postgres') 227 mshelper = get_adv_func_helper('mysql') 228 slhelper = get_adv_func_helper('sqlite') 229 self.assertEquals(pghelper.func_sqlname('MYFUNC'), 'MYFUNC') 230 self.assertEquals(mshelper.func_sqlname('MYFUNC'), 'MYF') 231 self.assertEquals(slhelper.func_sqlname('MYFUNC'), 'SQLITE_MYFUNC') 232 233 234 if __name__ == '__main__': 235 unittest_main() 236