psycopg2/tests/dbapi20.py

   1 #!/usr/bin/env python
   2 ''' Python DB API 2.0 driver compliance unit test suite.
   3
   4     This software is Public Domain and may be used without restrictions.
   5
   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."
  10
  11     -- Ian Bicking
  12 '''
  13
  14 __rcs_id__  = '$Id$'
  15 __version__ = '$Revision$'[11:-2]
  16 __author__ = 'Stuart Bishop <zen@shangri-la.dropbear.id.au>'
  17
  18 import unittest
  19 import time
  20
  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.
  25 #
  26 # Revision 1.9  2003/08/13 01:16:36  zenzen
  27 # Minor tweak from Stefan Fleiter
  28 #
  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
  32 #
  33 # Revision 1.7  2003/02/26 23:33:37  zenzen
  34 # Break out DDL into helper functions, as per request by David Rushby
  35 #
  36 # Revision 1.6  2003/02/21 03:04:33  zenzen
  37 # Stuff from Henrik Ekelund:
  38 #     added test_None
  39 #     added test_nextset & hooks
  40 #
  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
  44 #
  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
  48 # - Class renamed
  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
  60 #   nothing
  61 # - Fix bugs in test_setoutputsize_basic and test_setinputsizes
  62 #
  63
  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.
  71
  72         The 'Optional Extensions' are not yet being tested.
  73
  74         self.drivers should subclass this test, overriding setUp, tearDown,
  75         self.driver, connect_args and connect_kw_args. Class specification
  76         should be as follows:
  77
  78         import dbapi20
  79         class mytest(dbapi20.DatabaseAPI20Test):
  80            [...]
  81
  82         Don't 'import DatabaseAPI20Test from dbapi20', or you will
  83         confuse the unit tester - just 'import dbapi20'.
  84     '''
  85
  86     # The self.driver module. This should be the module where the 'connect'
  87     # method is to be found
  88     driver = None
  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
  92
  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
  97
  98     lowerfunc = 'lower' # Name of stored procedure to convert string->lowercase
  99
 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)
 104
 105     def executeDDL2(self,cursor):
 106         cursor.execute(self.ddl2)
 107
 108     def setUp(self):
 109         ''' self.drivers should override this method to perform required setup
 110             if any is necessary, such as creating the database.
 111         '''
 112         pass
 113
 114     def tearDown(self):
 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.
 118         '''
 119         con = self._connect()
 120         try:
 121             cur = con.cursor()
 122             for ddl in (self.xddl1,self.xddl2):
 123                 try:
 124                     cur.execute(ddl)
 125                     con.commit()
 126                 except self.driver.Error:
 127                     # Assume table didn't exist. Other tests will check if
 128                     # execute is busted.
 129                     pass
 130         finally:
 131             con.close()
 132
 133     def _connect(self):
 134         try:
 135             return self.driver.connect(
 136                 *self.connect_args,**self.connect_kw_args
 137                 )
 138         except AttributeError:
 139             self.fail("No connect method found in self.driver module")
 140
 141     def test_connect(self):
 142         con = self._connect()
 143         con.close()
 144
 145     def test_apilevel(self):
 146         try:
 147             # Must exist
 148             apilevel = self.driver.apilevel
 149             # Must equal 2.0
 150             self.assertEqual(apilevel,'2.0')
 151         except AttributeError:
 152             self.fail("Driver doesn't define apilevel")
 153
 154     def test_threadsafety(self):
 155         try:
 156             # Must exist
 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")
 162
 163     def test_paramstyle(self):
 164         try:
 165             # Must exist
 166             paramstyle = self.driver.paramstyle
 167             # Must be a valid value
 168             self.failUnless(paramstyle in (
 169                 'qmark','numeric','named','format','pyformat'
 170                 ))
 171         except AttributeError:
 172             self.fail("Driver doesn't define paramstyle")
 173
 174     def test_Exceptions(self):
 175         # Make sure required exceptions exist, and are in the
 176         # defined heirarchy.
 177         self.failUnless(issubclass(self.driver.Warning,StandardError))
 178         self.failUnless(issubclass(self.driver.Error,StandardError))
 179         self.failUnless(
 180             issubclass(self.driver.InterfaceError,self.driver.Error)
 181             )
 182         self.failUnless(
 183             issubclass(self.driver.DatabaseError,self.driver.Error)
 184             )
 185         self.failUnless(
 186             issubclass(self.driver.OperationalError,self.driver.Error)
 187             )
 188         self.failUnless(
 189             issubclass(self.driver.IntegrityError,self.driver.Error)
 190             )
 191         self.failUnless(
 192             issubclass(self.driver.InternalError,self.driver.Error)
 193             )
 194         self.failUnless(
 195             issubclass(self.driver.ProgrammingError,self.driver.Error)
 196             )
 197         self.failUnless(
 198             issubclass(self.driver.NotSupportedError,self.driver.Error)
 199             )
 200
 201     def test_ExceptionsAsConnectionAttributes(self):
 202         # OPTIONAL EXTENSION
 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
 207         # by default.
 208         con = self._connect()
 209         drv = self.driver
 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)
 219
 220
 221     def test_commit(self):
 222         con = self._connect()
 223         try:
 224             # Commit must work, even if it doesn't do anything
 225             con.commit()
 226         finally:
 227             con.close()
 228
 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'):
 234             try:
 235                 con.rollback()
 236             except self.driver.NotSupportedError:
 237                 pass
 238
 239     def test_cursor(self):
 240         con = self._connect()
 241         try:
 242             cur = con.cursor()
 243         finally:
 244             con.close()
 245
 246     def test_cursor_isolation(self):
 247         con = self._connect()
 248         try:
 249             # Make sure cursors created from the same connection have
 250             # the documented transaction isolation level
 251             cur1 = con.cursor()
 252             cur2 = con.cursor()
 253             self.executeDDL1(cur1)
 254             cur1.execute("insert into %sbooze values ('Victoria Bitter')" % (
 255                 self.table_prefix
 256                 ))
 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')
 262         finally:
 263             con.close()
 264
 265     def test_description(self):
 266         con = self._connect()
 267         try:
 268             cur = con.cursor()
 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)'
 273                 )
 274             cur.execute('select name from %sbooze' % self.table_prefix)
 275             self.assertEqual(len(cur.description),1,
 276                 'cursor.description describes too many columns'
 277                 )
 278             self.assertEqual(len(cur.description[0]),7,
 279                 'cursor.description[x] tuples must have 7 elements'
 280                 )
 281             self.assertEqual(cur.description[0][0].lower(),'name',
 282                 'cursor.description[x][0] must return column name'
 283                 )
 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]
 287                 )
 288
 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)'
 294                 )
 295         finally:
 296             con.close()
 297
 298     def test_rowcount(self):
 299         con = self._connect()
 300         try:
 301             cur = con.cursor()
 302             self.executeDDL1(cur)
 303             self.assertEqual(cur.rowcount,-1,
 304                 'cursor.rowcount should be -1 after executing no-result '
 305                 'statements'
 306                 )
 307             cur.execute("insert into %sbooze values ('Victoria Bitter')" % (
 308                 self.table_prefix
 309                 ))
 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'
 313                 )
 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'
 318                 )
 319             self.executeDDL2(cur)
 320             self.assertEqual(cur.rowcount,-1,
 321                 'cursor.rowcount not being reset to -1 after executing '
 322                 'no-result statements'
 323                 )
 324         finally:
 325             con.close()
 326
 327     lower_func = 'lower'
 328     def test_callproc(self):
 329         con = self._connect()
 330         try:
 331             cur = con.cursor()
 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')
 336                 r = cur.fetchall()
 337                 self.assertEqual(len(r),1,'callproc produced no result set')
 338                 self.assertEqual(len(r[0]),1,
 339                     'callproc produced invalid result set'
 340                     )
 341                 self.assertEqual(r[0][0],'foo',
 342                     'callproc produced invalid results'
 343                     )
 344         finally:
 345             con.close()
 346
 347     def test_close(self):
 348         con = self._connect()
 349         try:
 350             cur = con.cursor()
 351         finally:
 352             con.close()
 353
 354         # cursor.execute should raise an Error if called after connection
 355         # closed
 356         self.assertRaises(self.driver.Error,self.executeDDL1,cur)
 357
 358         # connection.commit should raise an Error if called after connection'
 359         # closed.'
 360         self.assertRaises(self.driver.Error,con.commit)
 361
 362         # connection.close should raise an Error if called more than once
 363         self.assertRaises(self.driver.Error,con.close)
 364
 365     def test_execute(self):
 366         con = self._connect()
 367         try:
 368             cur = con.cursor()
 369             self._paraminsert(cur)
 370         finally:
 371             con.close()
 372
 373     def _paraminsert(self,cur):
 374         self.executeDDL1(cur)
 375         cur.execute("insert into %sbooze values ('Victoria Bitter')" % (
 376             self.table_prefix
 377             ))
 378         self.failUnless(cur.rowcount in (-1,1))
 379
 380         if self.driver.paramstyle == 'qmark':
 381             cur.execute(
 382                 'insert into %sbooze values (?)' % self.table_prefix,
 383                 ("Cooper's",)
 384                 )
 385         elif self.driver.paramstyle == 'numeric':
 386             cur.execute(
 387                 'insert into %sbooze values (:1)' % self.table_prefix,
 388                 ("Cooper's",)
 389                 )
 390         elif self.driver.paramstyle == 'named':
 391             cur.execute(
 392                 'insert into %sbooze values (:beer)' % self.table_prefix,
 393                 {'beer':"Cooper's"}
 394                 )
 395         elif self.driver.paramstyle == 'format':
 396             cur.execute(
 397                 'insert into %sbooze values (%%s)' % self.table_prefix,
 398                 ("Cooper's",)
 399                 )
 400         elif self.driver.paramstyle == 'pyformat':
 401             cur.execute(
 402                 'insert into %sbooze values (%%(beer)s)' % self.table_prefix,
 403                 {'beer':"Cooper's"}
 404                 )
 405         else:
 406             self.fail('Invalid paramstyle')
 407         self.failUnless(cur.rowcount in (-1,1))
 408
 409         cur.execute('select name from %sbooze' % self.table_prefix)
 410         res = cur.fetchall()
 411         self.assertEqual(len(res),2,'cursor.fetchall returned too few rows')
 412         beers = [res[0][0],res[1][0]]
 413         beers.sort()
 414         self.assertEqual(beers[0],"Cooper's",
 415             'cursor.fetchall retrieved incorrect data, or data inserted '
 416             'incorrectly'
 417             )
 418         self.assertEqual(beers[1],"Victoria Bitter",
 419             'cursor.fetchall retrieved incorrect data, or data inserted '
 420             'incorrectly'
 421             )
 422
 423     def test_executemany(self):
 424         con = self._connect()
 425         try:
 426             cur = con.cursor()
 427             self.executeDDL1(cur)
 428             largs = [ ("Cooper's",) , ("Boag's",) ]
 429             margs = [ {'beer': "Cooper's"}, {'beer': "Boag's"} ]
 430             if self.driver.paramstyle == 'qmark':
 431                 cur.executemany(
 432                     'insert into %sbooze values (?)' % self.table_prefix,
 433                     largs
 434                     )
 435             elif self.driver.paramstyle == 'numeric':
 436                 cur.executemany(
 437                     'insert into %sbooze values (:1)' % self.table_prefix,
 438                     largs
 439                     )
 440             elif self.driver.paramstyle == 'named':
 441                 cur.executemany(
 442                     'insert into %sbooze values (:beer)' % self.table_prefix,
 443                     margs
 444                     )
 445             elif self.driver.paramstyle == 'format':
 446                 cur.executemany(
 447                     'insert into %sbooze values (%%s)' % self.table_prefix,
 448                     largs
 449                     )
 450             elif self.driver.paramstyle == 'pyformat':
 451                 cur.executemany(
 452                     'insert into %sbooze values (%%(beer)s)' % (
 453                         self.table_prefix
 454                         ),
 455                     margs
 456                     )
 457             else:
 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
 462                 )
 463             cur.execute('select name from %sbooze' % self.table_prefix)
 464             res = cur.fetchall()
 465             self.assertEqual(len(res),2,
 466                 'cursor.fetchall retrieved incorrect number of rows'
 467                 )
 468             beers = [res[0][0],res[1][0]]
 469             beers.sort()
 470             self.assertEqual(beers[0],"Boag's",'incorrect data retrieved')
 471             self.assertEqual(beers[1],"Cooper's",'incorrect data retrieved')
 472         finally:
 473             con.close()
 474
 475     def test_fetchone(self):
 476         con = self._connect()
 477         try:
 478             cur = con.cursor()
 479
 480             # cursor.fetchone should raise an Error if called before
 481             # executing a select-type query
 482             self.assertRaises(self.driver.Error,cur.fetchone)
 483
 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)
 488
 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 '
 492                 'no rows'
 493                 )
 494             self.failUnless(cur.rowcount in (-1,0))
 495
 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')" % (
 499                 self.table_prefix
 500                 ))
 501             self.assertRaises(self.driver.Error,cur.fetchone)
 502
 503             cur.execute('select name from %sbooze' % self.table_prefix)
 504             r = cur.fetchone()
 505             self.assertEqual(len(r),1,
 506                 'cursor.fetchone should have retrieved a single row'
 507                 )
 508             self.assertEqual(r[0],'Victoria Bitter',
 509                 'cursor.fetchone retrieved incorrect data'
 510                 )
 511             self.assertEqual(cur.fetchone(),None,
 512                 'cursor.fetchone should return None if no more rows available'
 513                 )
 514             self.failUnless(cur.rowcount in (-1,1))
 515         finally:
 516             con.close()
 517
 518     samples = [
 519         'Carlton Cold',
 520         'Carlton Draft',
 521         'Mountain Goat',
 522         'Redback',
 523         'Victoria Bitter',
 524         'XXXX'
 525         ]
 526
 527     def _populate(self):
 528         ''' Return a list of sql commands to setup the DB for the fetch
 529             tests.
 530         '''
 531         populate = [
 532             "insert into %sbooze values ('%s')" % (self.table_prefix,s)
 533                 for s in self.samples
 534             ]
 535         return populate
 536
 537     def test_fetchmany(self):
 538         con = self._connect()
 539         try:
 540             cur = con.cursor()
 541
 542             # cursor.fetchmany should raise an Error if called without
 543             #issuing a query
 544             self.assertRaises(self.driver.Error,cur.fetchmany,4)
 545
 546             self.executeDDL1(cur)
 547             for sql in self._populate():
 548                 cur.execute(sql)
 549
 550             cur.execute('select name from %sbooze' % self.table_prefix)
 551             r = cur.fetchmany()
 552             self.assertEqual(len(r),1,
 553                 'cursor.fetchmany retrieved incorrect number of rows, '
 554                 'default of arraysize is one.'
 555                 )
 556             cur.arraysize=10
 557             r = cur.fetchmany(3) # Should get 3 rows
 558             self.assertEqual(len(r),3,
 559                 'cursor.fetchmany retrieved incorrect number of rows'
 560                 )
 561             r = cur.fetchmany(4) # Should get 2 more
 562             self.assertEqual(len(r),2,
 563                 'cursor.fetchmany retrieved incorrect number of rows'
 564                 )
 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'
 569             )
 570             self.failUnless(cur.rowcount in (-1,6))
 571
 572             # Same as above, using cursor.arraysize
 573             cur.arraysize=4
 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'
 578                 )
 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))
 584
 585             cur.arraysize=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]
 592             rows.sort()
 593
 594             # Make sure we get the right data back out
 595             for i in range(0,6):
 596                 self.assertEqual(rows[i],self.samples[i],
 597                     'incorrect data retrieved by cursor.fetchmany'
 598                     )
 599
 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'
 604                 )
 605             self.failUnless(cur.rowcount in (-1,6))
 606
 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'
 613                 )
 614             self.failUnless(cur.rowcount in (-1,0))
 615
 616         finally:
 617             con.close()
 618
 619     def test_fetchall(self):
 620         con = self._connect()
 621         try:
 622             cur = con.cursor()
 623             # cursor.fetchall should raise an Error if called
 624             # without executing a query that may return rows (such
 625             # as a select)
 626             self.assertRaises(self.driver.Error, cur.fetchall)
 627
 628             self.executeDDL1(cur)
 629             for sql in self._populate():
 630                 cur.execute(sql)
 631
 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)
 635
 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'
 641                 )
 642             rows = [r[0] for r in rows]
 643             rows.sort()
 644             for i in range(0,len(self.samples)):
 645                 self.assertEqual(rows[i],self.samples[i],
 646                 'cursor.fetchall retrieved incorrect rows'
 647                 )
 648             rows = cur.fetchall()
 649             self.assertEqual(
 650                 len(rows),0,
 651                 'cursor.fetchall should return an empty list if called '
 652                 'after the whole result set has been fetched'
 653                 )
 654             self.failUnless(cur.rowcount in (-1,len(self.samples)))
 655
 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'
 663                 )
 664
 665         finally:
 666             con.close()
 667
 668     def test_mixedfetch(self):
 669         con = self._connect()
 670         try:
 671             cur = con.cursor()
 672             self.executeDDL1(cur)
 673             for sql in self._populate():
 674                 cur.execute(sql)
 675
 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'
 684                 )
 685             self.assertEqual(len(rows56),2,
 686                 'fetchall returned incorrect number of rows'
 687                 )
 688
 689             rows = [rows1[0]]
 690             rows.extend([rows23[0][0],rows23[1][0]])
 691             rows.append(rows4[0])
 692             rows.extend([rows56[0][0],rows56[1][0]])
 693             rows.sort()
 694             for i in range(0,len(self.samples)):
 695                 self.assertEqual(rows[i],self.samples[i],
 696                     'incorrect data retrieved or inserted'
 697                     )
 698         finally:
 699             con.close()
 700
 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"
 705         '''
 706         raise NotImplementedError,'Helper not implemented'
 707         #sql="""
 708         #    create procedure deleteme as
 709         #    begin
 710         #        select count(*) from booze
 711         #        select name from booze
 712         #    end
 713         #"""
 714         #cur.execute(sql)
 715
 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")
 720
 721     def test_nextset(self):
 722         con = self._connect()
 723         try:
 724             cur = con.cursor()
 725             if not hasattr(cur,'nextset'):
 726                 return
 727
 728             try:
 729                 self.executeDDL1(cur)
 730                 sql=self._populate()
 731                 for sql in self._populate():
 732                     cur.execute(sql)
 733
 734                 self.help_nextset_setUp(cur)
 735
 736                 cur.callproc('deleteme')
 737                 numberofrows=cur.fetchone()
 738                 assert numberofrows[0]== len(self.samples)
 739                 assert cur.nextset()
 740                 names=cur.fetchall()
 741                 assert len(names) == len(self.samples)
 742                 s=cur.nextset()
 743                 assert s == None,'No more return sets, should return None'
 744             finally:
 745                 self.help_nextset_tearDown(cur)
 746
 747         finally:
 748             con.close()
 749
 750     def test_nextset(self):
 751         raise NotImplementedError,'Drivers need to override this test'
 752
 753     def test_arraysize(self):
 754         # Not much here - rest of the tests for this are in test_fetchmany
 755         con = self._connect()
 756         try:
 757             cur = con.cursor()
 758             self.failUnless(hasattr(cur,'arraysize'),
 759                 'cursor.arraysize must be defined'
 760                 )
 761         finally:
 762             con.close()
 763
 764     def test_setinputsizes(self):
 765         con = self._connect()
 766         try:
 767             cur = con.cursor()
 768             cur.setinputsizes( (25,) )
 769             self._paraminsert(cur) # Make sure cursor still works
 770         finally:
 771             con.close()
 772
 773     def test_setoutputsize_basic(self):
 774         # Basic test is to make sure setoutputsize doesn't blow up
 775         con = self._connect()
 776         try:
 777             cur = con.cursor()
 778             cur.setoutputsize(1000)
 779             cur.setoutputsize(2000,0)
 780             self._paraminsert(cur) # Make sure the cursor still works
 781         finally:
 782             con.close()
 783
 784     def test_setoutputsize(self):
 785         # Real test for setoutputsize is driver dependant
 786         raise NotImplementedError,'Driver need to override this test'
 787
 788     def test_None(self):
 789         con = self._connect()
 790         try:
 791             cur = con.cursor()
 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)
 795             r = cur.fetchall()
 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')
 799         finally:
 800             con.close()
 801
 802     def test_Date(self):
 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))
 807
 808     def test_Time(self):
 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))
 813
 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))
 818             )
 819         # Can we assume this? API doesn't specify, but it seems implied
 820         # self.assertEqual(str(t1),str(t2))
 821
 822     def test_Binary(self):
 823         b = self.driver.Binary('Something')
 824         b = self.driver.Binary('')
 825
 826     def test_STRING(self):
 827         self.failUnless(hasattr(self.driver,'STRING'),
 828             'module.STRING must be defined'
 829             )
 830
 831     def test_BINARY(self):
 832         self.failUnless(hasattr(self.driver,'BINARY'),
 833             'module.BINARY must be defined.'
 834             )
 835
 836     def test_NUMBER(self):
 837         self.failUnless(hasattr(self.driver,'NUMBER'),
 838             'module.NUMBER must be defined.'
 839             )
 840
 841     def test_DATETIME(self):
 842         self.failUnless(hasattr(self.driver,'DATETIME'),
 843             'module.DATETIME must be defined.'
 844             )
 845
 846     def test_ROWID(self):
 847         self.failUnless(hasattr(self.driver,'ROWID'),
 848             'module.ROWID must be defined.'
 849             )
 850