1 """Unit tests version 2.6.1.0 for adodbapi""" 2 3 """ 4 adodbapi - A python DB API 2.0 interface to Microsoft ADO 5 6 Copyright (C) 2002 Henrik Ekelund 7 8 This library is free software; you can redistribute it and/or 9 modify it under the terms of the GNU Lesser General Public 10 License as published by the Free Software Foundation; either 11 version 2.1 of the License, or (at your option) any later version. 12 13 This library is distributed in the hope that it will be useful, 14 but WITHOUT ANY WARRANTY; without even the implied warranty of 15 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU 16 Lesser General Public License for more details. 17 18 You should have received a copy of the GNU Lesser General Public 19 License along with this library; if not, write to the Free Software 20 Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA 21 22 Updates by Vernon Cole 23 """ 24 25 import copy 26 import datetime 27 import decimal 28 import random 29 import string 30 import time 31 import unittest 32 33 import adodbapitestconfig as config # run the configuration module. # will set sys.path to find correct version of adodbapi 34 import tryconnection # in our code below, all our switches are from config.whatever 35 36 import adodbapi 37 import adodbapi.apibase as api 38 39 40 def randomstring(length): 41 return "".join([random.choice(string.ascii_letters) for n in range(32)]) 42 43 44 class CommonDBTests(unittest.TestCase): 45 "Self contained super-simple tests in easy syntax, should work on everything between mySQL and Oracle" 46 47 def setUp(self): 48 self.engine = "unknown" 49 50 def getEngine(self): 51 return self.engine 52 53 def getConnection(self): 54 raise NotImplementedError # "This method must be overriden by a subclass" 55 56 def getCursor(self): 57 return self.getConnection().cursor() 58 59 def testConnection(self): 60 crsr = self.getCursor() 61 assert crsr.__class__.__name__ == "Cursor" 62 63 def testErrorHandlerInherits(self): 64 conn = self.getConnection() 65 mycallable = lambda connection, cursor, errorclass, errorvalue: 1 66 conn.errorhandler = mycallable 67 crsr = conn.cursor() 68 assert ( 69 crsr.errorhandler == mycallable 70 ), "Error handler on crsr should be same as on connection" 71 72 def testDefaultErrorHandlerConnection(self): 73 conn = self.getConnection() 74 del conn.messages[:] 75 try: 76 conn.close() 77 conn.commit() # Should not be able to use connection after it is closed 78 except: 79 assert len(conn.messages) == 1 80 assert len(conn.messages[0]) == 2 81 assert conn.messages[0][0] == api.ProgrammingError 82 83 def testOwnErrorHandlerConnection(self): 84 mycallable = ( 85 lambda connection, cursor, errorclass, errorvalue: 1 86 ) # does not raise anything 87 conn = self.getConnection() 88 conn.errorhandler = mycallable 89 conn.close() 90 conn.commit() # Should not be able to use connection after it is closed 91 assert len(conn.messages) == 0 92 93 conn.errorhandler = None # This should bring back the standard error handler 94 try: 95 conn.close() 96 conn.commit() # Should not be able to use connection after it is closed 97 except: 98 pass 99 # The Standard errorhandler appends error to messages attribute 100 assert ( 101 len(conn.messages) > 0 102 ), "Setting errorhandler to none should bring back the standard error handler" 103 104 def testDefaultErrorHandlerCursor(self): 105 crsr = self.getConnection().cursor() 106 del crsr.messages[:] 107 try: 108 crsr.execute("SELECT abbtytddrf FROM dasdasd") 109 except: 110 assert len(crsr.messages) == 1 111 assert len(crsr.messages[0]) == 2 112 assert crsr.messages[0][0] == api.DatabaseError 113 114 def testOwnErrorHandlerCursor(self): 115 mycallable = ( 116 lambda connection, cursor, errorclass, errorvalue: 1 117 ) # does not raise anything 118 crsr = self.getConnection().cursor() 119 crsr.errorhandler = mycallable 120 crsr.execute("SELECT abbtytddrf FROM dasdasd") 121 assert len(crsr.messages) == 0 122 123 crsr.errorhandler = None # This should bring back the standard error handler 124 try: 125 crsr.execute("SELECT abbtytddrf FROM dasdasd") 126 except: 127 pass 128 # The Standard errorhandler appends error to messages attribute 129 assert ( 130 len(crsr.messages) > 0 131 ), "Setting errorhandler to none should bring back the standard error handler" 132 133 def testUserDefinedConversions(self): 134 try: 135 duplicatingConverter = lambda aStringField: aStringField * 2 136 assert duplicatingConverter("gabba") == "gabbagabba" 137 138 self.helpForceDropOnTblTemp() 139 conn = self.getConnection() 140 # the variantConversions attribute should not exist on a normal connection object 141 self.assertRaises(AttributeError, lambda x: conn.variantConversions[x], [2]) 142 # create a variantConversions attribute on the connection 143 conn.variantConversions = copy.copy(api.variantConversions) 144 crsr = conn.cursor() 145 tabdef = ( 146 "CREATE TABLE xx_%s (fldData VARCHAR(100) NOT NULL, fld2 VARCHAR(20))" 147 % config.tmp 148 ) 149 crsr.execute(tabdef) 150 crsr.execute( 151 "INSERT INTO xx_%s(fldData,fld2) VALUES('gabba','booga')" % config.tmp 152 ) 153 crsr.execute( 154 "INSERT INTO xx_%s(fldData,fld2) VALUES('hey','yo')" % config.tmp 155 ) 156 # change converter for ALL adoStringTypes columns 157 conn.variantConversions[api.adoStringTypes] = duplicatingConverter 158 crsr.execute("SELECT fldData,fld2 FROM xx_%s ORDER BY fldData" % config.tmp) 159 160 rows = crsr.fetchall() 161 row = rows[0] 162 self.assertEqual(row[0], "gabbagabba") 163 row = rows[1] 164 self.assertEqual(row[0], "heyhey") 165 self.assertEqual(row[1], "yoyo") 166 167 upcaseConverter = lambda aStringField: aStringField.upper() 168 assert upcaseConverter("upThis") == "UPTHIS" 169 170 # now use a single column converter 171 rows.converters[1] = upcaseConverter # convert second column 172 self.assertEqual(row[0], "heyhey") # first will be unchanged 173 self.assertEqual(row[1], "YO") # second will convert to upper case 174 175 finally: 176 try: 177 del conn.variantConversions # Restore the default 178 except: 179 pass 180 self.helpRollbackTblTemp() 181 182 def helpTestDataType( 183 self, 184 sqlDataTypeString, 185 DBAPIDataTypeString, 186 pyData, 187 pyDataInputAlternatives=None, 188 compareAlmostEqual=None, 189 allowedReturnValues=None, 190 ): 191 self.helpForceDropOnTblTemp() 192 conn = self.getConnection() 193 crsr = conn.cursor() 194 tabdef = ( 195 """ 196 CREATE TABLE xx_%s ( 197 fldId integer NOT NULL, 198 fldData """ 199 % config.tmp 200 + sqlDataTypeString 201 + ")\n" 202 ) 203 204 crsr.execute(tabdef) 205 206 # Test Null values mapped to None 207 crsr.execute("INSERT INTO xx_%s (fldId) VALUES (1)" % config.tmp) 208 209 crsr.execute("SELECT fldId,fldData FROM xx_%s" % config.tmp) 210 rs = crsr.fetchone() 211 self.assertEqual(rs[1], None) # Null should be mapped to None 212 assert rs[0] == 1 213 214 # Test description related 215 descTuple = crsr.description[1] 216 assert descTuple[0] in ["fldData", "flddata"], 'was "%s" expected "%s"' % ( 217 descTuple[0], 218 "fldData", 219 ) 220 221 if DBAPIDataTypeString == "STRING": 222 assert descTuple[1] == api.STRING, 'was "%s" expected "%s"' % ( 223 descTuple[1], 224 api.STRING.values, 225 ) 226 elif DBAPIDataTypeString == "NUMBER": 227 assert descTuple[1] == api.NUMBER, 'was "%s" expected "%s"' % ( 228 descTuple[1], 229 api.NUMBER.values, 230 ) 231 elif DBAPIDataTypeString == "BINARY": 232 assert descTuple[1] == api.BINARY, 'was "%s" expected "%s"' % ( 233 descTuple[1], 234 api.BINARY.values, 235 ) 236 elif DBAPIDataTypeString == "DATETIME": 237 assert descTuple[1] == api.DATETIME, 'was "%s" expected "%s"' % ( 238 descTuple[1], 239 api.DATETIME.values, 240 ) 241 elif DBAPIDataTypeString == "ROWID": 242 assert descTuple[1] == api.ROWID, 'was "%s" expected "%s"' % ( 243 descTuple[1], 244 api.ROWID.values, 245 ) 246 elif DBAPIDataTypeString == "UUID": 247 assert descTuple[1] == api.OTHER, 'was "%s" expected "%s"' % ( 248 descTuple[1], 249 api.OTHER.values, 250 ) 251 else: 252 raise NotImplementedError # "DBAPIDataTypeString not provided" 253 254 # Test data binding 255 inputs = [pyData] 256 if pyDataInputAlternatives: 257 inputs.extend(pyDataInputAlternatives) 258 inputs = set(inputs) # removes redundant string==unicode tests 259 fldId = 1 260 for inParam in inputs: 261 fldId += 1 262 try: 263 crsr.execute( 264 "INSERT INTO xx_%s (fldId,fldData) VALUES (?,?)" % config.tmp, 265 (fldId, inParam), 266 ) 267 except: 268 conn.printADOerrors() 269 raise 270 crsr.execute( 271 "SELECT fldData FROM xx_%s WHERE ?=fldID" % config.tmp, [fldId] 272 ) 273 rs = crsr.fetchone() 274 if allowedReturnValues: 275 allowedTypes = tuple([type(aRV) for aRV in allowedReturnValues]) 276 assert isinstance(rs[0], allowedTypes), ( 277 'result type "%s" must be one of %s' % (type(rs[0]), allowedTypes) 278 ) 279 else: 280 assert isinstance(rs[0], type(pyData)), ( 281 'result type "%s" must be instance of %s' 282 % ( 283 type(rs[0]), 284 type(pyData), 285 ) 286 ) 287 288 if compareAlmostEqual and DBAPIDataTypeString == "DATETIME": 289 iso1 = adodbapi.dateconverter.DateObjectToIsoFormatString(rs[0]) 290 iso2 = adodbapi.dateconverter.DateObjectToIsoFormatString(pyData) 291 self.assertEqual(iso1, iso2) 292 elif compareAlmostEqual: 293 s = float(pyData) 294 v = float(rs[0]) 295 assert abs(v - s) / s < 0.00001, ( 296 "Values not almost equal recvd=%s, expected=%f" % (rs[0], s) 297 ) 298 else: 299 if allowedReturnValues: 300 ok = False 301 self.assertTrue( 302 rs[0] in allowedReturnValues, 303 f'Value "{rs[0]!r}" not in {allowedReturnValues}', 304 ) 305 else: 306 self.assertEqual( 307 rs[0], 308 pyData, 309 'Values are not equal recvd="%s", expected="%s"' 310 % (rs[0], pyData), 311 ) 312 313 def testDataTypeFloat(self): 314 self.helpTestDataType("real", "NUMBER", 3.45, compareAlmostEqual=True) 315 self.helpTestDataType("float", "NUMBER", 1.79e37, compareAlmostEqual=True) 316 317 def testDataTypeDecmal(self): 318 self.helpTestDataType( 319 "decimal(18,2)", 320 "NUMBER", 321 3.45, 322 allowedReturnValues=["3.45", "3,45", decimal.Decimal("3.45")], 323 ) 324 self.helpTestDataType( 325 "numeric(18,2)", 326 "NUMBER", 327 3.45, 328 allowedReturnValues=["3.45", "3,45", decimal.Decimal("3.45")], 329 ) 330 self.helpTestDataType( 331 "decimal(20,2)", 332 "NUMBER", 333 444444444444444444, 334 allowedReturnValues=[ 335 "444444444444444444.00", 336 "444444444444444444,00", 337 decimal.Decimal("444444444444444444"), 338 ], 339 ) 340 if self.getEngine() == "MSSQL": 341 self.helpTestDataType( 342 "uniqueidentifier", 343 "UUID", 344 "{71A4F49E-39F3-42B1-A41E-48FF154996E6}", 345 allowedReturnValues=["{71A4F49E-39F3-42B1-A41E-48FF154996E6}"], 346 ) 347 348 def testDataTypeMoney(self): # v2.1 Cole -- use decimal for money 349 if self.getEngine() == "MySQL": 350 self.helpTestDataType( 351 "DECIMAL(20,4)", "NUMBER", decimal.Decimal("-922337203685477.5808") 352 ) 353 elif self.getEngine() == "PostgreSQL": 354 self.helpTestDataType( 355 "money", 356 "NUMBER", 357 decimal.Decimal("-922337203685477.5808"), 358 compareAlmostEqual=True, 359 allowedReturnValues=[ 360 -922337203685477.5808, 361 decimal.Decimal("-922337203685477.5808"), 362 ], 363 ) 364 else: 365 self.helpTestDataType("smallmoney", "NUMBER", decimal.Decimal("214748.02")) 366 self.helpTestDataType( 367 "money", "NUMBER", decimal.Decimal("-922337203685477.5808") 368 ) 369 370 def testDataTypeInt(self): 371 if self.getEngine() != "PostgreSQL": 372 self.helpTestDataType("tinyint", "NUMBER", 115) 373 self.helpTestDataType("smallint", "NUMBER", -32768) 374 if self.getEngine() not in ["ACCESS", "PostgreSQL"]: 375 self.helpTestDataType( 376 "bit", "NUMBER", 1 377 ) # Does not work correctly with access 378 if self.getEngine() in ["MSSQL", "PostgreSQL"]: 379 self.helpTestDataType( 380 "bigint", 381 "NUMBER", 382 3000000000, 383 allowedReturnValues=[3000000000, 3000000000], 384 ) 385 self.helpTestDataType("int", "NUMBER", 2147483647) 386 387 def testDataTypeChar(self): 388 for sqlDataType in ("char(6)", "nchar(6)"): 389 self.helpTestDataType( 390 sqlDataType, 391 "STRING", 392 "spam ", 393 allowedReturnValues=["spam", "spam", "spam ", "spam "], 394 ) 395 396 def testDataTypeVarChar(self): 397 if self.getEngine() == "MySQL": 398 stringKinds = ["varchar(10)", "text"] 399 elif self.getEngine() == "PostgreSQL": 400 stringKinds = ["varchar(10)", "text", "character varying"] 401 else: 402 stringKinds = [ 403 "varchar(10)", 404 "nvarchar(10)", 405 "text", 406 "ntext", 407 ] # ,"varchar(max)"] 408 409 for sqlDataType in stringKinds: 410 self.helpTestDataType(sqlDataType, "STRING", "spam", ["spam"]) 411 412 def testDataTypeDate(self): 413 if self.getEngine() == "PostgreSQL": 414 dt = "timestamp" 415 else: 416 dt = "datetime" 417 self.helpTestDataType( 418 dt, "DATETIME", adodbapi.Date(2002, 10, 28), compareAlmostEqual=True 419 ) 420 if self.getEngine() not in ["MySQL", "PostgreSQL"]: 421 self.helpTestDataType( 422 "smalldatetime", 423 "DATETIME", 424 adodbapi.Date(2002, 10, 28), 425 compareAlmostEqual=True, 426 ) 427 if tag != "pythontime" and self.getEngine() not in [ 428 "MySQL", 429 "PostgreSQL", 430 ]: # fails when using pythonTime 431 self.helpTestDataType( 432 dt, 433 "DATETIME", 434 adodbapi.Timestamp(2002, 10, 28, 12, 15, 1), 435 compareAlmostEqual=True, 436 ) 437 438 def testDataTypeBinary(self): 439 binfld = b"\x07\x00\xe2\x40*" 440 arv = [binfld, adodbapi.Binary(binfld), bytes(binfld)] 441 if self.getEngine() == "PostgreSQL": 442 self.helpTestDataType( 443 "bytea", "BINARY", adodbapi.Binary(binfld), allowedReturnValues=arv 444 ) 445 else: 446 self.helpTestDataType( 447 "binary(5)", "BINARY", adodbapi.Binary(binfld), allowedReturnValues=arv 448 ) 449 self.helpTestDataType( 450 "varbinary(100)", 451 "BINARY", 452 adodbapi.Binary(binfld), 453 allowedReturnValues=arv, 454 ) 455 if self.getEngine() != "MySQL": 456 self.helpTestDataType( 457 "image", "BINARY", adodbapi.Binary(binfld), allowedReturnValues=arv 458 ) 459 460 def helpRollbackTblTemp(self): 461 self.helpForceDropOnTblTemp() 462 463 def helpForceDropOnTblTemp(self): 464 conn = self.getConnection() 465 with conn.cursor() as crsr: 466 try: 467 crsr.execute("DROP TABLE xx_%s" % config.tmp) 468 if not conn.autocommit: 469 conn.commit() 470 except: 471 pass 472 473 def helpCreateAndPopulateTableTemp(self, crsr): 474 tabdef = ( 475 """ 476 CREATE TABLE xx_%s ( 477 fldData INTEGER 478 ) 479 """ 480 % config.tmp 481 ) 482 try: # EAFP 483 crsr.execute(tabdef) 484 except api.DatabaseError: # was not dropped before 485 self.helpForceDropOnTblTemp() # so drop it now 486 crsr.execute(tabdef) 487 for i in range(9): # note: this poor SQL code, but a valid test 488 crsr.execute("INSERT INTO xx_%s (fldData) VALUES (%i)" % (config.tmp, i)) 489 # NOTE: building the test table without using parameter substitution 490 491 def testFetchAll(self): 492 crsr = self.getCursor() 493 self.helpCreateAndPopulateTableTemp(crsr) 494 crsr.execute("SELECT fldData FROM xx_%s" % config.tmp) 495 rs = crsr.fetchall() 496 assert len(rs) == 9 497 # test slice of rows 498 i = 3 499 for row in rs[3:-2]: # should have rowid 3..6 500 assert row[0] == i 501 i += 1 502 self.helpRollbackTblTemp() 503 504 def testPreparedStatement(self): 505 crsr = self.getCursor() 506 self.helpCreateAndPopulateTableTemp(crsr) 507 crsr.prepare("SELECT fldData FROM xx_%s" % config.tmp) 508 crsr.execute(crsr.command) # remember the one that was prepared 509 rs = crsr.fetchall() 510 assert len(rs) == 9 511 assert rs[2][0] == 2 512 self.helpRollbackTblTemp() 513 514 def testWrongPreparedStatement(self): 515 crsr = self.getCursor() 516 self.helpCreateAndPopulateTableTemp(crsr) 517 crsr.prepare("SELECT * FROM nowhere") 518 crsr.execute( 519 "SELECT fldData FROM xx_%s" % config.tmp 520 ) # should execute this one, not the prepared one 521 rs = crsr.fetchall() 522 assert len(rs) == 9 523 assert rs[2][0] == 2 524 self.helpRollbackTblTemp() 525 526 def testIterator(self): 527 crsr = self.getCursor() 528 self.helpCreateAndPopulateTableTemp(crsr) 529 crsr.execute("SELECT fldData FROM xx_%s" % config.tmp) 530 for i, row in enumerate( 531 crsr 532 ): # using cursor as an iterator, rather than fetchxxx 533 assert row[0] == i 534 self.helpRollbackTblTemp() 535 536 def testExecuteMany(self): 537 crsr = self.getCursor() 538 self.helpCreateAndPopulateTableTemp(crsr) 539 seq_of_values = [(111,), (222,)] 540 crsr.executemany( 541 "INSERT INTO xx_%s (fldData) VALUES (?)" % config.tmp, seq_of_values 542 ) 543 if crsr.rowcount == -1: 544 print( 545 self.getEngine() 546 + " Provider does not support rowcount (on .executemany())" 547 ) 548 else: 549 self.assertEqual(crsr.rowcount, 2) 550 crsr.execute("SELECT fldData FROM xx_%s" % config.tmp) 551 rs = crsr.fetchall() 552 assert len(rs) == 11 553 self.helpRollbackTblTemp() 554 555 def testRowCount(self): 556 crsr = self.getCursor() 557 self.helpCreateAndPopulateTableTemp(crsr) 558 crsr.execute("SELECT fldData FROM xx_%s" % config.tmp) 559 if crsr.rowcount == -1: 560 # print("provider does not support rowcount on select") 561 pass 562 else: 563 self.assertEqual(crsr.rowcount, 9) 564 self.helpRollbackTblTemp() 565 566 def testRowCountNoRecordset(self): 567 crsr = self.getCursor() 568 self.helpCreateAndPopulateTableTemp(crsr) 569 crsr.execute("DELETE FROM xx_%s WHERE fldData >= 5" % config.tmp) 570 if crsr.rowcount == -1: 571 print(self.getEngine() + " Provider does not support rowcount (on DELETE)") 572 else: 573 self.assertEqual(crsr.rowcount, 4) 574 self.helpRollbackTblTemp() 575 576 def testFetchMany(self): 577 crsr = self.getCursor() 578 self.helpCreateAndPopulateTableTemp(crsr) 579 crsr.execute("SELECT fldData FROM xx_%s" % config.tmp) 580 rs = crsr.fetchmany(3) 581 assert len(rs) == 3 582 rs = crsr.fetchmany(5) 583 assert len(rs) == 5 584 rs = crsr.fetchmany(5) 585 assert len(rs) == 1 # Asked for five, but there is only one left 586 self.helpRollbackTblTemp() 587 588 def testFetchManyWithArraySize(self): 589 crsr = self.getCursor() 590 self.helpCreateAndPopulateTableTemp(crsr) 591 crsr.execute("SELECT fldData FROM xx_%s" % config.tmp) 592 rs = crsr.fetchmany() 593 assert len(rs) == 1 # arraysize Defaults to one 594 crsr.arraysize = 4 595 rs = crsr.fetchmany() 596 assert len(rs) == 4 597 rs = crsr.fetchmany() 598 assert len(rs) == 4 599 rs = crsr.fetchmany() 600 assert len(rs) == 0 601 self.helpRollbackTblTemp() 602 603 def testErrorConnect(self): 604 conn = self.getConnection() 605 conn.close() 606 self.assertRaises(api.DatabaseError, self.db, "not a valid connect string", {}) 607 608 def testRowIterator(self): 609 self.helpForceDropOnTblTemp() 610 conn = self.getConnection() 611 crsr = conn.cursor() 612 tabdef = ( 613 """ 614 CREATE TABLE xx_%s ( 615 fldId integer NOT NULL, 616 fldTwo integer, 617 fldThree integer, 618 fldFour integer) 619 """ 620 % config.tmp 621 ) 622 crsr.execute(tabdef) 623 624 inputs = [(2, 3, 4), (102, 103, 104)] 625 fldId = 1 626 for inParam in inputs: 627 fldId += 1 628 try: 629 crsr.execute( 630 "INSERT INTO xx_%s (fldId,fldTwo,fldThree,fldFour) VALUES (?,?,?,?)" 631 % config.tmp, 632 (fldId, inParam[0], inParam[1], inParam[2]), 633 ) 634 except: 635 conn.printADOerrors() 636 raise 637 crsr.execute( 638 "SELECT fldTwo,fldThree,fldFour FROM xx_%s WHERE ?=fldID" % config.tmp, 639 [fldId], 640 ) 641 rec = crsr.fetchone() 642 # check that stepping through an emulated row works 643 for j in range(len(inParam)): 644 assert rec[j] == inParam[j], ( 645 'returned value:"%s" != test value:"%s"' % (rec[j], inParam[j]) 646 ) 647 # check that we can get a complete tuple from a row 648 assert ( 649 tuple(rec) == inParam 650 ), f'returned value:"{rec!r}" != test value:"{inParam!r}"' 651 # test that slices of rows work 652 slice1 = tuple(rec[:-1]) 653 slice2 = tuple(inParam[0:2]) 654 assert ( 655 slice1 == slice2 656 ), f'returned value:"{slice1!r}" != test value:"{slice2!r}"' 657 # now test named column retrieval 658 assert rec["fldTwo"] == inParam[0] 659 assert rec.fldThree == inParam[1] 660 assert rec.fldFour == inParam[2] 661 # test array operation 662 # note that the fields vv vv vv are out of order 663 crsr.execute("select fldThree,fldFour,fldTwo from xx_%s" % config.tmp) 664 recs = crsr.fetchall() 665 assert recs[1][0] == 103 666 assert recs[0][1] == 4 667 assert recs[1]["fldFour"] == 104 668 assert recs[0, 0] == 3 669 assert recs[0, "fldTwo"] == 2 670 assert recs[1, 2] == 102 671 for i in range(1): 672 for j in range(2): 673 assert recs[i][j] == recs[i, j] 674 675 def testFormatParamstyle(self): 676 self.helpForceDropOnTblTemp() 677 conn = self.getConnection() 678 conn.paramstyle = "format" # test nonstandard use of paramstyle 679 crsr = conn.cursor() 680 tabdef = ( 681 """ 682 CREATE TABLE xx_%s ( 683 fldId integer NOT NULL, 684 fldData varchar(10), 685 fldConst varchar(30)) 686 """ 687 % config.tmp 688 ) 689 crsr.execute(tabdef) 690 691 inputs = ["one", "two", "three"] 692 fldId = 2 693 for inParam in inputs: 694 fldId += 1 695 sql = ( 696 "INSERT INTO xx_" 697 + config.tmp 698 + " (fldId,fldConst,fldData) VALUES (%s,'thi%s :may cause? trouble', %s)" 699 ) 700 try: 701 crsr.execute(sql, (fldId, inParam)) 702 except: 703 conn.printADOerrors() 704 raise 705 crsr.execute( 706 "SELECT fldData, fldConst FROM xx_" + config.tmp + " WHERE %s=fldID", 707 [fldId], 708 ) 709 rec = crsr.fetchone() 710 self.assertEqual( 711 rec[0], 712 inParam, 713 'returned value:"%s" != test value:"%s"' % (rec[0], inParam), 714 ) 715 self.assertEqual(rec[1], "thi%s :may cause? trouble") 716 717 # now try an operation with a "%s" as part of a literal 718 sel = ( 719 "insert into xx_" + config.tmp + " (fldId,fldData) VALUES (%s,'four%sfive')" 720 ) 721 params = (20,) 722 crsr.execute(sel, params) 723 724 # test the .query implementation 725 assert "(?," in crsr.query, 'expected:"%s" in "%s"' % ("(?,", crsr.query) 726 # test the .command attribute 727 assert crsr.command == sel, 'expected:"%s" but found "%s"' % (sel, crsr.command) 728 729 # test the .parameters attribute 730 self.assertEqual(crsr.parameters, params) 731 # now make sure the data made it 732 crsr.execute("SELECT fldData FROM xx_%s WHERE fldID=20" % config.tmp) 733 rec = crsr.fetchone() 734 self.assertEqual(rec[0], "four%sfive") 735 736 def testNamedParamstyle(self): 737 self.helpForceDropOnTblTemp() 738 conn = self.getConnection() 739 crsr = conn.cursor() 740 crsr.paramstyle = "named" # test nonstandard use of paramstyle 741 tabdef = ( 742 """ 743 CREATE TABLE xx_%s ( 744 fldId integer NOT NULL, 745 fldData varchar(10)) 746 """ 747 % config.tmp 748 ) 749 crsr.execute(tabdef) 750 751 inputs = ["four", "five", "six"] 752 fldId = 10 753 for inParam in inputs: 754 fldId += 1 755 try: 756 crsr.execute( 757 "INSERT INTO xx_%s (fldId,fldData) VALUES (:Id,:f_Val)" 758 % config.tmp, 759 {"f_Val": inParam, "Id": fldId}, 760 ) 761 except: 762 conn.printADOerrors() 763 raise 764 crsr.execute( 765 "SELECT fldData FROM xx_%s WHERE fldID=:Id" % config.tmp, {"Id": fldId} 766 ) 767 rec = crsr.fetchone() 768 self.assertEqual( 769 rec[0], 770 inParam, 771 'returned value:"%s" != test value:"%s"' % (rec[0], inParam), 772 ) 773 # now a test with a ":" as part of a literal 774 crsr.execute( 775 "insert into xx_%s (fldId,fldData) VALUES (:xyz,'six:five')" % config.tmp, 776 {"xyz": 30}, 777 ) 778 crsr.execute("SELECT fldData FROM xx_%s WHERE fldID=30" % config.tmp) 779 rec = crsr.fetchone() 780 self.assertEqual(rec[0], "six:five") 781 782 def testPyformatParamstyle(self): 783 self.helpForceDropOnTblTemp() 784 conn = self.getConnection() 785 crsr = conn.cursor() 786 crsr.paramstyle = "pyformat" # test nonstandard use of paramstyle 787 tabdef = ( 788 """ 789 CREATE TABLE xx_%s ( 790 fldId integer NOT NULL, 791 fldData varchar(10)) 792 """ 793 % config.tmp 794 ) 795 crsr.execute(tabdef) 796 797 inputs = ["four", "five", "six"] 798 fldId = 10 799 for inParam in inputs: 800 fldId += 1 801 try: 802 crsr.execute( 803 "INSERT INTO xx_%s (fldId,fldData) VALUES (%%(Id)s,%%(f_Val)s)" 804 % config.tmp, 805 {"f_Val": inParam, "Id": fldId}, 806 ) 807 except: 808 conn.printADOerrors() 809 raise 810 crsr.execute( 811 "SELECT fldData FROM xx_%s WHERE fldID=%%(Id)s" % config.tmp, 812 {"Id": fldId}, 813 ) 814 rec = crsr.fetchone() 815 self.assertEqual( 816 rec[0], 817 inParam, 818 'returned value:"%s" != test value:"%s"' % (rec[0], inParam), 819 ) 820 # now a test with a "%" as part of a literal 821 crsr.execute( 822 "insert into xx_%s (fldId,fldData) VALUES (%%(xyz)s,'six%%five')" 823 % config.tmp, 824 {"xyz": 30}, 825 ) 826 crsr.execute("SELECT fldData FROM xx_%s WHERE fldID=30" % config.tmp) 827 rec = crsr.fetchone() 828 self.assertEqual(rec[0], "six%five") 829 830 def testAutomaticParamstyle(self): 831 self.helpForceDropOnTblTemp() 832 conn = self.getConnection() 833 conn.paramstyle = "dynamic" # test nonstandard use of paramstyle 834 crsr = conn.cursor() 835 tabdef = ( 836 """ 837 CREATE TABLE xx_%s ( 838 fldId integer NOT NULL, 839 fldData varchar(10), 840 fldConst varchar(30)) 841 """ 842 % config.tmp 843 ) 844 crsr.execute(tabdef) 845 inputs = ["one", "two", "three"] 846 fldId = 2 847 for inParam in inputs: 848 fldId += 1 849 try: 850 crsr.execute( 851 "INSERT INTO xx_" 852 + config.tmp 853 + " (fldId,fldConst,fldData) VALUES (?,'thi%s :may cause? troub:1e', ?)", 854 (fldId, inParam), 855 ) 856 except: 857 conn.printADOerrors() 858 raise 859 trouble = "thi%s :may cause? troub:1e" 860 crsr.execute( 861 "SELECT fldData, fldConst FROM xx_" + config.tmp + " WHERE ?=fldID", 862 [fldId], 863 ) 864 rec = crsr.fetchone() 865 self.assertEqual( 866 rec[0], 867 inParam, 868 'returned value:"%s" != test value:"%s"' % (rec[0], inParam), 869 ) 870 self.assertEqual(rec[1], trouble) 871 # inputs = [u'four',u'five',u'six'] 872 fldId = 10 873 for inParam in inputs: 874 fldId += 1 875 try: 876 crsr.execute( 877 "INSERT INTO xx_%s (fldId,fldData) VALUES (:Id,:f_Val)" 878 % config.tmp, 879 {"f_Val": inParam, "Id": fldId}, 880 ) 881 except: 882 conn.printADOerrors() 883 raise 884 crsr.execute( 885 "SELECT fldData FROM xx_%s WHERE :Id=fldID" % config.tmp, {"Id": fldId} 886 ) 887 rec = crsr.fetchone() 888 self.assertEqual( 889 rec[0], 890 inParam, 891 'returned value:"%s" != test value:"%s"' % (rec[0], inParam), 892 ) 893 # now a test with a ":" as part of a literal -- and use a prepared query 894 ppdcmd = ( 895 "insert into xx_%s (fldId,fldData) VALUES (:xyz,'six:five')" % config.tmp 896 ) 897 crsr.prepare(ppdcmd) 898 crsr.execute(ppdcmd, {"xyz": 30}) 899 crsr.execute("SELECT fldData FROM xx_%s WHERE fldID=30" % config.tmp) 900 rec = crsr.fetchone() 901 self.assertEqual(rec[0], "six:five") 902 903 def testRollBack(self): 904 conn = self.getConnection() 905 crsr = conn.cursor() 906 assert not crsr.connection.autocommit, "Unexpected beginning condition" 907 self.helpCreateAndPopulateTableTemp(crsr) 908 crsr.connection.commit() # commit the first bunch 909 910 crsr.execute("INSERT INTO xx_%s (fldData) VALUES(100)" % config.tmp) 911 912 selectSql = "SELECT fldData FROM xx_%s WHERE fldData=100" % config.tmp 913 crsr.execute(selectSql) 914 rs = crsr.fetchall() 915 assert len(rs) == 1 916 self.conn.rollback() 917 crsr.execute(selectSql) 918 assert ( 919 crsr.fetchone() is None 920 ), "cursor.fetchone should return None if a query retrieves no rows" 921 crsr.execute("SELECT fldData from xx_%s" % config.tmp) 922 rs = crsr.fetchall() 923 assert len(rs) == 9, "the original records should still be present" 924 self.helpRollbackTblTemp() 925 926 def testCommit(self): 927 try: 928 con2 = self.getAnotherConnection() 929 except NotImplementedError: 930 return # should be "SKIP" for ACCESS 931 assert not con2.autocommit, "default should be manual commit" 932 crsr = con2.cursor() 933 self.helpCreateAndPopulateTableTemp(crsr) 934 935 crsr.execute("INSERT INTO xx_%s (fldData) VALUES(100)" % config.tmp) 936 con2.commit() 937 938 selectSql = "SELECT fldData FROM xx_%s WHERE fldData=100" % config.tmp 939 crsr.execute(selectSql) 940 rs = crsr.fetchall() 941 assert len(rs) == 1 942 crsr.close() 943 con2.close() 944 conn = self.getConnection() 945 crsr = self.getCursor() 946 with conn.cursor() as crsr: 947 crsr.execute(selectSql) 948 rs = crsr.fetchall() 949 assert len(rs) == 1 950 assert rs[0][0] == 100 951 self.helpRollbackTblTemp() 952 953 def testAutoRollback(self): 954 try: 955 con2 = self.getAnotherConnection() 956 except NotImplementedError: 957 return # should be "SKIP" for ACCESS 958 assert not con2.autocommit, "unexpected beginning condition" 959 crsr = con2.cursor() 960 self.helpCreateAndPopulateTableTemp(crsr) 961 crsr.execute("INSERT INTO xx_%s (fldData) VALUES(100)" % config.tmp) 962 selectSql = "SELECT fldData FROM xx_%s WHERE fldData=100" % config.tmp 963 crsr.execute(selectSql) 964 rs = crsr.fetchall() 965 assert len(rs) == 1 966 crsr.close() 967 con2.close() 968 crsr = self.getCursor() 969 try: 970 crsr.execute( 971 selectSql 972 ) # closing the connection should have forced rollback 973 row = crsr.fetchone() 974 except api.DatabaseError: 975 row = None # if the entire table disappeared the rollback was perfect and the test passed 976 assert ( 977 row is None 978 ), f"cursor.fetchone should return None if a query retrieves no rows. Got {row!r}" 979 self.helpRollbackTblTemp() 980 981 def testAutoCommit(self): 982 try: 983 ac_conn = self.getAnotherConnection({"autocommit": True}) 984 except NotImplementedError: 985 return # should be "SKIP" for ACCESS 986 crsr = ac_conn.cursor() 987 self.helpCreateAndPopulateTableTemp(crsr) 988 crsr.execute("INSERT INTO xx_%s (fldData) VALUES(100)" % config.tmp) 989 crsr.close() 990 with self.getCursor() as crsr: 991 selectSql = "SELECT fldData from xx_%s" % config.tmp 992 crsr.execute( 993 selectSql 994 ) # closing the connection should _not_ have forced rollback 995 rs = crsr.fetchall() 996 assert len(rs) == 10, "all records should still be present" 997 ac_conn.close() 998 self.helpRollbackTblTemp() 999 1000 def testSwitchedAutoCommit(self): 1001 try: 1002 ac_conn = self.getAnotherConnection() 1003 except NotImplementedError: 1004 return # should be "SKIP" for ACCESS 1005 ac_conn.autocommit = True 1006 crsr = ac_conn.cursor() 1007 self.helpCreateAndPopulateTableTemp(crsr) 1008 crsr.execute("INSERT INTO xx_%s (fldData) VALUES(100)" % config.tmp) 1009 crsr.close() 1010 conn = self.getConnection() 1011 ac_conn.close() 1012 with self.getCursor() as crsr: 1013 selectSql = "SELECT fldData from xx_%s" % config.tmp 1014 crsr.execute( 1015 selectSql 1016 ) # closing the connection should _not_ have forced rollback 1017 rs = crsr.fetchall() 1018 assert len(rs) == 10, "all records should still be present" 1019 self.helpRollbackTblTemp() 1020 1021 def testExtendedTypeHandling(self): 1022 class XtendString(str): 1023 pass 1024 1025 class XtendInt(int): 1026 pass 1027 1028 class XtendFloat(float): 1029 pass 1030 1031 xs = XtendString(randomstring(30)) 1032 xi = XtendInt(random.randint(-100, 500)) 1033 xf = XtendFloat(random.random()) 1034 self.helpForceDropOnTblTemp() 1035 conn = self.getConnection() 1036 crsr = conn.cursor() 1037 tabdef = ( 1038 """ 1039 CREATE TABLE xx_%s ( 1040 s VARCHAR(40) NOT NULL, 1041 i INTEGER NOT NULL, 1042 f REAL NOT NULL)""" 1043 % config.tmp 1044 ) 1045 crsr.execute(tabdef) 1046 crsr.execute( 1047 "INSERT INTO xx_%s (s, i, f) VALUES (?, ?, ?)" % config.tmp, (xs, xi, xf) 1048 ) 1049 crsr.close() 1050 conn = self.getConnection() 1051 with self.getCursor() as crsr: 1052 selectSql = "SELECT s, i, f from xx_%s" % config.tmp 1053 crsr.execute( 1054 selectSql 1055 ) # closing the connection should _not_ have forced rollback 1056 row = crsr.fetchone() 1057 self.assertEqual(row.s, xs) 1058 self.assertEqual(row.i, xi) 1059 self.assertAlmostEqual(row.f, xf) 1060 self.helpRollbackTblTemp() 1061 1062 1063 class TestADOwithSQLServer(CommonDBTests): 1064 def setUp(self): 1065 self.conn = config.dbSqlServerconnect( 1066 *config.connStrSQLServer[0], **config.connStrSQLServer[1] 1067 ) 1068 self.conn.timeout = 30 # turn timeout back up 1069 self.engine = "MSSQL" 1070 self.db = config.dbSqlServerconnect 1071 1072 def tearDown(self): 1073 try: 1074 self.conn.rollback() 1075 except: 1076 pass 1077 try: 1078 self.conn.close() 1079 except: 1080 pass 1081 self.conn = None 1082 1083 def getConnection(self): 1084 return self.conn 1085 1086 def getAnotherConnection(self, addkeys=None): 1087 keys = config.connStrSQLServer[1].copy() 1088 if addkeys: 1089 keys.update(addkeys) 1090 return config.dbSqlServerconnect(*config.connStrSQLServer[0], **keys) 1091 1092 def testVariableReturningStoredProcedure(self): 1093 crsr = self.conn.cursor() 1094 spdef = """ 1095 CREATE PROCEDURE sp_DeleteMeOnlyForTesting 1096 @theInput varchar(50), 1097 @theOtherInput varchar(50), 1098 @theOutput varchar(100) OUTPUT 1099 AS 1100 SET @theOutput=@theInput+@theOtherInput 1101 """ 1102 try: 1103 crsr.execute("DROP PROCEDURE sp_DeleteMeOnlyForTesting") 1104 self.conn.commit() 1105 except: # Make sure it is empty 1106 pass 1107 crsr.execute(spdef) 1108 1109 retvalues = crsr.callproc( 1110 "sp_DeleteMeOnlyForTesting", ("Dodsworth", "Anne", " ") 1111 ) 1112 assert retvalues[0] == "Dodsworth", f'{retvalues[0]!r} is not "Dodsworth"' 1113 assert retvalues[1] == "Anne", f'{retvalues[1]!r} is not "Anne"' 1114 assert ( 1115 retvalues[2] == "DodsworthAnne" 1116 ), f'{retvalues[2]!r} is not "DodsworthAnne"' 1117 self.conn.rollback() 1118 1119 def testMultipleSetReturn(self): 1120 crsr = self.getCursor() 1121 self.helpCreateAndPopulateTableTemp(crsr) 1122 1123 spdef = """ 1124 CREATE PROCEDURE sp_DeleteMe_OnlyForTesting 1125 AS 1126 SELECT fldData FROM xx_%s ORDER BY fldData ASC 1127 SELECT fldData From xx_%s where fldData = -9999 1128 SELECT fldData FROM xx_%s ORDER BY fldData DESC 1129 """ % ( 1130 config.tmp, 1131 config.tmp, 1132 config.tmp, 1133 ) 1134 try: 1135 crsr.execute("DROP PROCEDURE sp_DeleteMe_OnlyForTesting") 1136 self.conn.commit() 1137 except: # Make sure it is empty 1138 pass 1139 crsr.execute(spdef) 1140 1141 retvalues = crsr.callproc("sp_DeleteMe_OnlyForTesting") 1142 row = crsr.fetchone() 1143 self.assertEqual(row[0], 0) 1144 assert crsr.nextset() == True, "Operation should succeed" 1145 assert not crsr.fetchall(), "Should be an empty second set" 1146 assert crsr.nextset() == True, "third set should be present" 1147 rowdesc = crsr.fetchall() 1148 self.assertEqual(rowdesc[0][0], 8) 1149 assert crsr.nextset() is None, "No more return sets, should return None" 1150 1151 self.helpRollbackTblTemp() 1152 1153 def testDatetimeProcedureParameter(self): 1154 crsr = self.conn.cursor() 1155 spdef = """ 1156 CREATE PROCEDURE sp_DeleteMeOnlyForTesting 1157 @theInput DATETIME, 1158 @theOtherInput varchar(50), 1159 @theOutput varchar(100) OUTPUT 1160 AS 1161 SET @theOutput = CONVERT(CHARACTER(20), @theInput, 0) + @theOtherInput 1162 """ 1163 try: 1164 crsr.execute("DROP PROCEDURE sp_DeleteMeOnlyForTesting") 1165 self.conn.commit() 1166 except: # Make sure it is empty 1167 pass 1168 crsr.execute(spdef) 1169 1170 result = crsr.callproc( 1171 "sp_DeleteMeOnlyForTesting", 1172 [adodbapi.Timestamp(2014, 12, 25, 0, 1, 0), "Beep", " " * 30], 1173 ) 1174 1175 assert result[2] == "Dec 25 2014 12:01AM Beep", 'value was="%s"' % result[2] 1176 self.conn.rollback() 1177 1178 def testIncorrectStoredProcedureParameter(self): 1179 crsr = self.conn.cursor() 1180 spdef = """ 1181 CREATE PROCEDURE sp_DeleteMeOnlyForTesting 1182 @theInput DATETIME, 1183 @theOtherInput varchar(50), 1184 @theOutput varchar(100) OUTPUT 1185 AS 1186 SET @theOutput = CONVERT(CHARACTER(20), @theInput) + @theOtherInput 1187 """ 1188 try: 1189 crsr.execute("DROP PROCEDURE sp_DeleteMeOnlyForTesting") 1190 self.conn.commit() 1191 except: # Make sure it is empty 1192 pass 1193 crsr.execute(spdef) 1194 1195 # calling the sproc with a string for the first parameter where a DateTime is expected 1196 result = tryconnection.try_operation_with_expected_exception( 1197 (api.DataError, api.DatabaseError), 1198 crsr.callproc, 1199 ["sp_DeleteMeOnlyForTesting"], 1200 {"parameters": ["this is wrong", "Anne", "not Alice"]}, 1201 ) 1202 if result[0]: # the expected exception was raised 1203 assert "@theInput" in str(result[1]) or "DatabaseError" in str( 1204 result 1205 ), "Identifies the wrong erroneous parameter" 1206 else: 1207 assert result[0], result[1] # incorrect or no exception 1208 self.conn.rollback() 1209 1210 1211 class TestADOwithAccessDB(CommonDBTests): 1212 def setUp(self): 1213 self.conn = config.dbAccessconnect( 1214 *config.connStrAccess[0], **config.connStrAccess[1] 1215 ) 1216 self.conn.timeout = 30 # turn timeout back up 1217 self.engine = "ACCESS" 1218 self.db = config.dbAccessconnect 1219 1220 def tearDown(self): 1221 try: 1222 self.conn.rollback() 1223 except: 1224 pass 1225 try: 1226 self.conn.close() 1227 except: 1228 pass 1229 self.conn = None 1230 1231 def getConnection(self): 1232 return self.conn 1233 1234 def getAnotherConnection(self, addkeys=None): 1235 raise NotImplementedError("Jet cannot use a second connection to the database") 1236 1237 def testOkConnect(self): 1238 c = self.db(*config.connStrAccess[0], **config.connStrAccess[1]) 1239 assert c is not None 1240 c.close() 1241 1242 1243 class TestADOwithMySql(CommonDBTests): 1244 def setUp(self): 1245 self.conn = config.dbMySqlconnect( 1246 *config.connStrMySql[0], **config.connStrMySql[1] 1247 ) 1248 self.conn.timeout = 30 # turn timeout back up 1249 self.engine = "MySQL" 1250 self.db = config.dbMySqlconnect 1251 1252 def tearDown(self): 1253 try: 1254 self.conn.rollback() 1255 except: 1256 pass 1257 try: 1258 self.conn.close() 1259 except: 1260 pass 1261 self.conn = None 1262 1263 def getConnection(self): 1264 return self.conn 1265 1266 def getAnotherConnection(self, addkeys=None): 1267 keys = config.connStrMySql[1].copy() 1268 if addkeys: 1269 keys.update(addkeys) 1270 return config.dbMySqlconnect(*config.connStrMySql[0], **keys) 1271 1272 def testOkConnect(self): 1273 c = self.db(*config.connStrMySql[0], **config.connStrMySql[1]) 1274 assert c is not None 1275 1276 # def testStoredProcedure(self): 1277 # crsr = self.conn.cursor() 1278 # try: 1279 # crsr.execute("DROP PROCEDURE DeleteMeOnlyForTesting") 1280 # self.conn.commit() 1281 # except: # Make sure it is empty 1282 # pass 1283 # spdef = """ 1284 # DELIMITER $$ 1285 # CREATE PROCEDURE DeleteMeOnlyForTesting (onein CHAR(10), twoin CHAR(10), OUT theout CHAR(20)) 1286 # DETERMINISTIC 1287 # BEGIN 1288 # SET theout = onein //|| twoin; 1289 # /* (SELECT 'a small string' as result; */ 1290 # END $$ 1291 # """ 1292 # crsr.execute(spdef) 1293 # retvalues = crsr.callproc( 1294 # "DeleteMeOnlyForTesting", ("Dodsworth", "Anne", " ") 1295 # ) 1296 # # print(f"return value (mysql)={crsr.returnValue!r}") 1297 # assert retvalues[0] == "Dodsworth", f'{retvalues[0]!r} is not "Dodsworth"' 1298 # assert retvalues[1] == "Anne", f'{retvalues[1]!r} is not "Anne"' 1299 # assert ( 1300 # retvalues[2] == "DodsworthAnne" 1301 # ), f'{retvalues[2]!r} is not "DodsworthAnne"' 1302 # try: 1303 # crsr.execute("DROP PROCEDURE, DeleteMeOnlyForTesting") 1304 # self.conn.commit() 1305 # except: # Make sure it is empty 1306 # pass 1307 1308 1309 class TestADOwithPostgres(CommonDBTests): 1310 def setUp(self): 1311 self.conn = config.dbPostgresConnect( 1312 *config.connStrPostgres[0], **config.connStrPostgres[1] 1313 ) 1314 self.conn.timeout = 30 # turn timeout back up 1315 self.engine = "PostgreSQL" 1316 self.db = config.dbPostgresConnect 1317 1318 def tearDown(self): 1319 try: 1320 self.conn.rollback() 1321 except: 1322 pass 1323 try: 1324 self.conn.close() 1325 except: 1326 pass 1327 self.conn = None 1328 1329 def getConnection(self): 1330 return self.conn 1331 1332 def getAnotherConnection(self, addkeys=None): 1333 keys = config.connStrPostgres[1].copy() 1334 if addkeys: 1335 keys.update(addkeys) 1336 return config.dbPostgresConnect(*config.connStrPostgres[0], **keys) 1337 1338 def testOkConnect(self): 1339 c = self.db(*config.connStrPostgres[0], **config.connStrPostgres[1]) 1340 assert c is not None 1341 1342 # def testStoredProcedure(self): 1343 # crsr = self.conn.cursor() 1344 # spdef = """ 1345 # CREATE OR REPLACE FUNCTION DeleteMeOnlyForTesting (text, text) 1346 # RETURNS text AS $funk$ 1347 # BEGIN 1348 # RETURN $1 || $2; 1349 # END; 1350 # $funk$ 1351 # LANGUAGE SQL; 1352 # """ 1353 1354 # crsr.execute(spdef) 1355 # retvalues = crsr.callproc( 1356 # "DeleteMeOnlyForTesting", ("Dodsworth", "Anne", " ") 1357 # ) 1358 # # print(f"return value (pg)={crsr.returnValue!r}") 1359 # assert retvalues[0] == "Dodsworth", f'{retvalues[0]!r} is not "Dodsworth"' 1360 # assert retvalues[1] == "Anne", f'{retvalues[1]!r} is not "Anne"' 1361 # assert ( 1362 # retvalues[2] == "DodsworthAnne" 1363 # ), f'{retvalues[2]!r} is not "DodsworthAnne"' 1364 # self.conn.rollback() 1365 # try: 1366 # crsr.execute("DROP PROCEDURE, DeleteMeOnlyForTesting") 1367 # self.conn.commit() 1368 # except: # Make sure it is empty 1369 # pass 1370 1371 1372 class TimeConverterInterfaceTest(unittest.TestCase): 1373 def testIDate(self): 1374 assert self.tc.Date(1990, 2, 2) 1375 1376 def testITime(self): 1377 assert self.tc.Time(13, 2, 2) 1378 1379 def testITimestamp(self): 1380 assert self.tc.Timestamp(1990, 2, 2, 13, 2, 1) 1381 1382 def testIDateObjectFromCOMDate(self): 1383 assert self.tc.DateObjectFromCOMDate(37435.7604282) 1384 1385 def testICOMDate(self): 1386 assert hasattr(self.tc, "COMDate") 1387 1388 def testExactDate(self): 1389 d = self.tc.Date(1994, 11, 15) 1390 comDate = self.tc.COMDate(d) 1391 correct = 34653.0 1392 assert comDate == correct, comDate 1393 1394 def testExactTimestamp(self): 1395 d = self.tc.Timestamp(1994, 11, 15, 12, 0, 0) 1396 comDate = self.tc.COMDate(d) 1397 correct = 34653.5 1398 self.assertEqual(comDate, correct) 1399 1400 d = self.tc.Timestamp(2003, 5, 6, 14, 15, 17) 1401 comDate = self.tc.COMDate(d) 1402 correct = 37747.593946759262 1403 self.assertEqual(comDate, correct) 1404 1405 def testIsoFormat(self): 1406 d = self.tc.Timestamp(1994, 11, 15, 12, 3, 10) 1407 iso = self.tc.DateObjectToIsoFormatString(d) 1408 self.assertEqual(str(iso[:19]), "1994-11-15 12:03:10") 1409 1410 dt = self.tc.Date(2003, 5, 2) 1411 iso = self.tc.DateObjectToIsoFormatString(dt) 1412 self.assertEqual(str(iso[:10]), "2003-05-02") 1413 1414 1415 class TestPythonTimeConverter(TimeConverterInterfaceTest): 1416 def setUp(self): 1417 self.tc = api.pythonTimeConverter() 1418 1419 def testCOMDate(self): 1420 mk = time.mktime((2002, 6, 28, 18, 15, 1, 4, 31 + 28 + 31 + 30 + 31 + 28, -1)) 1421 t = time.localtime(mk) 1422 # Fri, 28 Jun 2002 18:15:01 +0000 1423 cmd = self.tc.COMDate(t) 1424 assert abs(cmd - 37435.7604282) < 1.0 / 24, "%f more than an hour wrong" % cmd 1425 1426 def testDateObjectFromCOMDate(self): 1427 cmd = self.tc.DateObjectFromCOMDate(37435.7604282) 1428 t1 = time.gmtime( 1429 time.mktime((2002, 6, 28, 0, 14, 1, 4, 31 + 28 + 31 + 30 + 31 + 28, -1)) 1430 ) 1431 # there are errors in the implementation of gmtime which we ignore 1432 t2 = time.gmtime( 1433 time.mktime((2002, 6, 29, 12, 14, 2, 4, 31 + 28 + 31 + 30 + 31 + 28, -1)) 1434 ) 1435 assert t1 < cmd < t2, f'"{cmd}" should be about 2002-6-28 12:15:01' 1436 1437 def testDate(self): 1438 t1 = time.mktime((2002, 6, 28, 18, 15, 1, 4, 31 + 28 + 31 + 30 + 31 + 30, 0)) 1439 t2 = time.mktime((2002, 6, 30, 18, 15, 1, 4, 31 + 28 + 31 + 30 + 31 + 28, 0)) 1440 obj = self.tc.Date(2002, 6, 29) 1441 assert t1 < time.mktime(obj) < t2, obj 1442 1443 def testTime(self): 1444 self.assertEqual( 1445 self.tc.Time(18, 15, 2), time.gmtime(18 * 60 * 60 + 15 * 60 + 2) 1446 ) 1447 1448 def testTimestamp(self): 1449 t1 = time.localtime( 1450 time.mktime((2002, 6, 28, 18, 14, 1, 4, 31 + 28 + 31 + 30 + 31 + 28, -1)) 1451 ) 1452 t2 = time.localtime( 1453 time.mktime((2002, 6, 28, 18, 16, 1, 4, 31 + 28 + 31 + 30 + 31 + 28, -1)) 1454 ) 1455 obj = self.tc.Timestamp(2002, 6, 28, 18, 15, 2) 1456 assert t1 < obj < t2, obj 1457 1458 1459 class TestPythonDateTimeConverter(TimeConverterInterfaceTest): 1460 def setUp(self): 1461 self.tc = api.pythonDateTimeConverter() 1462 1463 def testCOMDate(self): 1464 t = datetime.datetime(2002, 6, 28, 18, 15, 1) 1465 # Fri, 28 Jun 2002 18:15:01 +0000 1466 cmd = self.tc.COMDate(t) 1467 assert abs(cmd - 37435.7604282) < 1.0 / 24, "more than an hour wrong" 1468 1469 def testDateObjectFromCOMDate(self): 1470 cmd = self.tc.DateObjectFromCOMDate(37435.7604282) 1471 t1 = datetime.datetime(2002, 6, 28, 18, 14, 1) 1472 t2 = datetime.datetime(2002, 6, 28, 18, 16, 1) 1473 assert t1 < cmd < t2, cmd 1474 1475 tx = datetime.datetime( 1476 2002, 6, 28, 18, 14, 1, 900000 1477 ) # testing that microseconds don't become milliseconds 1478 c1 = self.tc.DateObjectFromCOMDate(self.tc.COMDate(tx)) 1479 assert t1 < c1 < t2, c1 1480 1481 def testDate(self): 1482 t1 = datetime.date(2002, 6, 28) 1483 t2 = datetime.date(2002, 6, 30) 1484 obj = self.tc.Date(2002, 6, 29) 1485 assert t1 < obj < t2, obj 1486 1487 def testTime(self): 1488 self.assertEqual(self.tc.Time(18, 15, 2).isoformat()[:8], "18:15:02") 1489 1490 def testTimestamp(self): 1491 t1 = datetime.datetime(2002, 6, 28, 18, 14, 1) 1492 t2 = datetime.datetime(2002, 6, 28, 18, 16, 1) 1493 obj = self.tc.Timestamp(2002, 6, 28, 18, 15, 2) 1494 assert t1 < obj < t2, obj 1495 1496 1497 suites = [ 1498 unittest.defaultTestLoader.loadTestsFromModule(TestPythonDateTimeConverter, "test") 1499 ] 1500 if config.doTimeTest: 1501 suites.append( 1502 unittest.defaultTestLoader.loadTestsFromModule(TestPythonTimeConverter, "test") 1503 ) 1504 if config.doAccessTest: 1505 suites.append( 1506 unittest.defaultTestLoader.loadTestsFromModule(TestADOwithAccessDB, "test") 1507 ) 1508 if config.doSqlServerTest: 1509 suites.append( 1510 unittest.defaultTestLoader.loadTestsFromModule(TestADOwithSQLServer, "test") 1511 ) 1512 if config.doMySqlTest: 1513 suites.append( 1514 unittest.defaultTestLoader.loadTestsFromModule(TestADOwithMySql, "test") 1515 ) 1516 if config.doPostgresTest: 1517 suites.append( 1518 unittest.defaultTestLoader.loadTestsFromModule(TestADOwithPostgres, "test") 1519 ) 1520 1521 1522 class cleanup_manager: 1523 def __enter__(self): 1524 pass 1525 1526 def __exit__(self, exc_type, exc_val, exc_tb): 1527 config.cleanup(config.testfolder, config.mdb_name) 1528 1529 1530 suite = unittest.TestSuite(suites) 1531 if __name__ == "__main__": 1532 mysuite = copy.deepcopy(suite) 1533 with cleanup_manager(): 1534 defaultDateConverter = adodbapi.dateconverter 1535 print(__doc__) 1536 print("Default Date Converter is %s" % (defaultDateConverter,)) 1537 dateconverter = defaultDateConverter 1538 unittest.TextTestRunner().run(mysuite) 1539 1540 if config.doTimeTest: 1541 mysuite = copy.deepcopy( 1542 suite 1543 ) # work around a side effect of unittest.TextTestRunner 1544 adodbapi.adodbapi.dateconverter = api.pythonTimeConverter() 1545 print("Changed dateconverter to ") 1546 print(adodbapi.adodbapi.dateconverter) 1547 unittest.TextTestRunner().run(mysuite) 1548