Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Newer
Older
100755 467 lines (339 sloc) 16.819 kb
c3f6b46 @mkleehammer Import from Subversion 2.0.63; reworked versioning
authored
1 #!/usr/bin/python
2
3 # Unit tests for PostgreSQL on Linux (Fedora)
4 # This is a stripped down copy of the SQL Server tests.
5
6 import sys, os, re
7 import unittest
8 from decimal import Decimal
9 from testutils import *
10
11 _TESTSTR = '0123456789-abcdefghijklmnopqrstuvwxyz-'
12
13 def _generate_test_string(length):
14 """
15 Returns a string of composed of `seed` to make a string `length` characters long.
16
17 To enhance performance, there are 3 ways data is read, based on the length of the value, so most data types are
18 tested with 3 lengths. This function helps us generate the test data.
19
20 We use a recognizable data set instead of a single character to make it less likely that "overlap" errors will
21 be hidden and to help us manually identify where a break occurs.
22 """
23 if length <= len(_TESTSTR):
24 return _TESTSTR[:length]
25
26 c = (length + len(_TESTSTR)-1) / len(_TESTSTR)
27 v = _TESTSTR * c
28 return v[:length]
29
30 class PGTestCase(unittest.TestCase):
31
32 # These are from the C++ code. Keep them up to date.
33
34 # If we are reading a binary, string, or unicode value and do not know how large it is, we'll try reading 2K into a
35 # buffer on the stack. We then copy into a new Python object.
36 SMALL_READ = 2048
37
38 # A read guaranteed not to fit in the MAX_STACK_STACK stack buffer, but small enough to be used for varchar (4K max).
39 LARGE_READ = 4000
40
41 SMALL_STRING = _generate_test_string(SMALL_READ)
42 LARGE_STRING = _generate_test_string(LARGE_READ)
43
734f261 @mkleehammer pgtests: Added --unicode
authored
44 def __init__(self, connection_string, ansi, unicode_results, method_name):
c3f6b46 @mkleehammer Import from Subversion 2.0.63; reworked versioning
authored
45 unittest.TestCase.__init__(self, method_name)
46 self.connection_string = connection_string
734f261 @mkleehammer pgtests: Added --unicode
authored
47 self.ansi = ansi
48 self.unicode = unicode_results
c3f6b46 @mkleehammer Import from Subversion 2.0.63; reworked versioning
authored
49
50 def setUp(self):
f85004f UCS4 fixes; printf fixes;
Michael Kleehammer authored
51 self.cnxn = pyodbc.connect(self.connection_string, ansi=self.ansi)
c3f6b46 @mkleehammer Import from Subversion 2.0.63; reworked versioning
authored
52 self.cursor = self.cnxn.cursor()
53
54 for i in range(3):
55 try:
56 self.cursor.execute("drop table t%d" % i)
57 self.cnxn.commit()
58 except:
59 pass
c29b04f @mkleehammer Cursor.executemany now accepts an iterator or a generator. closes #12
authored
60
c3f6b46 @mkleehammer Import from Subversion 2.0.63; reworked versioning
authored
61 self.cnxn.rollback()
62
63
64 def tearDown(self):
65 try:
66 self.cursor.close()
67 self.cnxn.close()
68 except:
69 # If we've already closed the cursor or connection, exceptions are thrown.
70 pass
71
72 def test_datasources(self):
73 p = pyodbc.dataSources()
74 self.assert_(isinstance(p, dict))
75
76 def test_getinfo_string(self):
77 value = self.cnxn.getinfo(pyodbc.SQL_CATALOG_NAME_SEPARATOR)
78 self.assert_(isinstance(value, str))
79
80 def test_getinfo_bool(self):
81 value = self.cnxn.getinfo(pyodbc.SQL_ACCESSIBLE_TABLES)
82 self.assert_(isinstance(value, bool))
83
84 def test_getinfo_int(self):
85 value = self.cnxn.getinfo(pyodbc.SQL_DEFAULT_TXN_ISOLATION)
86 self.assert_(isinstance(value, (int, long)))
87
88 def test_getinfo_smallint(self):
89 value = self.cnxn.getinfo(pyodbc.SQL_CONCAT_NULL_BEHAVIOR)
90 self.assert_(isinstance(value, int))
91
92
93 def test_negative_float(self):
94 value = -200
95 self.cursor.execute("create table t1(n float)")
96 self.cursor.execute("insert into t1 values (?)", value)
97 result = self.cursor.execute("select n from t1").fetchone()[0]
98 self.assertEqual(value, result)
99
100
101 def _test_strtype(self, sqltype, value, colsize=None):
102 """
103 The implementation for string, Unicode, and binary tests.
104 """
105 assert colsize is None or (value is None or colsize >= len(value))
106
107 if colsize:
108 sql = "create table t1(s %s(%s))" % (sqltype, colsize)
109 else:
110 sql = "create table t1(s %s)" % sqltype
111
112 self.cursor.execute(sql)
113 self.cursor.execute("insert into t1 values(?)", value)
734f261 @mkleehammer pgtests: Added --unicode
authored
114 result = self.cursor.execute("select * from t1").fetchone()[0]
115
116 if self.unicode and value != None:
117 self.assertEqual(type(result), unicode)
118 else:
119 self.assertEqual(type(result), type(value))
c3f6b46 @mkleehammer Import from Subversion 2.0.63; reworked versioning
authored
120
121 if value is not None:
734f261 @mkleehammer pgtests: Added --unicode
authored
122 self.assertEqual(len(result), len(value))
c3f6b46 @mkleehammer Import from Subversion 2.0.63; reworked versioning
authored
123
734f261 @mkleehammer pgtests: Added --unicode
authored
124 self.assertEqual(result, value)
c3f6b46 @mkleehammer Import from Subversion 2.0.63; reworked versioning
authored
125
126 #
127 # varchar
128 #
129
130 def test_empty_varchar(self):
131 self._test_strtype('varchar', '', self.SMALL_READ)
132
133 def test_null_varchar(self):
134 self._test_strtype('varchar', None, self.SMALL_READ)
135
136 def test_large_null_varchar(self):
137 # There should not be a difference, but why not find out?
138 self._test_strtype('varchar', None, self.LARGE_READ)
139
140 def test_small_varchar(self):
141 self._test_strtype('varchar', self.SMALL_STRING, self.SMALL_READ)
142
143 def test_large_varchar(self):
144 self._test_strtype('varchar', self.LARGE_STRING, self.LARGE_READ)
145
146 def test_varchar_many(self):
147 self.cursor.execute("create table t1(c1 varchar(300), c2 varchar(300), c3 varchar(300))")
148
149 v1 = 'ABCDEFGHIJ' * 30
150 v2 = '0123456789' * 30
151 v3 = '9876543210' * 30
152
153 self.cursor.execute("insert into t1(c1, c2, c3) values (?,?,?)", v1, v2, v3);
154 row = self.cursor.execute("select c1, c2, c3 from t1").fetchone()
155
156 self.assertEqual(v1, row.c1)
157 self.assertEqual(v2, row.c2)
158 self.assertEqual(v3, row.c3)
159
160
161
162 def test_small_decimal(self):
163 # value = Decimal('1234567890987654321')
164 value = Decimal('100010') # (I use this because the ODBC docs tell us how the bytes should look in the C struct)
165 self.cursor.execute("create table t1(d numeric(19))")
166 self.cursor.execute("insert into t1 values(?)", value)
167 v = self.cursor.execute("select * from t1").fetchone()[0]
168 self.assertEqual(type(v), Decimal)
169 self.assertEqual(v, value)
170
171
172 def test_small_decimal_scale(self):
173 # The same as small_decimal, except with a different scale. This value exactly matches the ODBC documentation
174 # example in the C Data Types appendix.
175 value = '1000.10'
176 value = Decimal(value)
177 self.cursor.execute("create table t1(d numeric(20,6))")
178 self.cursor.execute("insert into t1 values(?)", value)
179 v = self.cursor.execute("select * from t1").fetchone()[0]
180 self.assertEqual(type(v), Decimal)
181 self.assertEqual(v, value)
182
183
184 def test_negative_decimal_scale(self):
185 value = Decimal('-10.0010')
186 self.cursor.execute("create table t1(d numeric(19,4))")
187 self.cursor.execute("insert into t1 values(?)", value)
188 v = self.cursor.execute("select * from t1").fetchone()[0]
189 self.assertEqual(type(v), Decimal)
190 self.assertEqual(v, value)
191
192
193 def _exec(self):
194 self.cursor.execute(self.sql)
c29b04f @mkleehammer Cursor.executemany now accepts an iterator or a generator. closes #12
authored
195
c3f6b46 @mkleehammer Import from Subversion 2.0.63; reworked versioning
authored
196 def test_close_cnxn(self):
197 """Make sure using a Cursor after closing its connection doesn't crash."""
198
199 self.cursor.execute("create table t1(id integer, s varchar(20))")
200 self.cursor.execute("insert into t1 values (?,?)", 1, 'test')
201 self.cursor.execute("select * from t1")
202
203 self.cnxn.close()
c29b04f @mkleehammer Cursor.executemany now accepts an iterator or a generator. closes #12
authored
204
c3f6b46 @mkleehammer Import from Subversion 2.0.63; reworked versioning
authored
205 # Now that the connection is closed, we expect an exception. (If the code attempts to use
206 # the HSTMT, we'll get an access violation instead.)
207 self.sql = "select * from t1"
208 self.assertRaises(pyodbc.ProgrammingError, self._exec)
209
210 def test_empty_string(self):
211 self.cursor.execute("create table t1(s varchar(20))")
212 self.cursor.execute("insert into t1 values(?)", "")
213
214 def test_fixed_str(self):
215 value = "testing"
216 self.cursor.execute("create table t1(s char(7))")
217 self.cursor.execute("insert into t1 values(?)", "testing")
218 v = self.cursor.execute("select * from t1").fetchone()[0]
734f261 @mkleehammer pgtests: Added --unicode
authored
219 self.assertEqual(type(v), self.unicode and unicode or str)
c3f6b46 @mkleehammer Import from Subversion 2.0.63; reworked versioning
authored
220 self.assertEqual(len(v), len(value)) # If we alloc'd wrong, the test below might work because of an embedded NULL
221 self.assertEqual(v, value)
222
223 def test_negative_row_index(self):
224 self.cursor.execute("create table t1(s varchar(20))")
225 self.cursor.execute("insert into t1 values(?)", "1")
226 row = self.cursor.execute("select * from t1").fetchone()
227 self.assertEquals(row[0], "1")
228 self.assertEquals(row[-1], "1")
229
230 def test_version(self):
231 self.assertEquals(3, len(pyodbc.version.split('.'))) # 1.3.1 etc.
232
233 def test_rowcount_delete(self):
234 self.assertEquals(self.cursor.rowcount, -1)
235 self.cursor.execute("create table t1(i int)")
236 count = 4
237 for i in range(count):
238 self.cursor.execute("insert into t1 values (?)", i)
239 self.cursor.execute("delete from t1")
240 self.assertEquals(self.cursor.rowcount, count)
241
242 def test_rowcount_nodata(self):
243 """
244 This represents a different code path than a delete that deleted something.
245
246 The return value is SQL_NO_DATA and code after it was causing an error. We could use SQL_NO_DATA to step over
247 the code that errors out and drop down to the same SQLRowCount code. On the other hand, we could hardcode a
248 zero return value.
249 """
250 self.cursor.execute("create table t1(i int)")
251 # This is a different code path internally.
252 self.cursor.execute("delete from t1")
253 self.assertEquals(self.cursor.rowcount, 0)
254
255 def test_rowcount_select(self):
256 self.cursor.execute("create table t1(i int)")
257 count = 4
258 for i in range(count):
259 self.cursor.execute("insert into t1 values (?)", i)
260 self.cursor.execute("select * from t1")
261 self.assertEquals(self.cursor.rowcount, 4)
262
263 # PostgreSQL driver fails here?
264 # def test_rowcount_reset(self):
265 # "Ensure rowcount is reset to -1"
c29b04f @mkleehammer Cursor.executemany now accepts an iterator or a generator. closes #12
authored
266 #
c3f6b46 @mkleehammer Import from Subversion 2.0.63; reworked versioning
authored
267 # self.cursor.execute("create table t1(i int)")
268 # count = 4
269 # for i in range(count):
270 # self.cursor.execute("insert into t1 values (?)", i)
271 # self.assertEquals(self.cursor.rowcount, 1)
c29b04f @mkleehammer Cursor.executemany now accepts an iterator or a generator. closes #12
authored
272 #
c3f6b46 @mkleehammer Import from Subversion 2.0.63; reworked versioning
authored
273 # self.cursor.execute("create table t2(i int)")
274 # self.assertEquals(self.cursor.rowcount, -1)
275
276 def test_lower_case(self):
277 "Ensure pyodbc.lowercase forces returned column names to lowercase."
278
279 # Has to be set before creating the cursor, so we must recreate self.cursor.
280
281 pyodbc.lowercase = True
282 self.cursor = self.cnxn.cursor()
283
284 self.cursor.execute("create table t1(Abc int, dEf int)")
285 self.cursor.execute("select * from t1")
286
287 names = [ t[0] for t in self.cursor.description ]
288 names.sort()
289
290 self.assertEquals(names, [ "abc", "def" ])
291
292 # Put it back so other tests don't fail.
293 pyodbc.lowercase = False
c29b04f @mkleehammer Cursor.executemany now accepts an iterator or a generator. closes #12
authored
294
c3f6b46 @mkleehammer Import from Subversion 2.0.63; reworked versioning
authored
295 def test_row_description(self):
296 """
297 Ensure Cursor.description is accessible as Row.cursor_description.
298 """
299 self.cursor = self.cnxn.cursor()
300 self.cursor.execute("create table t1(a int, b char(3))")
301 self.cnxn.commit()
302 self.cursor.execute("insert into t1 values(1, 'abc')")
303
304 row = self.cursor.execute("select * from t1").fetchone()
305 self.assertEquals(self.cursor.description, row.cursor_description)
c29b04f @mkleehammer Cursor.executemany now accepts an iterator or a generator. closes #12
authored
306
c3f6b46 @mkleehammer Import from Subversion 2.0.63; reworked versioning
authored
307
308 def test_executemany(self):
309 self.cursor.execute("create table t1(a int, b varchar(10))")
310
311 params = [ (i, str(i)) for i in range(1, 6) ]
312
313 self.cursor.executemany("insert into t1(a, b) values (?,?)", params)
314
315 # REVIEW: Without the cast, we get the following error:
316 # [07006] [unixODBC]Received an unsupported type from Postgres.;\nERROR: table "t2" does not exist (14)
317
318 count = self.cursor.execute("select cast(count(*) as int) from t1").fetchone()[0]
319 self.assertEqual(count, len(params))
320
321 self.cursor.execute("select a, b from t1 order by a")
322 rows = self.cursor.fetchall()
323 self.assertEqual(count, len(rows))
324
325 for param, row in zip(params, rows):
326 self.assertEqual(param[0], row[0])
327 self.assertEqual(param[1], row[1])
328
329
330 def test_executemany_failure(self):
331 """
332 Ensure that an exception is raised if one query in an executemany fails.
333 """
334 self.cursor.execute("create table t1(a int, b varchar(10))")
335
336 params = [ (1, 'good'),
337 ('error', 'not an int'),
338 (3, 'good') ]
c29b04f @mkleehammer Cursor.executemany now accepts an iterator or a generator. closes #12
authored
339
c3f6b46 @mkleehammer Import from Subversion 2.0.63; reworked versioning
authored
340 self.failUnlessRaises(pyodbc.Error, self.cursor.executemany, "insert into t1(a, b) value (?, ?)", params)
341
c29b04f @mkleehammer Cursor.executemany now accepts an iterator or a generator. closes #12
authored
342
343 def test_executemany_generator(self):
344 self.cursor.execute("create table t1(a int)")
345
346 self.cursor.executemany("insert into t1(a) values (?)", ((i,) for i in range(4)))
347
348 row = self.cursor.execute("select min(a) mina, max(a) maxa from t1").fetchone()
349
350 self.assertEqual(row.mina, 0)
351 self.assertEqual(row.maxa, 3)
352
353
354 def test_executemany_iterator(self):
355 self.cursor.execute("create table t1(a int)")
356
357 values = [ (i,) for i in range(4) ]
358
359 self.cursor.executemany("insert into t1(a) values (?)", iter(values))
360
361 row = self.cursor.execute("select min(a) mina, max(a) maxa from t1").fetchone()
362
363 self.assertEqual(row.mina, 0)
364 self.assertEqual(row.maxa, 3)
365
366
c3f6b46 @mkleehammer Import from Subversion 2.0.63; reworked versioning
authored
367 def test_row_slicing(self):
368 self.cursor.execute("create table t1(a int, b int, c int, d int)");
369 self.cursor.execute("insert into t1 values(1,2,3,4)")
370
371 row = self.cursor.execute("select * from t1").fetchone()
372
373 result = row[:]
374 self.failUnless(result is row)
375
376 result = row[:-1]
377 self.assertEqual(result, (1,2,3))
378
379 result = row[0:4]
380 self.failUnless(result is row)
381
382
383 def test_row_repr(self):
384 self.cursor.execute("create table t1(a int, b int, c int, d int)");
385 self.cursor.execute("insert into t1 values(1,2,3,4)")
386
387 row = self.cursor.execute("select * from t1").fetchone()
388
389 result = str(row)
390 self.assertEqual(result, "(1, 2, 3, 4)")
391
392 result = str(row[:-1])
393 self.assertEqual(result, "(1, 2, 3)")
394
395 result = str(row[:1])
396 self.assertEqual(result, "(1,)")
397
398
b3e8648 @mkleehammer Added pgtests test_int_limits
authored
399 def test_int_limits(self):
400 values = [ (-sys.maxint - 1), -1, 0, 1, 3230392212, sys.maxint ]
401
402 self.cursor.execute("create table t1(a bigint)")
403
404 for value in values:
405 self.cursor.execute("delete from t1")
406 self.cursor.execute("insert into t1 values(?)", value)
407 v = self.cursor.execute("select a from t1").fetchone()[0]
408 self.assertEqual(v, value)
409
410
c3f6b46 @mkleehammer Import from Subversion 2.0.63; reworked versioning
authored
411 def main():
412 from optparse import OptionParser
413 parser = OptionParser(usage="usage: %prog [options] connection_string")
414 parser.add_option("-v", "--verbose", action="count", help="Increment test verbosity (can be used multiple times)")
415 parser.add_option("-d", "--debug", action="store_true", default=False, help="Print debugging items")
416 parser.add_option("-t", "--test", help="Run only the named test")
f85004f UCS4 fixes; printf fixes;
Michael Kleehammer authored
417 parser.add_option('-a', '--ansi', help='ANSI only', default=False, action='store_true')
734f261 @mkleehammer pgtests: Added --unicode
authored
418 parser.add_option('-u', '--unicode', help='Expect results in Unicode', default=False, action='store_true')
c3f6b46 @mkleehammer Import from Subversion 2.0.63; reworked versioning
authored
419
420 (options, args) = parser.parse_args()
421
422 if len(args) > 1:
423 parser.error('Only one argument is allowed. Do you need quotes around the connection string?')
424
425 if not args:
426 connection_string = load_setup_connection_string('pgtests')
427
428 if not connection_string:
429 parser.print_help()
430 raise SystemExit()
431 else:
432 connection_string = args[0]
433
434 if options.verbose:
f85004f UCS4 fixes; printf fixes;
Michael Kleehammer authored
435 cnxn = pyodbc.connect(connection_string, ansi=options.ansi)
c3f6b46 @mkleehammer Import from Subversion 2.0.63; reworked versioning
authored
436 print 'library:', os.path.abspath(pyodbc.__file__)
437 print 'odbc: %s' % cnxn.getinfo(pyodbc.SQL_ODBC_VER)
438 print 'driver: %s %s' % (cnxn.getinfo(pyodbc.SQL_DRIVER_NAME), cnxn.getinfo(pyodbc.SQL_DRIVER_VER))
439 print 'driver supports ODBC version %s' % cnxn.getinfo(pyodbc.SQL_DRIVER_ODBC_VER)
f85004f UCS4 fixes; printf fixes;
Michael Kleehammer authored
440 print 'unicode:', pyodbc.UNICODE_SIZE, 'sqlwchar:', pyodbc.SQLWCHAR_SIZE
c3f6b46 @mkleehammer Import from Subversion 2.0.63; reworked versioning
authored
441 cnxn.close()
442
443 if options.test:
444 # Run a single test
445 if not options.test.startswith('test_'):
446 options.test = 'test_%s' % (options.test)
447
734f261 @mkleehammer pgtests: Added --unicode
authored
448 s = unittest.TestSuite([ PGTestCase(connection_string, options.ansi, options.unicode, options.test) ])
c3f6b46 @mkleehammer Import from Subversion 2.0.63; reworked versioning
authored
449 else:
450 # Run all tests in the class
451
452 methods = [ m for m in dir(PGTestCase) if m.startswith('test_') ]
453 methods.sort()
734f261 @mkleehammer pgtests: Added --unicode
authored
454 s = unittest.TestSuite([ PGTestCase(connection_string, options.ansi, options.unicode, m) for m in methods ])
c3f6b46 @mkleehammer Import from Subversion 2.0.63; reworked versioning
authored
455
456 testRunner = unittest.TextTestRunner(verbosity=options.verbose)
457 result = testRunner.run(s)
458
459 if __name__ == '__main__':
460
461 # Add the build directory to the path so we're testing the latest build, not the installed version.
462
463 add_to_path()
464
465 import pyodbc
466 main()
Something went wrong with that request. Please try again.