2 ''' Python DB API 2.0 driver compliance unit test suite.
4 This software is Public Domain and may be used without restrictions.
6 "Now we have booze and barflies entering the discussion, plus rumours of
7 DBAs on drugs... and I won't tell you what flashes through my mind each
8 time I read the subject line with 'Anal Compliance' in it. All around
9 this is turning out to be a thoroughly unwholesome unit test."
15 __version__ = '$Revision$'[11:-2]
16 __author__ = 'Stuart Bishop <zen@shangri-la.dropbear.id.au>'
21 # $Log: dbapi20.py,v $
22 # Revision 1.10 2003/10/09 03:14:14 zenzen
23 # Add test for DB API 2.0 optional extension, where database exceptions
24 # are exposed as attributes on the Connection object.
26 # Revision 1.9 2003/08/13 01:16:36 zenzen
27 # Minor tweak from Stefan Fleiter
29 # Revision 1.8 2003/04/10 00:13:25 zenzen
30 # Changes, as per suggestions by M.-A. Lemburg
31 # - Add a table prefix, to ensure namespace collisions can always be avoided
33 # Revision 1.7 2003/02/26 23:33:37 zenzen
34 # Break out DDL into helper functions, as per request by David Rushby
36 # Revision 1.6 2003/02/21 03:04:33 zenzen
37 # Stuff from Henrik Ekelund:
39 # added test_nextset & hooks
41 # Revision 1.5 2003/02/17 22:08:43 zenzen
42 # Implement suggestions and code from Henrik Eklund - test that cursor.arraysize
43 # defaults to 1 & generic cursor.callproc test added
45 # Revision 1.4 2003/02/15 00:16:33 zenzen
46 # Changes, as per suggestions and bug reports by M.-A. Lemburg,
47 # Matthew T. Kromer, Federico Di Gregorio and Daniel Dittmar
49 # - Now a subclass of TestCase, to avoid requiring the driver stub
50 # to use multiple inheritance
51 # - Reversed the polarity of buggy test in test_description
52 # - Test exception heirarchy correctly
53 # - self.populate is now self._populate(), so if a driver stub
54 # overrides self.ddl1 this change propogates
55 # - VARCHAR columns now have a width, which will hopefully make the
56 # DDL even more portible (this will be reversed if it causes more problems)
57 # - cursor.rowcount being checked after various execute and fetchXXX methods
58 # - Check for fetchall and fetchmany returning empty lists after results
59 # are exhausted (already checking for empty lists if select retrieved
61 # - Fix bugs in test_setoutputsize_basic and test_setinputsizes
64 class DatabaseAPI20Test(unittest.TestCase):
65 ''' Test a database self.driver for DB API 2.0 compatibility.
66 This implementation tests Gadfly, but the TestCase
67 is structured so that other self.drivers can subclass this
68 test case to ensure compiliance with the DB-API. It is
69 expected that this TestCase may be expanded in the future
70 if ambiguities or edge conditions are discovered.
72 The 'Optional Extensions' are not yet being tested.
74 self.drivers should subclass this test, overriding setUp, tearDown,
75 self.driver, connect_args and connect_kw_args. Class specification
79 class mytest(dbapi20.DatabaseAPI20Test):
82 Don't 'import DatabaseAPI20Test from dbapi20', or you will
83 confuse the unit tester - just 'import dbapi20'.
86 # The self.driver module. This should be the module where the 'connect'
87 # method is to be found
89 connect_args = () # List of arguments to pass to connect
90 connect_kw_args = {} # Keyword arguments for connect
91 table_prefix = 'dbapi20test_' # If you need to specify a prefix for tables
93 ddl1 = 'create table %sbooze (name varchar(20))' % table_prefix
94 ddl2 = 'create table %sbarflys (name varchar(20))' % table_prefix
95 xddl1 = 'drop table %sbooze' % table_prefix
96 xddl2 = 'drop table %sbarflys' % table_prefix
98 lowerfunc = 'lower' # Name of stored procedure to convert string->lowercase
100 # Some drivers may need to override these helpers, for example adding
101 # a 'commit' after the execute.
102 def executeDDL1(self,cursor):
103 cursor.execute(self.ddl1)
105 def executeDDL2(self,cursor):
106 cursor.execute(self.ddl2)
109 ''' self.drivers should override this method to perform required setup
110 if any is necessary, such as creating the database.
115 ''' self.drivers should override this method to perform required cleanup
116 if any is necessary, such as deleting the test database.
117 The default drops the tables that may be created.
119 con = self._connect()
122 for ddl in (self.xddl1,self.xddl2):
126 except self.driver.Error:
127 # Assume table didn't exist. Other tests will check if
135 return self.driver.connect(
136 *self.connect_args,**self.connect_kw_args
138 except AttributeError:
139 self.fail("No connect method found in self.driver module")
141 def test_connect(self):
142 con = self._connect()
145 def test_apilevel(self):
148 apilevel = self.driver.apilevel
150 self.assertEqual(apilevel,'2.0')
151 except AttributeError:
152 self.fail("Driver doesn't define apilevel")
154 def test_threadsafety(self):
157 threadsafety = self.driver.threadsafety
158 # Must be a valid value
159 self.failUnless(threadsafety in (0,1,2,3))
160 except AttributeError:
161 self.fail("Driver doesn't define threadsafety")
163 def test_paramstyle(self):
166 paramstyle = self.driver.paramstyle
167 # Must be a valid value
168 self.failUnless(paramstyle in (
169 'qmark','numeric','named','format','pyformat'
171 except AttributeError:
172 self.fail("Driver doesn't define paramstyle")
174 def test_Exceptions(self):
175 # Make sure required exceptions exist, and are in the
177 self.failUnless(issubclass(self.driver.Warning,StandardError))
178 self.failUnless(issubclass(self.driver.Error,StandardError))
180 issubclass(self.driver.InterfaceError,self.driver.Error)
183 issubclass(self.driver.DatabaseError,self.driver.Error)
186 issubclass(self.driver.OperationalError,self.driver.Error)
189 issubclass(self.driver.IntegrityError,self.driver.Error)
192 issubclass(self.driver.InternalError,self.driver.Error)
195 issubclass(self.driver.ProgrammingError,self.driver.Error)
198 issubclass(self.driver.NotSupportedError,self.driver.Error)
201 def test_ExceptionsAsConnectionAttributes(self):
203 # Test for the optional DB API 2.0 extension, where the exceptions
204 # are exposed as attributes on the Connection object
205 # I figure this optional extension will be implemented by any
206 # driver author who is using this test suite, so it is enabled
208 con = self._connect()
210 self.failUnless(con.Warning is drv.Warning)
211 self.failUnless(con.Error is drv.Error)
212 self.failUnless(con.InterfaceError is drv.InterfaceError)
213 self.failUnless(con.DatabaseError is drv.DatabaseError)
214 self.failUnless(con.OperationalError is drv.OperationalError)
215 self.failUnless(con.IntegrityError is drv.IntegrityError)
216 self.failUnless(con.InternalError is drv.InternalError)
217 self.failUnless(con.ProgrammingError is drv.ProgrammingError)
218 self.failUnless(con.NotSupportedError is drv.NotSupportedError)
221 def test_commit(self):
222 con = self._connect()
224 # Commit must work, even if it doesn't do anything
229 def test_rollback(self):
230 con = self._connect()
231 # If rollback is defined, it should either work or throw
232 # the documented exception
233 if hasattr(con,'rollback'):
236 except self.driver.NotSupportedError:
239 def test_cursor(self):
240 con = self._connect()
246 def test_cursor_isolation(self):
247 con = self._connect()
249 # Make sure cursors created from the same connection have
250 # the documented transaction isolation level
253 self.executeDDL1(cur1)
254 cur1.execute("insert into %sbooze values ('Victoria Bitter')" % (
257 cur2.execute("select name from %sbooze" % self.table_prefix)
258 booze = cur2.fetchall()
259 self.assertEqual(len(booze),1)
260 self.assertEqual(len(booze[0]),1)
261 self.assertEqual(booze[0][0],'Victoria Bitter')
265 def test_description(self):
266 con = self._connect()
269 self.executeDDL1(cur)
270 self.assertEqual(cur.description,None,
271 'cursor.description should be none after executing a '
272 'statement that can return no rows (such as DDL)'
274 cur.execute('select name from %sbooze' % self.table_prefix)
275 self.assertEqual(len(cur.description),1,
276 'cursor.description describes too many columns'
278 self.assertEqual(len(cur.description[0]),7,
279 'cursor.description[x] tuples must have 7 elements'
281 self.assertEqual(cur.description[0][0].lower(),'name',
282 'cursor.description[x][0] must return column name'
284 self.assertEqual(cur.description[0][1],self.driver.STRING,
285 'cursor.description[x][1] must return column type. Got %r'
286 % cur.description[0][1]
289 # Make sure self.description gets reset
290 self.executeDDL2(cur)
291 self.assertEqual(cur.description,None,
292 'cursor.description not being set to None when executing '
293 'no-result statements (eg. DDL)'
298 def test_rowcount(self):
299 con = self._connect()
302 self.executeDDL1(cur)
303 self.assertEqual(cur.rowcount,-1,
304 'cursor.rowcount should be -1 after executing no-result '
307 cur.execute("insert into %sbooze values ('Victoria Bitter')" % (
310 self.failUnless(cur.rowcount in (-1,1),
311 'cursor.rowcount should == number or rows inserted, or '
312 'set to -1 after executing an insert statement'
314 cur.execute("select name from %sbooze" % self.table_prefix)
315 self.failUnless(cur.rowcount in (-1,1),
316 'cursor.rowcount should == number of rows returned, or '
317 'set to -1 after executing a select statement'
319 self.executeDDL2(cur)
320 self.assertEqual(cur.rowcount,-1,
321 'cursor.rowcount not being reset to -1 after executing '
322 'no-result statements'
328 def test_callproc(self):
329 con = self._connect()
332 if self.lower_func and hasattr(cur,'callproc'):
333 r = cur.callproc(self.lower_func,('FOO',))
334 self.assertEqual(len(r),1)
335 self.assertEqual(r[0],'FOO')
337 self.assertEqual(len(r),1,'callproc produced no result set')
338 self.assertEqual(len(r[0]),1,
339 'callproc produced invalid result set'
341 self.assertEqual(r[0][0],'foo',
342 'callproc produced invalid results'
347 def test_close(self):
348 con = self._connect()
354 # cursor.execute should raise an Error if called after connection
356 self.assertRaises(self.driver.Error,self.executeDDL1,cur)
358 # connection.commit should raise an Error if called after connection'
360 self.assertRaises(self.driver.Error,con.commit)
362 # connection.close should raise an Error if called more than once
363 self.assertRaises(self.driver.Error,con.close)
365 def test_execute(self):
366 con = self._connect()
369 self._paraminsert(cur)
373 def _paraminsert(self,cur):
374 self.executeDDL1(cur)
375 cur.execute("insert into %sbooze values ('Victoria Bitter')" % (
378 self.failUnless(cur.rowcount in (-1,1))
380 if self.driver.paramstyle == 'qmark':
382 'insert into %sbooze values (?)' % self.table_prefix,
385 elif self.driver.paramstyle == 'numeric':
387 'insert into %sbooze values (:1)' % self.table_prefix,
390 elif self.driver.paramstyle == 'named':
392 'insert into %sbooze values (:beer)' % self.table_prefix,
395 elif self.driver.paramstyle == 'format':
397 'insert into %sbooze values (%%s)' % self.table_prefix,
400 elif self.driver.paramstyle == 'pyformat':
402 'insert into %sbooze values (%%(beer)s)' % self.table_prefix,
406 self.fail('Invalid paramstyle')
407 self.failUnless(cur.rowcount in (-1,1))
409 cur.execute('select name from %sbooze' % self.table_prefix)
411 self.assertEqual(len(res),2,'cursor.fetchall returned too few rows')
412 beers = [res[0][0],res[1][0]]
414 self.assertEqual(beers[0],"Cooper's",
415 'cursor.fetchall retrieved incorrect data, or data inserted '
418 self.assertEqual(beers[1],"Victoria Bitter",
419 'cursor.fetchall retrieved incorrect data, or data inserted '
423 def test_executemany(self):
424 con = self._connect()
427 self.executeDDL1(cur)
428 largs = [ ("Cooper's",) , ("Boag's",) ]
429 margs = [ {'beer': "Cooper's"}, {'beer': "Boag's"} ]
430 if self.driver.paramstyle == 'qmark':
432 'insert into %sbooze values (?)' % self.table_prefix,
435 elif self.driver.paramstyle == 'numeric':
437 'insert into %sbooze values (:1)' % self.table_prefix,
440 elif self.driver.paramstyle == 'named':
442 'insert into %sbooze values (:beer)' % self.table_prefix,
445 elif self.driver.paramstyle == 'format':
447 'insert into %sbooze values (%%s)' % self.table_prefix,
450 elif self.driver.paramstyle == 'pyformat':
452 'insert into %sbooze values (%%(beer)s)' % (
458 self.fail('Unknown paramstyle')
459 self.failUnless(cur.rowcount in (-1,2),
460 'insert using cursor.executemany set cursor.rowcount to '
461 'incorrect value %r' % cur.rowcount
463 cur.execute('select name from %sbooze' % self.table_prefix)
465 self.assertEqual(len(res),2,
466 'cursor.fetchall retrieved incorrect number of rows'
468 beers = [res[0][0],res[1][0]]
470 self.assertEqual(beers[0],"Boag's",'incorrect data retrieved')
471 self.assertEqual(beers[1],"Cooper's",'incorrect data retrieved')
475 def test_fetchone(self):
476 con = self._connect()
480 # cursor.fetchone should raise an Error if called before
481 # executing a select-type query
482 self.assertRaises(self.driver.Error,cur.fetchone)
484 # cursor.fetchone should raise an Error if called after
485 # executing a query that cannnot return rows
486 self.executeDDL1(cur)
487 self.assertRaises(self.driver.Error,cur.fetchone)
489 cur.execute('select name from %sbooze' % self.table_prefix)
490 self.assertEqual(cur.fetchone(),None,
491 'cursor.fetchone should return None if a query retrieves '
494 self.failUnless(cur.rowcount in (-1,0))
496 # cursor.fetchone should raise an Error if called after
497 # executing a query that cannnot return rows
498 cur.execute("insert into %sbooze values ('Victoria Bitter')" % (
501 self.assertRaises(self.driver.Error,cur.fetchone)
503 cur.execute('select name from %sbooze' % self.table_prefix)
505 self.assertEqual(len(r),1,
506 'cursor.fetchone should have retrieved a single row'
508 self.assertEqual(r[0],'Victoria Bitter',
509 'cursor.fetchone retrieved incorrect data'
511 self.assertEqual(cur.fetchone(),None,
512 'cursor.fetchone should return None if no more rows available'
514 self.failUnless(cur.rowcount in (-1,1))
528 ''' Return a list of sql commands to setup the DB for the fetch
532 "insert into %sbooze values ('%s')" % (self.table_prefix,s)
533 for s in self.samples
537 def test_fetchmany(self):
538 con = self._connect()
542 # cursor.fetchmany should raise an Error if called without
544 self.assertRaises(self.driver.Error,cur.fetchmany,4)
546 self.executeDDL1(cur)
547 for sql in self._populate():
550 cur.execute('select name from %sbooze' % self.table_prefix)
552 self.assertEqual(len(r),1,
553 'cursor.fetchmany retrieved incorrect number of rows, '
554 'default of arraysize is one.'
557 r = cur.fetchmany(3) # Should get 3 rows
558 self.assertEqual(len(r),3,
559 'cursor.fetchmany retrieved incorrect number of rows'
561 r = cur.fetchmany(4) # Should get 2 more
562 self.assertEqual(len(r),2,
563 'cursor.fetchmany retrieved incorrect number of rows'
565 r = cur.fetchmany(4) # Should be an empty sequence
566 self.assertEqual(len(r),0,
567 'cursor.fetchmany should return an empty sequence after '
568 'results are exhausted'
570 self.failUnless(cur.rowcount in (-1,6))
572 # Same as above, using cursor.arraysize
574 cur.execute('select name from %sbooze' % self.table_prefix)
575 r = cur.fetchmany() # Should get 4 rows
576 self.assertEqual(len(r),4,
577 'cursor.arraysize not being honoured by fetchmany'
579 r = cur.fetchmany() # Should get 2 more
580 self.assertEqual(len(r),2)
581 r = cur.fetchmany() # Should be an empty sequence
582 self.assertEqual(len(r),0)
583 self.failUnless(cur.rowcount in (-1,6))
586 cur.execute('select name from %sbooze' % self.table_prefix)
587 rows = cur.fetchmany() # Should get all rows
588 self.failUnless(cur.rowcount in (-1,6))
589 self.assertEqual(len(rows),6)
590 self.assertEqual(len(rows),6)
591 rows = [r[0] for r in rows]
594 # Make sure we get the right data back out
596 self.assertEqual(rows[i],self.samples[i],
597 'incorrect data retrieved by cursor.fetchmany'
600 rows = cur.fetchmany() # Should return an empty list
601 self.assertEqual(len(rows),0,
602 'cursor.fetchmany should return an empty sequence if '
603 'called after the whole result set has been fetched'
605 self.failUnless(cur.rowcount in (-1,6))
607 self.executeDDL2(cur)
608 cur.execute('select name from %sbarflys' % self.table_prefix)
609 r = cur.fetchmany() # Should get empty sequence
610 self.assertEqual(len(r),0,
611 'cursor.fetchmany should return an empty sequence if '
612 'query retrieved no rows'
614 self.failUnless(cur.rowcount in (-1,0))
619 def test_fetchall(self):
620 con = self._connect()
623 # cursor.fetchall should raise an Error if called
624 # without executing a query that may return rows (such
626 self.assertRaises(self.driver.Error, cur.fetchall)
628 self.executeDDL1(cur)
629 for sql in self._populate():
632 # cursor.fetchall should raise an Error if called
633 # after executing a a statement that cannot return rows
634 self.assertRaises(self.driver.Error,cur.fetchall)
636 cur.execute('select name from %sbooze' % self.table_prefix)
637 rows = cur.fetchall()
638 self.failUnless(cur.rowcount in (-1,len(self.samples)))
639 self.assertEqual(len(rows),len(self.samples),
640 'cursor.fetchall did not retrieve all rows'
642 rows = [r[0] for r in rows]
644 for i in range(0,len(self.samples)):
645 self.assertEqual(rows[i],self.samples[i],
646 'cursor.fetchall retrieved incorrect rows'
648 rows = cur.fetchall()
651 'cursor.fetchall should return an empty list if called '
652 'after the whole result set has been fetched'
654 self.failUnless(cur.rowcount in (-1,len(self.samples)))
656 self.executeDDL2(cur)
657 cur.execute('select name from %sbarflys' % self.table_prefix)
658 rows = cur.fetchall()
659 self.failUnless(cur.rowcount in (-1,0))
660 self.assertEqual(len(rows),0,
661 'cursor.fetchall should return an empty list if '
662 'a select query returns no rows'
668 def test_mixedfetch(self):
669 con = self._connect()
672 self.executeDDL1(cur)
673 for sql in self._populate():
676 cur.execute('select name from %sbooze' % self.table_prefix)
677 rows1 = cur.fetchone()
678 rows23 = cur.fetchmany(2)
679 rows4 = cur.fetchone()
680 rows56 = cur.fetchall()
681 self.failUnless(cur.rowcount in (-1,6))
682 self.assertEqual(len(rows23),2,
683 'fetchmany returned incorrect number of rows'
685 self.assertEqual(len(rows56),2,
686 'fetchall returned incorrect number of rows'
690 rows.extend([rows23[0][0],rows23[1][0]])
691 rows.append(rows4[0])
692 rows.extend([rows56[0][0],rows56[1][0]])
694 for i in range(0,len(self.samples)):
695 self.assertEqual(rows[i],self.samples[i],
696 'incorrect data retrieved or inserted'
701 def help_nextset_setUp(self,cur):
702 ''' Should create a procedure called deleteme
703 that returns two result sets, first the
704 number of rows in booze then "name from booze"
706 raise NotImplementedError,'Helper not implemented'
708 # create procedure deleteme as
710 # select count(*) from booze
711 # select name from booze
716 def help_nextset_tearDown(self,cur):
717 'If cleaning up is needed after nextSetTest'
718 raise NotImplementedError,'Helper not implemented'
719 #cur.execute("drop procedure deleteme")
721 def test_nextset(self):
722 con = self._connect()
725 if not hasattr(cur,'nextset'):
729 self.executeDDL1(cur)
731 for sql in self._populate():
734 self.help_nextset_setUp(cur)
736 cur.callproc('deleteme')
737 numberofrows=cur.fetchone()
738 assert numberofrows[0]== len(self.samples)
741 assert len(names) == len(self.samples)
743 assert s == None,'No more return sets, should return None'
745 self.help_nextset_tearDown(cur)
750 def test_nextset(self):
751 raise NotImplementedError,'Drivers need to override this test'
753 def test_arraysize(self):
754 # Not much here - rest of the tests for this are in test_fetchmany
755 con = self._connect()
758 self.failUnless(hasattr(cur,'arraysize'),
759 'cursor.arraysize must be defined'
764 def test_setinputsizes(self):
765 con = self._connect()
768 cur.setinputsizes( (25,) )
769 self._paraminsert(cur) # Make sure cursor still works
773 def test_setoutputsize_basic(self):
774 # Basic test is to make sure setoutputsize doesn't blow up
775 con = self._connect()
778 cur.setoutputsize(1000)
779 cur.setoutputsize(2000,0)
780 self._paraminsert(cur) # Make sure the cursor still works
784 def test_setoutputsize(self):
785 # Real test for setoutputsize is driver dependant
786 raise NotImplementedError,'Driver need to override this test'
789 con = self._connect()
792 self.executeDDL1(cur)
793 cur.execute('insert into %sbooze values (NULL)' % self.table_prefix)
794 cur.execute('select name from %sbooze' % self.table_prefix)
796 self.assertEqual(len(r),1)
797 self.assertEqual(len(r[0]),1)
798 self.assertEqual(r[0][0],None,'NULL value not returned as None')
803 d1 = self.driver.Date(2002,12,25)
804 d2 = self.driver.DateFromTicks(time.mktime((2002,12,25,0,0,0,0,0,0)))
805 # Can we assume this? API doesn't specify, but it seems implied
806 # self.assertEqual(str(d1),str(d2))
809 t1 = self.driver.Time(13,45,30)
810 t2 = self.driver.TimeFromTicks(time.mktime((2001,1,1,13,45,30,0,0,0)))
811 # Can we assume this? API doesn't specify, but it seems implied
812 # self.assertEqual(str(t1),str(t2))
814 def test_Timestamp(self):
815 t1 = self.driver.Timestamp(2002,12,25,13,45,30)
816 t2 = self.driver.TimestampFromTicks(
817 time.mktime((2002,12,25,13,45,30,0,0,0))
819 # Can we assume this? API doesn't specify, but it seems implied
820 # self.assertEqual(str(t1),str(t2))
822 def test_Binary(self):
823 b = self.driver.Binary('Something')
824 b = self.driver.Binary('')
826 def test_STRING(self):
827 self.failUnless(hasattr(self.driver,'STRING'),
828 'module.STRING must be defined'
831 def test_BINARY(self):
832 self.failUnless(hasattr(self.driver,'BINARY'),
833 'module.BINARY must be defined.'
836 def test_NUMBER(self):
837 self.failUnless(hasattr(self.driver,'NUMBER'),
838 'module.NUMBER must be defined.'
841 def test_DATETIME(self):
842 self.failUnless(hasattr(self.driver,'DATETIME'),
843 'module.DATETIME must be defined.'
846 def test_ROWID(self):
847 self.failUnless(hasattr(self.driver,'ROWID'),
848 'module.ROWID must be defined.'