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 __version__ = "$Revision: 1.15.0 $"[11:-2] 15 __author__ = "Stuart Bishop <stuart@stuartbishop.net>" 16 17 import time 18 import unittest 19 20 # set this to "True" to follow API 2.0 to the letter 21 TEST_FOR_NON_IDEMPOTENT_CLOSE = False 22 23 # Revision 1.15 2019/11/22 00:50:00 kf7xm 24 # Make Turn off IDEMPOTENT_CLOSE a proper skipTest 25 26 # Revision 1.14 2013/05/20 11:02:05 kf7xm 27 # Add a literal string to the format insertion test to catch trivial re-format algorithms 28 29 # Revision 1.13 2013/05/08 14:31:50 kf7xm 30 # Quick switch to Turn off IDEMPOTENT_CLOSE test. Also: Silence teardown failure 31 32 33 # Revision 1.12 2009/02/06 03:35:11 kf7xm 34 # Tested okay with Python 3.0, includes last minute patches from Mark H. 35 # 36 # Revision 1.1.1.1.2.1 2008/09/20 19:54:59 rupole 37 # Include latest changes from main branch 38 # Updates for py3k 39 # 40 # Revision 1.11 2005/01/02 02:41:01 zenzen 41 # Update author email address 42 # 43 # Revision 1.10 2003/10/09 03:14:14 zenzen 44 # Add test for DB API 2.0 optional extension, where database exceptions 45 # are exposed as attributes on the Connection object. 46 # 47 # Revision 1.9 2003/08/13 01:16:36 zenzen 48 # Minor tweak from Stefan Fleiter 49 # 50 # Revision 1.8 2003/04/10 00:13:25 zenzen 51 # Changes, as per suggestions by M.-A. Lemburg 52 # - Add a table prefix, to ensure namespace collisions can always be avoided 53 # 54 # Revision 1.7 2003/02/26 23:33:37 zenzen 55 # Break out DDL into helper functions, as per request by David Rushby 56 # 57 # Revision 1.6 2003/02/21 03:04:33 zenzen 58 # Stuff from Henrik Ekelund: 59 # added test_None 60 # added test_nextset & hooks 61 # 62 # Revision 1.5 2003/02/17 22:08:43 zenzen 63 # Implement suggestions and code from Henrik Eklund - test that cursor.arraysize 64 # defaults to 1 & generic cursor.callproc test added 65 # 66 # Revision 1.4 2003/02/15 00:16:33 zenzen 67 # Changes, as per suggestions and bug reports by M.-A. Lemburg, 68 # Matthew T. Kromer, Federico Di Gregorio and Daniel Dittmar 69 # - Class renamed 70 # - Now a subclass of TestCase, to avoid requiring the driver stub 71 # to use multiple inheritance 72 # - Reversed the polarity of buggy test in test_description 73 # - Test exception hierarchy correctly 74 # - self.populate is now self._populate(), so if a driver stub 75 # overrides self.ddl1 this change propogates 76 # - VARCHAR columns now have a width, which will hopefully make the 77 # DDL even more portible (this will be reversed if it causes more problems) 78 # - cursor.rowcount being checked after various execute and fetchXXX methods 79 # - Check for fetchall and fetchmany returning empty lists after results 80 # are exhausted (already checking for empty lists if select retrieved 81 # nothing 82 # - Fix bugs in test_setoutputsize_basic and test_setinputsizes 83 # 84 85 86 class DatabaseAPI20Test(unittest.TestCase): 87 """Test a database self.driver for DB API 2.0 compatibility. 88 This implementation tests Gadfly, but the TestCase 89 is structured so that other self.drivers can subclass this 90 test case to ensure compiliance with the DB-API. It is 91 expected that this TestCase may be expanded in the future 92 if ambiguities or edge conditions are discovered. 93 94 The 'Optional Extensions' are not yet being tested. 95 96 self.drivers should subclass this test, overriding setUp, tearDown, 97 self.driver, connect_args and connect_kw_args. Class specification 98 should be as follows: 99 100 import dbapi20 101 class mytest(dbapi20.DatabaseAPI20Test): 102 [...] 103 104 Don't 'import DatabaseAPI20Test from dbapi20', or you will 105 confuse the unit tester - just 'import dbapi20'. 106 """ 107 108 # The self.driver module. This should be the module where the 'connect' 109 # method is to be found 110 driver = None 111 connect_args = () # List of arguments to pass to connect 112 connect_kw_args = {} # Keyword arguments for connect 113 table_prefix = "dbapi20test_" # If you need to specify a prefix for tables 114 115 ddl1 = "create table %sbooze (name varchar(20))" % table_prefix 116 ddl2 = "create table %sbarflys (name varchar(20), drink varchar(30))" % table_prefix 117 xddl1 = "drop table %sbooze" % table_prefix 118 xddl2 = "drop table %sbarflys" % table_prefix 119 120 lowerfunc = "lower" # Name of stored procedure to convert string->lowercase 121 122 # Some drivers may need to override these helpers, for example adding 123 # a 'commit' after the execute. 124 def executeDDL1(self, cursor): 125 cursor.execute(self.ddl1) 126 127 def executeDDL2(self, cursor): 128 cursor.execute(self.ddl2) 129 130 def setUp(self): 131 """self.drivers should override this method to perform required setup 132 if any is necessary, such as creating the database. 133 """ 134 pass 135 136 def tearDown(self): 137 """self.drivers should override this method to perform required cleanup 138 if any is necessary, such as deleting the test database. 139 The default drops the tables that may be created. 140 """ 141 try: 142 con = self._connect() 143 try: 144 cur = con.cursor() 145 for ddl in (self.xddl1, self.xddl2): 146 try: 147 cur.execute(ddl) 148 con.commit() 149 except self.driver.Error: 150 # Assume table didn't exist. Other tests will check if 151 # execute is busted. 152 pass 153 finally: 154 con.close() 155 except Exception: 156 pass 157 158 def _connect(self): 159 try: 160 r = self.driver.connect(*self.connect_args, **self.connect_kw_args) 161 except AttributeError: 162 self.fail("No connect method found in self.driver module") 163 return r 164 165 def test_connect(self): 166 con = self._connect() 167 con.close() 168 169 def test_apilevel(self): 170 try: 171 # Must exist 172 apilevel = self.driver.apilevel 173 # Must equal 2.0 174 self.assertEqual(apilevel, "2.0") 175 except AttributeError: 176 self.fail("Driver doesn't define apilevel") 177 178 def test_threadsafety(self): 179 try: 180 # Must exist 181 threadsafety = self.driver.threadsafety 182 # Must be a valid value 183 self.assertTrue(threadsafety in (0, 1, 2, 3)) 184 except AttributeError: 185 self.fail("Driver doesn't define threadsafety") 186 187 def test_paramstyle(self): 188 try: 189 # Must exist 190 paramstyle = self.driver.paramstyle 191 # Must be a valid value 192 self.assertTrue( 193 paramstyle in ("qmark", "numeric", "named", "format", "pyformat") 194 ) 195 except AttributeError: 196 self.fail("Driver doesn't define paramstyle") 197 198 def test_Exceptions(self): 199 # Make sure required exceptions exist, and are in the defined hierarchy. 200 self.assertTrue(issubclass(self.driver.Warning, Exception)) 201 self.assertTrue(issubclass(self.driver.Error, Exception)) 202 203 self.assertTrue(issubclass(self.driver.InterfaceError, self.driver.Error)) 204 self.assertTrue(issubclass(self.driver.DatabaseError, self.driver.Error)) 205 self.assertTrue(issubclass(self.driver.OperationalError, self.driver.Error)) 206 self.assertTrue(issubclass(self.driver.IntegrityError, self.driver.Error)) 207 self.assertTrue(issubclass(self.driver.InternalError, self.driver.Error)) 208 self.assertTrue(issubclass(self.driver.ProgrammingError, self.driver.Error)) 209 self.assertTrue(issubclass(self.driver.NotSupportedError, self.driver.Error)) 210 211 def test_ExceptionsAsConnectionAttributes(self): 212 # OPTIONAL EXTENSION 213 # Test for the optional DB API 2.0 extension, where the exceptions 214 # are exposed as attributes on the Connection object 215 # I figure this optional extension will be implemented by any 216 # driver author who is using this test suite, so it is enabled 217 # by default. 218 con = self._connect() 219 drv = self.driver 220 self.assertTrue(con.Warning is drv.Warning) 221 self.assertTrue(con.Error is drv.Error) 222 self.assertTrue(con.InterfaceError is drv.InterfaceError) 223 self.assertTrue(con.DatabaseError is drv.DatabaseError) 224 self.assertTrue(con.OperationalError is drv.OperationalError) 225 self.assertTrue(con.IntegrityError is drv.IntegrityError) 226 self.assertTrue(con.InternalError is drv.InternalError) 227 self.assertTrue(con.ProgrammingError is drv.ProgrammingError) 228 self.assertTrue(con.NotSupportedError is drv.NotSupportedError) 229 230 def test_commit(self): 231 con = self._connect() 232 try: 233 # Commit must work, even if it doesn't do anything 234 con.commit() 235 finally: 236 con.close() 237 238 def test_rollback(self): 239 con = self._connect() 240 # If rollback is defined, it should either work or throw 241 # the documented exception 242 if hasattr(con, "rollback"): 243 try: 244 con.rollback() 245 except self.driver.NotSupportedError: 246 pass 247 248 def test_cursor(self): 249 con = self._connect() 250 try: 251 cur = con.cursor() 252 finally: 253 con.close() 254 255 def test_cursor_isolation(self): 256 con = self._connect() 257 try: 258 # Make sure cursors created from the same connection have 259 # the documented transaction isolation level 260 cur1 = con.cursor() 261 cur2 = con.cursor() 262 self.executeDDL1(cur1) 263 cur1.execute( 264 "insert into %sbooze values ('Victoria Bitter')" % (self.table_prefix) 265 ) 266 cur2.execute("select name from %sbooze" % self.table_prefix) 267 booze = cur2.fetchall() 268 self.assertEqual(len(booze), 1) 269 self.assertEqual(len(booze[0]), 1) 270 self.assertEqual(booze[0][0], "Victoria Bitter") 271 finally: 272 con.close() 273 274 def test_description(self): 275 con = self._connect() 276 try: 277 cur = con.cursor() 278 self.executeDDL1(cur) 279 self.assertEqual( 280 cur.description, 281 None, 282 "cursor.description should be none after executing a " 283 "statement that can return no rows (such as DDL)", 284 ) 285 cur.execute("select name from %sbooze" % self.table_prefix) 286 self.assertEqual( 287 len(cur.description), 1, "cursor.description describes too many columns" 288 ) 289 self.assertEqual( 290 len(cur.description[0]), 291 7, 292 "cursor.description[x] tuples must have 7 elements", 293 ) 294 self.assertEqual( 295 cur.description[0][0].lower(), 296 "name", 297 "cursor.description[x][0] must return column name", 298 ) 299 self.assertEqual( 300 cur.description[0][1], 301 self.driver.STRING, 302 "cursor.description[x][1] must return column type. Got %r" 303 % cur.description[0][1], 304 ) 305 306 # Make sure self.description gets reset 307 self.executeDDL2(cur) 308 self.assertEqual( 309 cur.description, 310 None, 311 "cursor.description not being set to None when executing " 312 "no-result statements (eg. DDL)", 313 ) 314 finally: 315 con.close() 316 317 def test_rowcount(self): 318 con = self._connect() 319 try: 320 cur = con.cursor() 321 self.executeDDL1(cur) 322 self.assertTrue( 323 cur.rowcount in (-1, 0), # Bug #543885 324 "cursor.rowcount should be -1 or 0 after executing no-result " 325 "statements", 326 ) 327 cur.execute( 328 "insert into %sbooze values ('Victoria Bitter')" % (self.table_prefix) 329 ) 330 self.assertTrue( 331 cur.rowcount in (-1, 1), 332 "cursor.rowcount should == number or rows inserted, or " 333 "set to -1 after executing an insert statement", 334 ) 335 cur.execute("select name from %sbooze" % self.table_prefix) 336 self.assertTrue( 337 cur.rowcount in (-1, 1), 338 "cursor.rowcount should == number of rows returned, or " 339 "set to -1 after executing a select statement", 340 ) 341 self.executeDDL2(cur) 342 self.assertEqual( 343 cur.rowcount, 344 -1, 345 "cursor.rowcount not being reset to -1 after executing " 346 "no-result statements", 347 ) 348 finally: 349 con.close() 350 351 lower_func = "lower" 352 353 def test_callproc(self): 354 con = self._connect() 355 try: 356 cur = con.cursor() 357 if self.lower_func and hasattr(cur, "callproc"): 358 r = cur.callproc(self.lower_func, ("FOO",)) 359 self.assertEqual(len(r), 1) 360 self.assertEqual(r[0], "FOO") 361 r = cur.fetchall() 362 self.assertEqual(len(r), 1, "callproc produced no result set") 363 self.assertEqual(len(r[0]), 1, "callproc produced invalid result set") 364 self.assertEqual(r[0][0], "foo", "callproc produced invalid results") 365 finally: 366 con.close() 367 368 def test_close(self): 369 con = self._connect() 370 try: 371 cur = con.cursor() 372 finally: 373 con.close() 374 375 # cursor.execute should raise an Error if called after connection 376 # closed 377 self.assertRaises(self.driver.Error, self.executeDDL1, cur) 378 379 # connection.commit should raise an Error if called after connection' 380 # closed.' 381 self.assertRaises(self.driver.Error, con.commit) 382 383 # connection.close should raise an Error if called more than once 384 #!!! reasonable persons differ about the usefulness of this test and this feature !!! 385 if TEST_FOR_NON_IDEMPOTENT_CLOSE: 386 self.assertRaises(self.driver.Error, con.close) 387 else: 388 self.skipTest( 389 "Non-idempotent close is considered a bad thing by some people." 390 ) 391 392 def test_execute(self): 393 con = self._connect() 394 try: 395 cur = con.cursor() 396 self._paraminsert(cur) 397 finally: 398 con.close() 399 400 def _paraminsert(self, cur): 401 self.executeDDL2(cur) 402 cur.execute( 403 "insert into %sbarflys values ('Victoria Bitter', 'thi%%s :may ca%%(u)se? troub:1e')" 404 % (self.table_prefix) 405 ) 406 self.assertTrue(cur.rowcount in (-1, 1)) 407 408 if self.driver.paramstyle == "qmark": 409 cur.execute( 410 "insert into %sbarflys values (?, 'thi%%s :may ca%%(u)se? troub:1e')" 411 % self.table_prefix, 412 ("Cooper's",), 413 ) 414 elif self.driver.paramstyle == "numeric": 415 cur.execute( 416 "insert into %sbarflys values (:1, 'thi%%s :may ca%%(u)se? troub:1e')" 417 % self.table_prefix, 418 ("Cooper's",), 419 ) 420 elif self.driver.paramstyle == "named": 421 cur.execute( 422 "insert into %sbarflys values (:beer, 'thi%%s :may ca%%(u)se? troub:1e')" 423 % self.table_prefix, 424 {"beer": "Cooper's"}, 425 ) 426 elif self.driver.paramstyle == "format": 427 cur.execute( 428 "insert into %sbarflys values (%%s, 'thi%%s :may ca%%(u)se? troub:1e')" 429 % self.table_prefix, 430 ("Cooper's",), 431 ) 432 elif self.driver.paramstyle == "pyformat": 433 cur.execute( 434 "insert into %sbarflys values (%%(beer)s, 'thi%%s :may ca%%(u)se? troub:1e')" 435 % self.table_prefix, 436 {"beer": "Cooper's"}, 437 ) 438 else: 439 self.fail("Invalid paramstyle") 440 self.assertTrue(cur.rowcount in (-1, 1)) 441 442 cur.execute("select name, drink from %sbarflys" % self.table_prefix) 443 res = cur.fetchall() 444 self.assertEqual(len(res), 2, "cursor.fetchall returned too few rows") 445 beers = [res[0][0], res[1][0]] 446 beers.sort() 447 self.assertEqual( 448 beers[0], 449 "Cooper's", 450 "cursor.fetchall retrieved incorrect data, or data inserted incorrectly", 451 ) 452 self.assertEqual( 453 beers[1], 454 "Victoria Bitter", 455 "cursor.fetchall retrieved incorrect data, or data inserted incorrectly", 456 ) 457 trouble = "thi%s :may ca%(u)se? troub:1e" 458 self.assertEqual( 459 res[0][1], 460 trouble, 461 "cursor.fetchall retrieved incorrect data, or data inserted " 462 f"incorrectly. Got={res[0][1]!r}, Expected={trouble!r}", 463 ) 464 self.assertEqual( 465 res[1][1], 466 trouble, 467 "cursor.fetchall retrieved incorrect data, or data inserted " 468 f"incorrectly. Got={res[1][1]!r}, Expected={trouble!r}", 469 ) 470 471 def test_executemany(self): 472 con = self._connect() 473 try: 474 cur = con.cursor() 475 self.executeDDL1(cur) 476 largs = [("Cooper's",), ("Boag's",)] 477 margs = [{"beer": "Cooper's"}, {"beer": "Boag's"}] 478 if self.driver.paramstyle == "qmark": 479 cur.executemany( 480 "insert into %sbooze values (?)" % self.table_prefix, largs 481 ) 482 elif self.driver.paramstyle == "numeric": 483 cur.executemany( 484 "insert into %sbooze values (:1)" % self.table_prefix, largs 485 ) 486 elif self.driver.paramstyle == "named": 487 cur.executemany( 488 "insert into %sbooze values (:beer)" % self.table_prefix, margs 489 ) 490 elif self.driver.paramstyle == "format": 491 cur.executemany( 492 "insert into %sbooze values (%%s)" % self.table_prefix, largs 493 ) 494 elif self.driver.paramstyle == "pyformat": 495 cur.executemany( 496 "insert into %sbooze values (%%(beer)s)" % (self.table_prefix), 497 margs, 498 ) 499 else: 500 self.fail("Unknown paramstyle") 501 self.assertTrue( 502 cur.rowcount in (-1, 2), 503 "insert using cursor.executemany set cursor.rowcount to " 504 "incorrect value %r" % cur.rowcount, 505 ) 506 cur.execute("select name from %sbooze" % self.table_prefix) 507 res = cur.fetchall() 508 self.assertEqual( 509 len(res), 2, "cursor.fetchall retrieved incorrect number of rows" 510 ) 511 beers = [res[0][0], res[1][0]] 512 beers.sort() 513 self.assertEqual( 514 beers[0], "Boag's", 'incorrect data "%s" retrieved' % beers[0] 515 ) 516 self.assertEqual(beers[1], "Cooper's", "incorrect data retrieved") 517 finally: 518 con.close() 519 520 def test_fetchone(self): 521 con = self._connect() 522 try: 523 cur = con.cursor() 524 525 # cursor.fetchone should raise an Error if called before 526 # executing a select-type query 527 self.assertRaises(self.driver.Error, cur.fetchone) 528 529 # cursor.fetchone should raise an Error if called after 530 # executing a query that cannnot return rows 531 self.executeDDL1(cur) 532 self.assertRaises(self.driver.Error, cur.fetchone) 533 534 cur.execute("select name from %sbooze" % self.table_prefix) 535 self.assertEqual( 536 cur.fetchone(), 537 None, 538 "cursor.fetchone should return None if a query retrieves no rows", 539 ) 540 self.assertTrue(cur.rowcount in (-1, 0)) 541 542 # cursor.fetchone should raise an Error if called after 543 # executing a query that cannnot return rows 544 cur.execute( 545 "insert into %sbooze values ('Victoria Bitter')" % (self.table_prefix) 546 ) 547 self.assertRaises(self.driver.Error, cur.fetchone) 548 549 cur.execute("select name from %sbooze" % self.table_prefix) 550 r = cur.fetchone() 551 self.assertEqual( 552 len(r), 1, "cursor.fetchone should have retrieved a single row" 553 ) 554 self.assertEqual( 555 r[0], "Victoria Bitter", "cursor.fetchone retrieved incorrect data" 556 ) 557 self.assertEqual( 558 cur.fetchone(), 559 None, 560 "cursor.fetchone should return None if no more rows available", 561 ) 562 self.assertTrue(cur.rowcount in (-1, 1)) 563 finally: 564 con.close() 565 566 samples = [ 567 "Carlton Cold", 568 "Carlton Draft", 569 "Mountain Goat", 570 "Redback", 571 "Victoria Bitter", 572 "XXXX", 573 ] 574 575 def _populate(self): 576 """Return a list of sql commands to setup the DB for the fetch 577 tests. 578 """ 579 populate = [ 580 "insert into %sbooze values ('%s')" % (self.table_prefix, s) 581 for s in self.samples 582 ] 583 return populate 584 585 def test_fetchmany(self): 586 con = self._connect() 587 try: 588 cur = con.cursor() 589 590 # cursor.fetchmany should raise an Error if called without 591 # issuing a query 592 self.assertRaises(self.driver.Error, cur.fetchmany, 4) 593 594 self.executeDDL1(cur) 595 for sql in self._populate(): 596 cur.execute(sql) 597 598 cur.execute("select name from %sbooze" % self.table_prefix) 599 r = cur.fetchmany() 600 self.assertEqual( 601 len(r), 602 1, 603 "cursor.fetchmany retrieved incorrect number of rows, " 604 "default of arraysize is one.", 605 ) 606 cur.arraysize = 10 607 r = cur.fetchmany(3) # Should get 3 rows 608 self.assertEqual( 609 len(r), 3, "cursor.fetchmany retrieved incorrect number of rows" 610 ) 611 r = cur.fetchmany(4) # Should get 2 more 612 self.assertEqual( 613 len(r), 2, "cursor.fetchmany retrieved incorrect number of rows" 614 ) 615 r = cur.fetchmany(4) # Should be an empty sequence 616 self.assertEqual( 617 len(r), 618 0, 619 "cursor.fetchmany should return an empty sequence after " 620 "results are exhausted", 621 ) 622 self.assertTrue(cur.rowcount in (-1, 6)) 623 624 # Same as above, using cursor.arraysize 625 cur.arraysize = 4 626 cur.execute("select name from %sbooze" % self.table_prefix) 627 r = cur.fetchmany() # Should get 4 rows 628 self.assertEqual( 629 len(r), 4, "cursor.arraysize not being honoured by fetchmany" 630 ) 631 r = cur.fetchmany() # Should get 2 more 632 self.assertEqual(len(r), 2) 633 r = cur.fetchmany() # Should be an empty sequence 634 self.assertEqual(len(r), 0) 635 self.assertTrue(cur.rowcount in (-1, 6)) 636 637 cur.arraysize = 6 638 cur.execute("select name from %sbooze" % self.table_prefix) 639 rows = cur.fetchmany() # Should get all rows 640 self.assertTrue(cur.rowcount in (-1, 6)) 641 self.assertEqual(len(rows), 6) 642 self.assertEqual(len(rows), 6) 643 rows = [r[0] for r in rows] 644 rows.sort() 645 646 # Make sure we get the right data back out 647 for i in range(0, 6): 648 self.assertEqual( 649 rows[i], 650 self.samples[i], 651 "incorrect data retrieved by cursor.fetchmany", 652 ) 653 654 rows = cur.fetchmany() # Should return an empty list 655 self.assertEqual( 656 len(rows), 657 0, 658 "cursor.fetchmany should return an empty sequence if " 659 "called after the whole result set has been fetched", 660 ) 661 self.assertTrue(cur.rowcount in (-1, 6)) 662 663 self.executeDDL2(cur) 664 cur.execute("select name from %sbarflys" % self.table_prefix) 665 r = cur.fetchmany() # Should get empty sequence 666 self.assertEqual( 667 len(r), 668 0, 669 "cursor.fetchmany should return an empty sequence if " 670 "query retrieved no rows", 671 ) 672 self.assertTrue(cur.rowcount in (-1, 0)) 673 674 finally: 675 con.close() 676 677 def test_fetchall(self): 678 con = self._connect() 679 try: 680 cur = con.cursor() 681 # cursor.fetchall should raise an Error if called 682 # without executing a query that may return rows (such 683 # as a select) 684 self.assertRaises(self.driver.Error, cur.fetchall) 685 686 self.executeDDL1(cur) 687 for sql in self._populate(): 688 cur.execute(sql) 689 690 # cursor.fetchall should raise an Error if called 691 # after executing a a statement that cannot return rows 692 self.assertRaises(self.driver.Error, cur.fetchall) 693 694 cur.execute("select name from %sbooze" % self.table_prefix) 695 rows = cur.fetchall() 696 self.assertTrue(cur.rowcount in (-1, len(self.samples))) 697 self.assertEqual( 698 len(rows), 699 len(self.samples), 700 "cursor.fetchall did not retrieve all rows", 701 ) 702 rows = [r[0] for r in rows] 703 rows.sort() 704 for i in range(0, len(self.samples)): 705 self.assertEqual( 706 rows[i], self.samples[i], "cursor.fetchall retrieved incorrect rows" 707 ) 708 rows = cur.fetchall() 709 self.assertEqual( 710 len(rows), 711 0, 712 "cursor.fetchall should return an empty list if called " 713 "after the whole result set has been fetched", 714 ) 715 self.assertTrue(cur.rowcount in (-1, len(self.samples))) 716 717 self.executeDDL2(cur) 718 cur.execute("select name from %sbarflys" % self.table_prefix) 719 rows = cur.fetchall() 720 self.assertTrue(cur.rowcount in (-1, 0)) 721 self.assertEqual( 722 len(rows), 723 0, 724 "cursor.fetchall should return an empty list if " 725 "a select query returns no rows", 726 ) 727 728 finally: 729 con.close() 730 731 def test_mixedfetch(self): 732 con = self._connect() 733 try: 734 cur = con.cursor() 735 self.executeDDL1(cur) 736 for sql in self._populate(): 737 cur.execute(sql) 738 739 cur.execute("select name from %sbooze" % self.table_prefix) 740 rows1 = cur.fetchone() 741 rows23 = cur.fetchmany(2) 742 rows4 = cur.fetchone() 743 rows56 = cur.fetchall() 744 self.assertTrue(cur.rowcount in (-1, 6)) 745 self.assertEqual( 746 len(rows23), 2, "fetchmany returned incorrect number of rows" 747 ) 748 self.assertEqual( 749 len(rows56), 2, "fetchall returned incorrect number of rows" 750 ) 751 752 rows = [rows1[0]] 753 rows.extend([rows23[0][0], rows23[1][0]]) 754 rows.append(rows4[0]) 755 rows.extend([rows56[0][0], rows56[1][0]]) 756 rows.sort() 757 for i in range(0, len(self.samples)): 758 self.assertEqual( 759 rows[i], self.samples[i], "incorrect data retrieved or inserted" 760 ) 761 finally: 762 con.close() 763 764 def help_nextset_setUp(self, cur): 765 """Should create a procedure called deleteme 766 that returns two result sets, first the 767 number of rows in booze then "name from booze" 768 """ 769 raise NotImplementedError("Helper not implemented") 770 # sql=""" 771 # create procedure deleteme as 772 # begin 773 # select count(*) from booze 774 # select name from booze 775 # end 776 # """ 777 # cur.execute(sql) 778 779 def help_nextset_tearDown(self, cur): 780 "If cleaning up is needed after nextSetTest" 781 raise NotImplementedError("Helper not implemented") 782 # cur.execute("drop procedure deleteme") 783 784 def test_nextset(self): 785 raise NotImplementedError("Drivers need to override this test") 786 787 def test_arraysize(self): 788 # Not much here - rest of the tests for this are in test_fetchmany 789 con = self._connect() 790 try: 791 cur = con.cursor() 792 self.assertTrue( 793 hasattr(cur, "arraysize"), "cursor.arraysize must be defined" 794 ) 795 finally: 796 con.close() 797 798 def test_setinputsizes(self): 799 con = self._connect() 800 try: 801 cur = con.cursor() 802 cur.setinputsizes((25,)) 803 self._paraminsert(cur) # Make sure cursor still works 804 finally: 805 con.close() 806 807 def test_setoutputsize_basic(self): 808 # Basic test is to make sure setoutputsize doesn't blow up 809 con = self._connect() 810 try: 811 cur = con.cursor() 812 cur.setoutputsize(1000) 813 cur.setoutputsize(2000, 0) 814 self._paraminsert(cur) # Make sure the cursor still works 815 finally: 816 con.close() 817 818 def test_setoutputsize(self): 819 # Real test for setoutputsize is driver dependant 820 raise NotImplementedError("Driver needed to override this test") 821 822 def test_None(self): 823 con = self._connect() 824 try: 825 cur = con.cursor() 826 self.executeDDL1(cur) 827 cur.execute("insert into %sbooze values (NULL)" % self.table_prefix) 828 cur.execute("select name from %sbooze" % self.table_prefix) 829 r = cur.fetchall() 830 self.assertEqual(len(r), 1) 831 self.assertEqual(len(r[0]), 1) 832 self.assertEqual(r[0][0], None, "NULL value not returned as None") 833 finally: 834 con.close() 835 836 def test_Date(self): 837 d1 = self.driver.Date(2002, 12, 25) 838 d2 = self.driver.DateFromTicks(time.mktime((2002, 12, 25, 0, 0, 0, 0, 0, 0))) 839 # Can we assume this? API doesn't specify, but it seems implied 840 # self.assertEqual(str(d1),str(d2)) 841 842 def test_Time(self): 843 t1 = self.driver.Time(13, 45, 30) 844 t2 = self.driver.TimeFromTicks(time.mktime((2001, 1, 1, 13, 45, 30, 0, 0, 0))) 845 # Can we assume this? API doesn't specify, but it seems implied 846 # self.assertEqual(str(t1),str(t2)) 847 848 def test_Timestamp(self): 849 t1 = self.driver.Timestamp(2002, 12, 25, 13, 45, 30) 850 t2 = self.driver.TimestampFromTicks( 851 time.mktime((2002, 12, 25, 13, 45, 30, 0, 0, 0)) 852 ) 853 # Can we assume this? API doesn't specify, but it seems implied 854 # self.assertEqual(str(t1),str(t2)) 855 856 def test_Binary(self): 857 b = self.driver.Binary(b"Something") 858 b = self.driver.Binary(b"") 859 860 def test_STRING(self): 861 self.assertTrue(hasattr(self.driver, "STRING"), "module.STRING must be defined") 862 863 def test_BINARY(self): 864 self.assertTrue( 865 hasattr(self.driver, "BINARY"), "module.BINARY must be defined." 866 ) 867 868 def test_NUMBER(self): 869 self.assertTrue( 870 hasattr(self.driver, "NUMBER"), "module.NUMBER must be defined." 871 ) 872 873 def test_DATETIME(self): 874 self.assertTrue( 875 hasattr(self.driver, "DATETIME"), "module.DATETIME must be defined." 876 ) 877 878 def test_ROWID(self): 879 self.assertTrue(hasattr(self.driver, "ROWID"), "module.ROWID must be defined.") 880