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
18 self.drivers = {"pg":[('foo', None), ('bar', None)]}
19 self.drivers = {'pg' : ["foo", "bar"]}
20
22 set_prefered_driver('pg','bar', self.drivers)
23 self.assertEquals('bar', self.drivers['pg'][0])
24
31
38
51
52
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
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
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
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
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
100 import MySQLdb
101 if isinstance(ex, MySQLdb.OperationalError):
102 if ex.args[0] == 1045:
103 self.skip('mysql test requires a specific configuration')
104 elif ex.args[0] != 2003:
105 raise
106 raise
107
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
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
139 """Tests DbApi adapters management"""
140
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
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
160
162 """Tests that NUMBER really wraps all number types"""
163 PREFERED_DRIVERS['postgres'] = ['pgdb']
164
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
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
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
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