1388 lines
55 KiB
Python
1388 lines
55 KiB
Python
""" Unit tests version 2.6.1.0 for adodbapi"""
|
|
from __future__ import print_function
|
|
"""
|
|
adodbapi - A python DB API 2.0 interface to Microsoft ADO
|
|
|
|
Copyright (C) 2002 Henrik Ekelund
|
|
|
|
This library is free software; you can redistribute it and/or
|
|
modify it under the terms of the GNU Lesser General Public
|
|
License as published by the Free Software Foundation; either
|
|
version 2.1 of the License, or (at your option) any later version.
|
|
|
|
This library is distributed in the hope that it will be useful,
|
|
but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
|
|
Lesser General Public License for more details.
|
|
|
|
You should have received a copy of the GNU Lesser General Public
|
|
License along with this library; if not, write to the Free Software
|
|
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
|
|
|
|
Updates by Vernon Cole
|
|
"""
|
|
|
|
import unittest
|
|
import sys
|
|
import datetime
|
|
import decimal
|
|
import copy
|
|
import random
|
|
import string
|
|
|
|
try:
|
|
import win32com.client
|
|
win32 = True
|
|
except ImportError:
|
|
win32 = False
|
|
|
|
# run the configuration module.
|
|
import adodbapitestconfig as config # will set sys.path to find correct version of adodbapi
|
|
# in our code below, all our switches are from config.whatever
|
|
import tryconnection
|
|
|
|
import adodbapi
|
|
import adodbapi.apibase as api
|
|
|
|
|
|
try:
|
|
import adodbapi.ado_consts as ado_consts
|
|
except ImportError: #we are doing a shortcut import as a module -- so
|
|
try:
|
|
import ado_consts
|
|
except ImportError:
|
|
from adodbapi import ado_consts
|
|
|
|
if sys.version_info >= (3,0):
|
|
def str2bytes(sval):
|
|
return sval.encode("latin1")
|
|
str = str
|
|
long = int
|
|
else:
|
|
def str2bytes(sval):
|
|
if isinstance(sval, str):
|
|
return sval
|
|
return sval.encode("latin1")
|
|
try:
|
|
bytes
|
|
except NameError:
|
|
bytes = str
|
|
|
|
def randomstring(length):
|
|
return ''.join([random.choice(string.ascii_letters) for n in range(32)])
|
|
class CommonDBTests(unittest.TestCase):
|
|
"Self contained super-simple tests in easy syntax, should work on everything between mySQL and Oracle"
|
|
|
|
def setUp(self):
|
|
self.engine = 'unknown'
|
|
|
|
def getEngine(self):
|
|
return self.engine
|
|
|
|
def getConnection(self):
|
|
raise NotImplementedError #"This method must be overriden by a subclass"
|
|
|
|
def getCursor(self):
|
|
return self.getConnection().cursor()
|
|
|
|
def testConnection(self):
|
|
crsr=self.getCursor()
|
|
assert crsr.__class__.__name__ == 'Cursor'
|
|
|
|
def testErrorHandlerInherits(self):
|
|
if not self.remote:
|
|
conn=self.getConnection()
|
|
mycallable=lambda connection,cursor,errorclass,errorvalue: 1
|
|
conn.errorhandler=mycallable
|
|
crsr=conn.cursor()
|
|
assert crsr.errorhandler==mycallable,"Error handler on crsr should be same as on connection"
|
|
|
|
def testDefaultErrorHandlerConnection(self):
|
|
if not self.remote:
|
|
conn=self.getConnection()
|
|
del conn.messages[:]
|
|
try:
|
|
conn.close()
|
|
conn.commit() #Should not be able to use connection after it is closed
|
|
except:
|
|
assert len(conn.messages)==1
|
|
assert len(conn.messages[0])==2
|
|
assert conn.messages[0][0]==api.ProgrammingError
|
|
|
|
def testOwnErrorHandlerConnection(self):
|
|
if self.remote: # ToDo: use "skip"
|
|
return
|
|
mycallable=lambda connection,cursor,errorclass,errorvalue: 1 #does not raise anything
|
|
conn=self.getConnection()
|
|
conn.errorhandler=mycallable
|
|
conn.close()
|
|
conn.commit() #Should not be able to use connection after it is closed
|
|
assert len(conn.messages)==0
|
|
|
|
conn.errorhandler=None #This should bring back the standard error handler
|
|
try:
|
|
conn.close()
|
|
conn.commit() #Should not be able to use connection after it is closed
|
|
except:
|
|
pass
|
|
#The Standard errorhandler appends error to messages attribute
|
|
assert len(conn.messages)>0,"Setting errorhandler to none should bring back the standard error handler"
|
|
|
|
|
|
def testDefaultErrorHandlerCursor(self):
|
|
crsr=self.getConnection().cursor()
|
|
if not self.remote:
|
|
del crsr.messages[:]
|
|
try:
|
|
crsr.execute("SELECT abbtytddrf FROM dasdasd")
|
|
except:
|
|
assert len(crsr.messages)==1
|
|
assert len(crsr.messages[0])==2
|
|
assert crsr.messages[0][0]==api.DatabaseError
|
|
|
|
def testOwnErrorHandlerCursor(self):
|
|
if self.remote: # ToDo: should be a "skip"
|
|
return
|
|
mycallable=lambda connection,cursor,errorclass,errorvalue: 1 #does not raise anything
|
|
crsr=self.getConnection().cursor()
|
|
crsr.errorhandler=mycallable
|
|
crsr.execute("SELECT abbtytddrf FROM dasdasd")
|
|
assert len(crsr.messages)==0
|
|
|
|
crsr.errorhandler=None #This should bring back the standard error handler
|
|
try:
|
|
crsr.execute("SELECT abbtytddrf FROM dasdasd")
|
|
except:
|
|
pass
|
|
#The Standard errorhandler appends error to messages attribute
|
|
assert len(crsr.messages)>0,"Setting errorhandler to none should bring back the standard error handler"
|
|
|
|
|
|
def testUserDefinedConversions(self):
|
|
if self.remote: ## Todo: should be a "skip"
|
|
return
|
|
try:
|
|
duplicatingConverter=lambda aStringField: aStringField*2
|
|
assert duplicatingConverter('gabba') == 'gabbagabba'
|
|
|
|
self.helpForceDropOnTblTemp()
|
|
conn=self.getConnection()
|
|
# the variantConversions attribute should not exist on a normal connection object
|
|
self.assertRaises(AttributeError, lambda x:conn.variantConversions[x],[2])
|
|
if not self.remote:
|
|
# create a variantConversions attribute on the connection
|
|
conn.variantConversions = copy.copy(api.variantConversions)
|
|
crsr=conn.cursor()
|
|
tabdef = "CREATE TABLE xx_%s (fldData VARCHAR(100) NOT NULL, fld2 VARCHAR(20))" % config.tmp
|
|
crsr.execute(tabdef)
|
|
crsr.execute("INSERT INTO xx_%s(fldData,fld2) VALUES('gabba','booga')" % config.tmp)
|
|
crsr.execute("INSERT INTO xx_%s(fldData,fld2) VALUES('hey','yo')" % config.tmp)
|
|
# change converter for ALL adoStringTypes columns
|
|
conn.variantConversions[api.adoStringTypes]=duplicatingConverter
|
|
crsr.execute("SELECT fldData,fld2 FROM xx_%s ORDER BY fldData" % config.tmp)
|
|
|
|
rows=crsr.fetchall()
|
|
row = rows[0]
|
|
self.assertEqual(row[0],'gabbagabba')
|
|
row = rows[1]
|
|
self.assertEqual(row[0],'heyhey')
|
|
self.assertEqual(row[1],'yoyo')
|
|
|
|
upcaseConverter=lambda aStringField: aStringField.upper()
|
|
assert upcaseConverter('upThis') == 'UPTHIS'
|
|
|
|
# now use a single column converter
|
|
rows.converters[1] = upcaseConverter # convert second column
|
|
self.assertEqual(row[0],'heyhey') # first will be unchanged
|
|
self.assertEqual(row[1],'YO') # second will convert to upper case
|
|
|
|
finally:
|
|
try:
|
|
del conn.variantConversions #Restore the default
|
|
except: pass
|
|
self.helpRollbackTblTemp()
|
|
|
|
def testUserDefinedConversionForExactNumericTypes(self):
|
|
# variantConversions is a dictionary of conversion functions
|
|
# held internally in adodbapi.apibase
|
|
#
|
|
# !!! this test intentionally alters the value of what should be constant in the module
|
|
# !!! no new code should use this example, to is only a test to see that the
|
|
# !!! deprecated way of doing this still works. (use connection.variantConversions)
|
|
#
|
|
if not self.remote and sys.version_info < (3,0): ### Py3 need different test
|
|
oldconverter = adodbapi.variantConversions[ado_consts.adNumeric] #keep old function to restore later
|
|
# By default decimal and "numbers" are returned as decimals.
|
|
# Instead, make numbers return as floats
|
|
try:
|
|
adodbapi.variantConversions[ado_consts.adNumeric] = adodbapi.cvtFloat
|
|
self.helpTestDataType("decimal(18,2)",'NUMBER',3.45,compareAlmostEqual=1)
|
|
self.helpTestDataType("numeric(18,2)",'NUMBER',3.45,compareAlmostEqual=1)
|
|
# now return strings
|
|
adodbapi.variantConversions[ado_consts.adNumeric] = adodbapi.cvtString
|
|
self.helpTestDataType("numeric(18,2)",'NUMBER','3.45')
|
|
# now a completly weird user defined convertion
|
|
adodbapi.variantConversions[ado_consts.adNumeric] = lambda x: '!!This function returns a funny unicode string %s!!'%x
|
|
self.helpTestDataType("numeric(18,2)",'NUMBER','3.45',
|
|
allowedReturnValues=['!!This function returns a funny unicode string 3.45!!'])
|
|
finally:
|
|
# now reset the converter to its original function
|
|
adodbapi.variantConversions[ado_consts.adNumeric]=oldconverter #Restore the original convertion function
|
|
|
|
def helpTestDataType(self,sqlDataTypeString,
|
|
DBAPIDataTypeString,
|
|
pyData,
|
|
pyDataInputAlternatives=None,
|
|
compareAlmostEqual=None,
|
|
allowedReturnValues=None):
|
|
self.helpForceDropOnTblTemp()
|
|
conn=self.getConnection()
|
|
crsr=conn.cursor()
|
|
tabdef= """
|
|
CREATE TABLE xx_%s (
|
|
fldId integer NOT NULL,
|
|
fldData """ % config.tmp + sqlDataTypeString + ")\n"
|
|
|
|
crsr.execute(tabdef)
|
|
|
|
#Test Null values mapped to None
|
|
crsr.execute("INSERT INTO xx_%s (fldId) VALUES (1)" % config.tmp)
|
|
|
|
crsr.execute("SELECT fldId,fldData FROM xx_%s" % config.tmp)
|
|
rs=crsr.fetchone()
|
|
self.assertEqual(rs[1],None) #Null should be mapped to None
|
|
assert rs[0]==1
|
|
|
|
#Test description related
|
|
descTuple=crsr.description[1]
|
|
assert descTuple[0] in ['fldData','flddata'], 'was "%s" expected "%s"'%(descTuple[0],'fldData')
|
|
|
|
if DBAPIDataTypeString=='STRING':
|
|
assert descTuple[1] == api.STRING, 'was "%s" expected "%s"'%(descTuple[1],api.STRING.values)
|
|
elif DBAPIDataTypeString == 'NUMBER':
|
|
assert descTuple[1] == api.NUMBER, 'was "%s" expected "%s"'%(descTuple[1],api.NUMBER.values)
|
|
elif DBAPIDataTypeString == 'BINARY':
|
|
assert descTuple[1] == api.BINARY, 'was "%s" expected "%s"'%(descTuple[1],api.BINARY.values)
|
|
elif DBAPIDataTypeString == 'DATETIME':
|
|
assert descTuple[1] == api.DATETIME, 'was "%s" expected "%s"'%(descTuple[1],api.DATETIME.values)
|
|
elif DBAPIDataTypeString == 'ROWID':
|
|
assert descTuple[1] == api.ROWID, 'was "%s" expected "%s"'%(descTuple[1],api.ROWID.values)
|
|
elif DBAPIDataTypeString == 'UUID':
|
|
assert descTuple[1] == api.OTHER, 'was "%s" expected "%s"'%(descTuple[1],api.OTHER.values)
|
|
else:
|
|
raise NotImplementedError #"DBAPIDataTypeString not provided"
|
|
|
|
#Test data binding
|
|
inputs=[pyData]
|
|
if pyDataInputAlternatives:
|
|
inputs.extend(pyDataInputAlternatives)
|
|
if str is str:
|
|
inputs = set(inputs) # removes redundant string==unicode tests
|
|
fldId=1
|
|
for inParam in inputs:
|
|
fldId+=1
|
|
try:
|
|
crsr.execute("INSERT INTO xx_%s (fldId,fldData) VALUES (?,?)" % config.tmp, (fldId, inParam))
|
|
except:
|
|
if self.remote:
|
|
for message in crsr.messages:
|
|
print(message)
|
|
else:
|
|
conn.printADOerrors()
|
|
raise
|
|
crsr.execute("SELECT fldData FROM xx_%s WHERE ?=fldID" % config.tmp, [fldId])
|
|
rs=crsr.fetchone()
|
|
if allowedReturnValues:
|
|
allowedTypes = tuple([type(aRV) for aRV in allowedReturnValues])
|
|
assert isinstance(rs[0],allowedTypes), \
|
|
'result type "%s" must be one of %s'%(type(rs[0]),allowedTypes)
|
|
else:
|
|
assert isinstance(rs[0] ,type(pyData)), \
|
|
'result type "%s" must be instance of %s'%(type(rs[0]),type(pyData))
|
|
|
|
if compareAlmostEqual and DBAPIDataTypeString == 'DATETIME':
|
|
iso1=adodbapi.dateconverter.DateObjectToIsoFormatString(rs[0])
|
|
iso2=adodbapi.dateconverter.DateObjectToIsoFormatString(pyData)
|
|
self.assertEqual(iso1, iso2)
|
|
elif compareAlmostEqual:
|
|
s = float(pyData)
|
|
v = float(rs[0])
|
|
assert abs(v-s)/s < 0.00001, \
|
|
"Values not almost equal recvd=%s, expected=%f" %(rs[0],s)
|
|
else:
|
|
if allowedReturnValues:
|
|
ok=False
|
|
self.assertTrue(rs[0] in allowedReturnValues,
|
|
'Value "%s" not in %s' % (repr(rs[0]), allowedReturnValues))
|
|
else:
|
|
self.assertEqual(rs[0], pyData,
|
|
'Values are not equal recvd="%s", expected="%s"' %(rs[0],pyData))
|
|
|
|
def testDataTypeFloat(self):
|
|
self.helpTestDataType("real",'NUMBER',3.45,compareAlmostEqual=True)
|
|
self.helpTestDataType("float",'NUMBER',1.79e37,compareAlmostEqual=True)
|
|
|
|
def testDataTypeDecmal(self):
|
|
self.helpTestDataType("decimal(18,2)",'NUMBER',3.45,
|
|
allowedReturnValues=['3.45','3,45',decimal.Decimal('3.45')])
|
|
self.helpTestDataType("numeric(18,2)",'NUMBER',3.45,
|
|
allowedReturnValues=['3.45','3,45',decimal.Decimal('3.45')])
|
|
self.helpTestDataType("decimal(20,2)",'NUMBER',444444444444444444,
|
|
allowedReturnValues=['444444444444444444.00', '444444444444444444,00',
|
|
decimal.Decimal('444444444444444444')])
|
|
if self.getEngine() == 'MSSQL':
|
|
self.helpTestDataType("uniqueidentifier",'UUID','{71A4F49E-39F3-42B1-A41E-48FF154996E6}',
|
|
allowedReturnValues=['{71A4F49E-39F3-42B1-A41E-48FF154996E6}'])
|
|
|
|
def testDataTypeMoney(self): #v2.1 Cole -- use decimal for money
|
|
if self.getEngine() == 'MySQL':
|
|
self.helpTestDataType("DECIMAL(20,4)",'NUMBER',decimal.Decimal('-922337203685477.5808'))
|
|
elif self.getEngine() == 'PostgreSQL':
|
|
self.helpTestDataType("money",'NUMBER',decimal.Decimal('-922337203685477.5808'),
|
|
compareAlmostEqual=True,
|
|
allowedReturnValues=[-922337203685477.5808,
|
|
decimal.Decimal('-922337203685477.5808')])
|
|
else:
|
|
self.helpTestDataType("smallmoney",'NUMBER',decimal.Decimal('214748.02'))
|
|
self.helpTestDataType("money",'NUMBER',decimal.Decimal('-922337203685477.5808'))
|
|
|
|
def testDataTypeInt(self):
|
|
if self.getEngine() != 'PostgreSQL':
|
|
self.helpTestDataType("tinyint",'NUMBER',115)
|
|
self.helpTestDataType("smallint",'NUMBER',-32768)
|
|
if self.getEngine() not in ['ACCESS','PostgreSQL']:
|
|
self.helpTestDataType("bit",'NUMBER',1) #Does not work correctly with access
|
|
if self.getEngine() in ['MSSQL','PostgreSQL']:
|
|
self.helpTestDataType("bigint",'NUMBER',3000000000,
|
|
allowedReturnValues=[3000000000, int(3000000000)])
|
|
self.helpTestDataType("int",'NUMBER',2147483647)
|
|
|
|
def testDataTypeChar(self):
|
|
for sqlDataType in ("char(6)","nchar(6)"):
|
|
self.helpTestDataType(sqlDataType,'STRING','spam ',allowedReturnValues=['spam','spam','spam ','spam '])
|
|
|
|
def testDataTypeVarChar(self):
|
|
if self.getEngine() == 'MySQL':
|
|
stringKinds = ["varchar(10)","text"]
|
|
elif self.getEngine() == 'PostgreSQL':
|
|
stringKinds = ["varchar(10)","text","character varying"]
|
|
else:
|
|
stringKinds = ["varchar(10)","nvarchar(10)","text","ntext"] #,"varchar(max)"]
|
|
|
|
for sqlDataType in stringKinds:
|
|
self.helpTestDataType(sqlDataType,'STRING','spam',['spam'])
|
|
|
|
def testDataTypeDate(self):
|
|
if self.getEngine() == 'PostgreSQL':
|
|
dt = "timestamp"
|
|
else:
|
|
dt = "datetime"
|
|
self.helpTestDataType(dt,'DATETIME',adodbapi.Date(2002,10,28),
|
|
compareAlmostEqual=True)
|
|
if self.getEngine() not in ['MySQL','PostgreSQL']:
|
|
self.helpTestDataType("smalldatetime",'DATETIME',adodbapi.Date(2002,10,28),
|
|
compareAlmostEqual=True)
|
|
if tag != 'pythontime' and self.getEngine() not in ['MySQL','PostgreSQL']: # fails when using pythonTime
|
|
self.helpTestDataType(dt,'DATETIME', adodbapi.Timestamp(2002,10,28,12,15,1),
|
|
compareAlmostEqual=True)
|
|
|
|
def testDataTypeBinary(self):
|
|
binfld = str2bytes('\x07\x00\xE2\x40*')
|
|
arv = [binfld, adodbapi.Binary(binfld), bytes(binfld)]
|
|
if self.getEngine() == 'PostgreSQL':
|
|
self.helpTestDataType("bytea",'BINARY',adodbapi.Binary(binfld),
|
|
allowedReturnValues=arv)
|
|
else:
|
|
self.helpTestDataType("binary(5)",'BINARY',adodbapi.Binary(binfld),
|
|
allowedReturnValues=arv)
|
|
self.helpTestDataType("varbinary(100)",'BINARY',adodbapi.Binary(binfld),
|
|
allowedReturnValues=arv)
|
|
if self.getEngine() != 'MySQL':
|
|
self.helpTestDataType("image",'BINARY',adodbapi.Binary(binfld),
|
|
allowedReturnValues=arv)
|
|
|
|
def helpRollbackTblTemp(self):
|
|
self.helpForceDropOnTblTemp()
|
|
|
|
def helpForceDropOnTblTemp(self):
|
|
conn=self.getConnection()
|
|
with conn.cursor() as crsr:
|
|
try:
|
|
crsr.execute("DROP TABLE xx_%s" % config.tmp)
|
|
if not conn.autocommit:
|
|
conn.commit()
|
|
except:
|
|
pass
|
|
|
|
def helpCreateAndPopulateTableTemp(self,crsr):
|
|
tabdef= """
|
|
CREATE TABLE xx_%s (
|
|
fldData INTEGER
|
|
)
|
|
""" % config.tmp
|
|
try: #EAFP
|
|
crsr.execute(tabdef)
|
|
except api.DatabaseError: # was not dropped before
|
|
self.helpForceDropOnTblTemp() # so drop it now
|
|
crsr.execute(tabdef)
|
|
for i in range(9): # note: this poor SQL code, but a valid test
|
|
crsr.execute("INSERT INTO xx_%s (fldData) VALUES (%i)" % (config.tmp, i))
|
|
# NOTE: building the test table without using parameter substitution
|
|
|
|
def testFetchAll(self):
|
|
crsr=self.getCursor()
|
|
self.helpCreateAndPopulateTableTemp(crsr)
|
|
crsr.execute("SELECT fldData FROM xx_%s" % config.tmp)
|
|
rs=crsr.fetchall()
|
|
assert len(rs)==9
|
|
#test slice of rows
|
|
i = 3
|
|
for row in rs[3:-2]: #should have rowid 3..6
|
|
assert row[0]==i
|
|
i+=1
|
|
self.helpRollbackTblTemp()
|
|
|
|
def testPreparedStatement(self):
|
|
crsr=self.getCursor()
|
|
self.helpCreateAndPopulateTableTemp(crsr)
|
|
crsr.prepare("SELECT fldData FROM xx_%s" % config.tmp)
|
|
crsr.execute(crsr.command) # remembes the one that was prepared
|
|
rs=crsr.fetchall()
|
|
assert len(rs)==9
|
|
assert rs[2][0]==2
|
|
self.helpRollbackTblTemp()
|
|
|
|
def testWrongPreparedStatement(self):
|
|
crsr=self.getCursor()
|
|
self.helpCreateAndPopulateTableTemp(crsr)
|
|
crsr.prepare("SELECT * FROM nowhere")
|
|
crsr.execute("SELECT fldData FROM xx_%s" % config.tmp) # should execute this one, not the prepared one
|
|
rs=crsr.fetchall()
|
|
assert len(rs)==9
|
|
assert rs[2][0]==2
|
|
self.helpRollbackTblTemp()
|
|
|
|
def testIterator(self):
|
|
crsr=self.getCursor()
|
|
self.helpCreateAndPopulateTableTemp(crsr)
|
|
crsr.execute("SELECT fldData FROM xx_%s" % config.tmp)
|
|
for i,row in enumerate(crsr): # using cursor as an iterator, rather than fetchxxx
|
|
assert row[0]==i
|
|
self.helpRollbackTblTemp()
|
|
|
|
def testExecuteMany(self):
|
|
crsr=self.getCursor()
|
|
self.helpCreateAndPopulateTableTemp(crsr)
|
|
seq_of_values = [ (111,) , (222,) ]
|
|
crsr.executemany("INSERT INTO xx_%s (fldData) VALUES (?)" % config.tmp, seq_of_values)
|
|
if crsr.rowcount==-1:
|
|
print(self.getEngine()+" Provider does not support rowcount (on .executemany())")
|
|
else:
|
|
self.assertEqual( crsr.rowcount,2)
|
|
crsr.execute("SELECT fldData FROM xx_%s" % config.tmp)
|
|
rs=crsr.fetchall()
|
|
assert len(rs)==11
|
|
self.helpRollbackTblTemp()
|
|
|
|
|
|
def testRowCount(self):
|
|
crsr=self.getCursor()
|
|
self.helpCreateAndPopulateTableTemp(crsr)
|
|
crsr.execute("SELECT fldData FROM xx_%s" % config.tmp)
|
|
if crsr.rowcount == -1:
|
|
#print("provider does not support rowcount on select")
|
|
pass
|
|
else:
|
|
self.assertEqual( crsr.rowcount,9)
|
|
self.helpRollbackTblTemp()
|
|
|
|
def testRowCountNoRecordset(self):
|
|
crsr=self.getCursor()
|
|
self.helpCreateAndPopulateTableTemp(crsr)
|
|
crsr.execute("DELETE FROM xx_%s WHERE fldData >= 5" % config.tmp)
|
|
if crsr.rowcount==-1:
|
|
print(self.getEngine()+" Provider does not support rowcount (on DELETE)")
|
|
else:
|
|
self.assertEqual( crsr.rowcount,4)
|
|
self.helpRollbackTblTemp()
|
|
|
|
def testFetchMany(self):
|
|
crsr=self.getCursor()
|
|
self.helpCreateAndPopulateTableTemp(crsr)
|
|
crsr.execute("SELECT fldData FROM xx_%s" % config.tmp)
|
|
rs=crsr.fetchmany(3)
|
|
assert len(rs)==3
|
|
rs=crsr.fetchmany(5)
|
|
assert len(rs)==5
|
|
rs=crsr.fetchmany(5)
|
|
assert len(rs)==1 #Asked for five, but there is only one left
|
|
self.helpRollbackTblTemp()
|
|
|
|
def testFetchManyWithArraySize(self):
|
|
crsr=self.getCursor()
|
|
self.helpCreateAndPopulateTableTemp(crsr)
|
|
crsr.execute("SELECT fldData FROM xx_%s" % config.tmp)
|
|
rs=crsr.fetchmany()
|
|
assert len(rs)==1 #arraysize Defaults to one
|
|
crsr.arraysize=4
|
|
rs=crsr.fetchmany()
|
|
assert len(rs)==4
|
|
rs=crsr.fetchmany()
|
|
assert len(rs)==4
|
|
rs=crsr.fetchmany()
|
|
assert len(rs)==0
|
|
self.helpRollbackTblTemp()
|
|
|
|
def testErrorConnect(self):
|
|
conn = self.getConnection()
|
|
kw = {}
|
|
if 'proxy_host' in conn.kwargs:
|
|
kw['proxy_host'] = conn.kwargs['proxy_host']
|
|
conn.close()
|
|
self.assertRaises(api.DatabaseError, self.db, 'not a valid connect string', kw)
|
|
|
|
def testRowIterator(self):
|
|
self.helpForceDropOnTblTemp()
|
|
conn=self.getConnection()
|
|
crsr=conn.cursor()
|
|
tabdef= """
|
|
CREATE TABLE xx_%s (
|
|
fldId integer NOT NULL,
|
|
fldTwo integer,
|
|
fldThree integer,
|
|
fldFour integer)
|
|
""" % config.tmp
|
|
crsr.execute(tabdef)
|
|
|
|
inputs = [(2,3,4),(102,103,104)]
|
|
fldId=1
|
|
for inParam in inputs:
|
|
fldId+=1
|
|
try:
|
|
crsr.execute("INSERT INTO xx_%s (fldId,fldTwo,fldThree,fldFour) VALUES (?,?,?,?)" % config.tmp,
|
|
(fldId,inParam[0],inParam[1],inParam[2]))
|
|
except:
|
|
if self.remote:
|
|
for message in crsr.messages:
|
|
print(message)
|
|
else:
|
|
conn.printADOerrors()
|
|
raise
|
|
crsr.execute("SELECT fldTwo,fldThree,fldFour FROM xx_%s WHERE ?=fldID" % config.tmp, [fldId])
|
|
rec = crsr.fetchone()
|
|
# check that stepping through an emulated row works
|
|
for j in range(len(inParam)):
|
|
assert rec[j] == inParam[j], 'returned value:"%s" != test value:"%s"'%(rec[j],inParam[j])
|
|
# check that we can get a complete tuple from a row
|
|
assert tuple(rec) == inParam, 'returned value:"%s" != test value:"%s"'%(repr(rec),repr(inParam))
|
|
# test that slices of rows work
|
|
slice1 = tuple(rec[:-1])
|
|
slice2 = tuple(inParam[0:2])
|
|
assert slice1 == slice2, 'returned value:"%s" != test value:"%s"'%(repr(slice1),repr(slice2))
|
|
# now test named column retrieval
|
|
assert rec['fldTwo'] == inParam[0]
|
|
assert rec.fldThree == inParam[1]
|
|
assert rec.fldFour == inParam[2]
|
|
# test array operation
|
|
# note that the fields vv vv vv are out of order
|
|
crsr.execute("select fldThree,fldFour,fldTwo from xx_%s" % config.tmp)
|
|
recs = crsr.fetchall()
|
|
assert recs[1][0] == 103
|
|
assert recs[0][1] == 4
|
|
assert recs[1]['fldFour'] == 104
|
|
assert recs[0,0] == 3
|
|
assert recs[0,'fldTwo'] == 2
|
|
assert recs[1,2] == 102
|
|
for i in range(1):
|
|
for j in range(2):
|
|
assert recs[i][j] == recs[i,j]
|
|
|
|
def testFormatParamstyle(self):
|
|
self.helpForceDropOnTblTemp()
|
|
conn=self.getConnection()
|
|
conn.paramstyle = 'format' #test nonstandard use of paramstyle
|
|
crsr=conn.cursor()
|
|
tabdef= """
|
|
CREATE TABLE xx_%s (
|
|
fldId integer NOT NULL,
|
|
fldData varchar(10),
|
|
fldConst varchar(30))
|
|
""" % config.tmp
|
|
crsr.execute(tabdef)
|
|
|
|
inputs = ['one','two','three']
|
|
fldId=2
|
|
for inParam in inputs:
|
|
fldId+=1
|
|
sql = "INSERT INTO xx_" + \
|
|
config.tmp + \
|
|
" (fldId,fldConst,fldData) VALUES (%s,'thi%s :may cause? trouble', %s)"
|
|
try:
|
|
crsr.execute(sql, (fldId,inParam))
|
|
except:
|
|
if self.remote:
|
|
for message in crsr.messages:
|
|
print(message)
|
|
else:
|
|
conn.printADOerrors()
|
|
raise
|
|
crsr.execute("SELECT fldData, fldConst FROM xx_" + config.tmp + " WHERE %s=fldID", [fldId])
|
|
rec = crsr.fetchone()
|
|
self.assertEqual(rec[0], inParam, 'returned value:"%s" != test value:"%s"' % (rec[0],inParam))
|
|
self.assertEqual(rec[1], "thi%s :may cause? trouble")
|
|
|
|
# now try an operation with a "%s" as part of a literal
|
|
sel = "insert into xx_" + config.tmp + " (fldId,fldData) VALUES (%s,'four%sfive')"
|
|
params = (20,)
|
|
crsr.execute(sel,params)
|
|
|
|
#test the .query implementation
|
|
assert '(?,' in crsr.query, 'expected:"%s" in "%s"'%('(?,',crsr.query)
|
|
#test the .command attribute
|
|
assert crsr.command == sel, 'expected:"%s" but found "%s"' % (sel, crsr.command)
|
|
|
|
#test the .parameters attribute
|
|
if not self.remote: # parameter list will be altered in transit
|
|
self.assertEqual(crsr.parameters, params)
|
|
#now make sure the data made it
|
|
crsr.execute("SELECT fldData FROM xx_%s WHERE fldID=20" % config.tmp)
|
|
rec = crsr.fetchone()
|
|
self.assertEqual(rec[0], 'four%sfive')
|
|
|
|
def testNamedParamstyle(self):
|
|
self.helpForceDropOnTblTemp()
|
|
conn=self.getConnection()
|
|
crsr=conn.cursor()
|
|
crsr.paramstyle = 'named' #test nonstandard use of paramstyle
|
|
tabdef= """
|
|
CREATE TABLE xx_%s (
|
|
fldId integer NOT NULL,
|
|
fldData varchar(10))
|
|
""" % config.tmp
|
|
crsr.execute(tabdef)
|
|
|
|
inputs = ['four','five','six']
|
|
fldId=10
|
|
for inParam in inputs:
|
|
fldId+=1
|
|
try:
|
|
crsr.execute("INSERT INTO xx_%s (fldId,fldData) VALUES (:Id,:f_Val)" % config.tmp,
|
|
{"f_Val":inParam,'Id':fldId})
|
|
except:
|
|
if self.remote:
|
|
for message in crsr.messages:
|
|
print(message)
|
|
else:
|
|
conn.printADOerrors()
|
|
raise
|
|
crsr.execute("SELECT fldData FROM xx_%s WHERE fldID=:Id" % config.tmp, {'Id':fldId})
|
|
rec = crsr.fetchone()
|
|
self.assertEqual(rec[0], inParam, 'returned value:"%s" != test value:"%s"'%(rec[0],inParam))
|
|
# now a test with a ":" as part of a literal
|
|
crsr.execute("insert into xx_%s (fldId,fldData) VALUES (:xyz,'six:five')" % config.tmp,{'xyz':30})
|
|
crsr.execute("SELECT fldData FROM xx_%s WHERE fldID=30" % config.tmp)
|
|
rec = crsr.fetchone()
|
|
self.assertEqual(rec[0], 'six:five')
|
|
|
|
|
|
def testPyformatParamstyle(self):
|
|
self.helpForceDropOnTblTemp()
|
|
conn=self.getConnection()
|
|
crsr=conn.cursor()
|
|
crsr.paramstyle = 'pyformat' #test nonstandard use of paramstyle
|
|
tabdef= """
|
|
CREATE TABLE xx_%s (
|
|
fldId integer NOT NULL,
|
|
fldData varchar(10))
|
|
""" % config.tmp
|
|
crsr.execute(tabdef)
|
|
|
|
inputs = ['four', 'five', 'six']
|
|
fldId=10
|
|
for inParam in inputs:
|
|
fldId+=1
|
|
try:
|
|
crsr.execute("INSERT INTO xx_%s (fldId,fldData) VALUES (%%(Id)s,%%(f_Val)s)" % config.tmp,
|
|
{"f_Val": inParam, 'Id': fldId})
|
|
except:
|
|
if self.remote:
|
|
for message in crsr.messages:
|
|
print(message)
|
|
else:
|
|
conn.printADOerrors()
|
|
raise
|
|
crsr.execute("SELECT fldData FROM xx_%s WHERE fldID=%%(Id)s" % config.tmp, {'Id':fldId})
|
|
rec = crsr.fetchone()
|
|
self.assertEqual(rec[0], inParam, 'returned value:"%s" != test value:"%s"'%(rec[0],inParam))
|
|
# now a test with a "%" as part of a literal
|
|
crsr.execute("insert into xx_%s (fldId,fldData) VALUES (%%(xyz)s,'six%%five')" % config.tmp,{'xyz': 30})
|
|
crsr.execute("SELECT fldData FROM xx_%s WHERE fldID=30" % config.tmp)
|
|
rec = crsr.fetchone()
|
|
self.assertEqual(rec[0], 'six%five')
|
|
|
|
def testAutomaticParamstyle(self):
|
|
self.helpForceDropOnTblTemp()
|
|
conn=self.getConnection()
|
|
conn.paramstyle = 'dynamic' #test nonstandard use of paramstyle
|
|
crsr=conn.cursor()
|
|
tabdef= """
|
|
CREATE TABLE xx_%s (
|
|
fldId integer NOT NULL,
|
|
fldData varchar(10),
|
|
fldConst varchar(30))
|
|
""" % config.tmp
|
|
crsr.execute(tabdef)
|
|
inputs = ['one', 'two', 'three']
|
|
fldId=2
|
|
for inParam in inputs:
|
|
fldId+=1
|
|
try:
|
|
crsr.execute("INSERT INTO xx_" + config.tmp + \
|
|
" (fldId,fldConst,fldData) VALUES (?,'thi%s :may cause? troub:1e', ?)", (fldId,inParam))
|
|
except:
|
|
if self.remote:
|
|
for message in crsr.messages:
|
|
print(message)
|
|
else:
|
|
conn.printADOerrors()
|
|
raise
|
|
trouble = 'thi%s :may cause? troub:1e'
|
|
crsr.execute("SELECT fldData, fldConst FROM xx_" + config.tmp + " WHERE ?=fldID", [fldId])
|
|
rec = crsr.fetchone()
|
|
self.assertEqual(rec[0], inParam, 'returned value:"%s" != test value:"%s"'%(rec[0],inParam))
|
|
self.assertEqual(rec[1], trouble)
|
|
# inputs = [u'four',u'five',u'six']
|
|
fldId=10
|
|
for inParam in inputs:
|
|
fldId+=1
|
|
try:
|
|
crsr.execute("INSERT INTO xx_%s (fldId,fldData) VALUES (:Id,:f_Val)" % config.tmp,
|
|
{"f_Val":inParam,'Id':fldId})
|
|
except:
|
|
if self.remote:
|
|
for message in crsr.messages:
|
|
print(message)
|
|
else:
|
|
conn.printADOerrors()
|
|
raise
|
|
crsr.execute("SELECT fldData FROM xx_%s WHERE :Id=fldID" % config.tmp, {'Id':fldId})
|
|
rec = crsr.fetchone()
|
|
self.assertEqual(rec[0], inParam, 'returned value:"%s" != test value:"%s"'%(rec[0],inParam))
|
|
# now a test with a ":" as part of a literal -- and use a prepared query
|
|
ppdcmd = "insert into xx_%s (fldId,fldData) VALUES (:xyz,'six:five')" % config.tmp
|
|
crsr.prepare(ppdcmd)
|
|
crsr.execute(ppdcmd, {'xyz':30})
|
|
crsr.execute("SELECT fldData FROM xx_%s WHERE fldID=30" % config.tmp)
|
|
rec = crsr.fetchone()
|
|
self.assertEqual(rec[0], 'six:five')
|
|
|
|
def testRollBack(self):
|
|
conn = self.getConnection()
|
|
crsr = conn.cursor()
|
|
assert not crsr.connection.autocommit, 'Unexpected beginning condition'
|
|
self.helpCreateAndPopulateTableTemp(crsr)
|
|
crsr.connection.commit() # commit the first bunch
|
|
|
|
crsr.execute("INSERT INTO xx_%s (fldData) VALUES(100)" % config.tmp)
|
|
|
|
selectSql = "SELECT fldData FROM xx_%s WHERE fldData=100" % config.tmp
|
|
crsr.execute(selectSql)
|
|
rs = crsr.fetchall()
|
|
assert len(rs) == 1
|
|
self.conn.rollback()
|
|
crsr.execute(selectSql)
|
|
assert crsr.fetchone() == None, 'cursor.fetchone should return None if a query retrieves no rows'
|
|
crsr.execute('SELECT fldData from xx_%s' % config.tmp)
|
|
rs = crsr.fetchall()
|
|
assert len(rs) == 9, 'the original records should still be present'
|
|
self.helpRollbackTblTemp()
|
|
|
|
|
|
def testCommit(self):
|
|
try:
|
|
con2 = self.getAnotherConnection()
|
|
except NotImplementedError:
|
|
return # should be "SKIP" for ACCESS
|
|
assert not con2.autocommit, 'default should be manual commit'
|
|
crsr = con2.cursor()
|
|
self.helpCreateAndPopulateTableTemp(crsr)
|
|
|
|
crsr.execute("INSERT INTO xx_%s (fldData) VALUES(100)" % config.tmp)
|
|
con2.commit()
|
|
|
|
selectSql = "SELECT fldData FROM xx_%s WHERE fldData=100" % config.tmp
|
|
crsr.execute(selectSql)
|
|
rs = crsr.fetchall()
|
|
assert len(rs) == 1
|
|
crsr.close()
|
|
con2.close()
|
|
conn = self.getConnection()
|
|
crsr = self.getCursor()
|
|
with conn.cursor() as crsr:
|
|
crsr.execute(selectSql)
|
|
rs = crsr.fetchall()
|
|
assert len(rs) == 1
|
|
assert rs[0][0] == 100
|
|
self.helpRollbackTblTemp()
|
|
|
|
|
|
def testAutoRollback(self):
|
|
try:
|
|
con2 = self.getAnotherConnection()
|
|
except NotImplementedError:
|
|
return # should be "SKIP" for ACCESS
|
|
assert not con2.autocommit, 'unexpected beginning condition'
|
|
crsr = con2.cursor()
|
|
self.helpCreateAndPopulateTableTemp(crsr)
|
|
crsr.execute("INSERT INTO xx_%s (fldData) VALUES(100)" % config.tmp)
|
|
selectSql = "SELECT fldData FROM xx_%s WHERE fldData=100" % config.tmp
|
|
crsr.execute(selectSql)
|
|
rs = crsr.fetchall()
|
|
assert len(rs) == 1
|
|
crsr.close()
|
|
con2.close()
|
|
crsr = self.getCursor()
|
|
try:
|
|
crsr.execute(selectSql) # closing the connection should have forced rollback
|
|
row = crsr.fetchone()
|
|
except api.DatabaseError:
|
|
row = None # if the entire table disappeared the rollback was perfect and the test passed
|
|
assert row == None, 'cursor.fetchone should return None if a query retrieves no rows. Got %s' % repr(row)
|
|
self.helpRollbackTblTemp()
|
|
|
|
def testAutoCommit(self):
|
|
try:
|
|
ac_conn = self.getAnotherConnection({'autocommit': True})
|
|
except NotImplementedError:
|
|
return # should be "SKIP" for ACCESS
|
|
crsr = ac_conn.cursor()
|
|
self.helpCreateAndPopulateTableTemp(crsr)
|
|
crsr.execute("INSERT INTO xx_%s (fldData) VALUES(100)" % config.tmp)
|
|
crsr.close()
|
|
with self.getCursor() as crsr:
|
|
selectSql = 'SELECT fldData from xx_%s' % config.tmp
|
|
crsr.execute(selectSql) # closing the connection should _not_ have forced rollback
|
|
rs = crsr.fetchall()
|
|
assert len(rs) == 10, 'all records should still be present'
|
|
ac_conn.close()
|
|
self.helpRollbackTblTemp()
|
|
|
|
def testSwitchedAutoCommit(self):
|
|
try:
|
|
ac_conn = self.getAnotherConnection()
|
|
except NotImplementedError:
|
|
return # should be "SKIP" for ACCESS
|
|
ac_conn.autocommit = True
|
|
crsr = ac_conn.cursor()
|
|
self.helpCreateAndPopulateTableTemp(crsr)
|
|
crsr.execute("INSERT INTO xx_%s (fldData) VALUES(100)" % config.tmp)
|
|
crsr.close()
|
|
conn = self.getConnection()
|
|
ac_conn.close()
|
|
with self.getCursor() as crsr:
|
|
selectSql = 'SELECT fldData from xx_%s' % config.tmp
|
|
crsr.execute(selectSql) # closing the connection should _not_ have forced rollback
|
|
rs = crsr.fetchall()
|
|
assert len(rs) == 10, 'all records should still be present'
|
|
self.helpRollbackTblTemp()
|
|
|
|
|
|
def testExtendedTypeHandling(self):
|
|
class XtendString(str):
|
|
pass
|
|
class XtendInt(int):
|
|
pass
|
|
class XtendFloat(float):
|
|
pass
|
|
xs = XtendString(randomstring(30))
|
|
xi = XtendInt(random.randint(-100, 500))
|
|
xf = XtendFloat(random.random())
|
|
self.helpForceDropOnTblTemp()
|
|
conn = self.getConnection()
|
|
crsr = conn.cursor()
|
|
tabdef = """
|
|
CREATE TABLE xx_%s (
|
|
s VARCHAR(40) NOT NULL,
|
|
i INTEGER NOT NULL,
|
|
f REAL NOT NULL)""" % config.tmp
|
|
crsr.execute(tabdef)
|
|
crsr.execute("INSERT INTO xx_%s (s, i, f) VALUES (?, ?, ?)" % config.tmp, (xs, xi, xf))
|
|
crsr.close()
|
|
conn = self.getConnection()
|
|
with self.getCursor() as crsr:
|
|
selectSql = 'SELECT s, i, f from xx_%s' % config.tmp
|
|
crsr.execute(selectSql) # closing the connection should _not_ have forced rollback
|
|
row = crsr.fetchone()
|
|
self.assertEqual(row.s, xs)
|
|
self.assertEqual(row.i, xi)
|
|
self.assertAlmostEqual(row.f, xf)
|
|
self.helpRollbackTblTemp()
|
|
|
|
|
|
class TestADOwithSQLServer(CommonDBTests):
|
|
def setUp(self):
|
|
self.conn = config.dbSqlServerconnect(*config.connStrSQLServer[0], **config.connStrSQLServer[1])
|
|
self.conn.timeout = 30 # turn timeout back up
|
|
self.engine = 'MSSQL'
|
|
self.db = config.dbSqlServerconnect
|
|
self.remote = config.connStrSQLServer[2]
|
|
|
|
def tearDown(self):
|
|
try:
|
|
self.conn.rollback()
|
|
except:
|
|
pass
|
|
try:
|
|
self.conn.close()
|
|
except:
|
|
pass
|
|
self.conn=None
|
|
|
|
def getConnection(self):
|
|
return self.conn
|
|
|
|
def getAnotherConnection(self, addkeys=None):
|
|
keys = dict(config.connStrSQLServer[1])
|
|
if addkeys:
|
|
keys.update(addkeys)
|
|
return config.dbSqlServerconnect(*config.connStrSQLServer[0], **keys)
|
|
|
|
def testVariableReturningStoredProcedure(self):
|
|
crsr=self.conn.cursor()
|
|
spdef= """
|
|
CREATE PROCEDURE sp_DeleteMeOnlyForTesting
|
|
@theInput varchar(50),
|
|
@theOtherInput varchar(50),
|
|
@theOutput varchar(100) OUTPUT
|
|
AS
|
|
SET @theOutput=@theInput+@theOtherInput
|
|
"""
|
|
try:
|
|
crsr.execute("DROP PROCEDURE sp_DeleteMeOnlyForTesting")
|
|
self.conn.commit()
|
|
except: #Make sure it is empty
|
|
pass
|
|
crsr.execute(spdef)
|
|
|
|
retvalues=crsr.callproc('sp_DeleteMeOnlyForTesting',('Dodsworth','Anne',' '))
|
|
assert retvalues[0]=='Dodsworth', '%s is not "Dodsworth"'%repr(retvalues[0])
|
|
assert retvalues[1]=='Anne','%s is not "Anne"'%repr(retvalues[1])
|
|
assert retvalues[2]=='DodsworthAnne','%s is not "DodsworthAnne"'%repr(retvalues[2])
|
|
self.conn.rollback()
|
|
|
|
def testMultipleSetReturn(self):
|
|
crsr=self.getCursor()
|
|
self.helpCreateAndPopulateTableTemp(crsr)
|
|
|
|
spdef= """
|
|
CREATE PROCEDURE sp_DeleteMe_OnlyForTesting
|
|
AS
|
|
SELECT fldData FROM xx_%s ORDER BY fldData ASC
|
|
SELECT fldData From xx_%s where fldData = -9999
|
|
SELECT fldData FROM xx_%s ORDER BY fldData DESC
|
|
""" % (config.tmp, config.tmp, config.tmp)
|
|
try:
|
|
crsr.execute("DROP PROCEDURE sp_DeleteMe_OnlyForTesting")
|
|
self.conn.commit()
|
|
except: #Make sure it is empty
|
|
pass
|
|
crsr.execute(spdef)
|
|
|
|
retvalues=crsr.callproc('sp_DeleteMe_OnlyForTesting')
|
|
row=crsr.fetchone()
|
|
self.assertEqual(row[0], 0)
|
|
assert crsr.nextset() == True, 'Operation should succeed'
|
|
assert not crsr.fetchall(), 'Should be an empty second set'
|
|
assert crsr.nextset() == True, 'third set should be present'
|
|
rowdesc=crsr.fetchall()
|
|
self.assertEqual(rowdesc[0][0],8)
|
|
assert crsr.nextset() == None,'No more return sets, should return None'
|
|
|
|
self.helpRollbackTblTemp()
|
|
|
|
|
|
def testDatetimeProcedureParameter(self):
|
|
crsr=self.conn.cursor()
|
|
spdef= """
|
|
CREATE PROCEDURE sp_DeleteMeOnlyForTesting
|
|
@theInput DATETIME,
|
|
@theOtherInput varchar(50),
|
|
@theOutput varchar(100) OUTPUT
|
|
AS
|
|
SET @theOutput = CONVERT(CHARACTER(20), @theInput, 0) + @theOtherInput
|
|
"""
|
|
try:
|
|
crsr.execute("DROP PROCEDURE sp_DeleteMeOnlyForTesting")
|
|
self.conn.commit()
|
|
except: #Make sure it is empty
|
|
pass
|
|
crsr.execute(spdef)
|
|
|
|
result = crsr.callproc('sp_DeleteMeOnlyForTesting', [adodbapi.Timestamp(2014,12,25,0,1,0), 'Beep', ' ' * 30])
|
|
|
|
assert result[2] == 'Dec 25 2014 12:01AM Beep', 'value was="%s"' % result[2]
|
|
self.conn.rollback()
|
|
|
|
def testIncorrectStoredProcedureParameter(self):
|
|
crsr=self.conn.cursor()
|
|
spdef= """
|
|
CREATE PROCEDURE sp_DeleteMeOnlyForTesting
|
|
@theInput DATETIME,
|
|
@theOtherInput varchar(50),
|
|
@theOutput varchar(100) OUTPUT
|
|
AS
|
|
SET @theOutput = CONVERT(CHARACTER(20), @theInput) + @theOtherInput
|
|
"""
|
|
try:
|
|
crsr.execute("DROP PROCEDURE sp_DeleteMeOnlyForTesting")
|
|
self.conn.commit()
|
|
except: #Make sure it is empty
|
|
pass
|
|
crsr.execute(spdef)
|
|
|
|
# calling the sproc with a string for the first parameter where a DateTime is expected
|
|
result = tryconnection.try_operation_with_expected_exception(
|
|
(api.DataError,api.DatabaseError),
|
|
crsr.callproc,
|
|
['sp_DeleteMeOnlyForTesting'],
|
|
{'parameters': ['this is wrong', 'Anne', 'not Alice']}
|
|
)
|
|
if result[0]: # the expected exception was raised
|
|
assert '@theInput' in str(result[1]) or 'DatabaseError' in str(result), \
|
|
'Identifies the wrong erroneous parameter'
|
|
else:
|
|
assert result[0], result[1] # incorrect or no exception
|
|
self.conn.rollback()
|
|
|
|
class TestADOwithAccessDB(CommonDBTests):
|
|
def setUp(self):
|
|
self.conn = config.dbAccessconnect(*config.connStrAccess[0], **config.connStrAccess[1])
|
|
self.conn.timeout = 30 # turn timeout back up
|
|
self.engine = 'ACCESS'
|
|
self.db = config.dbAccessconnect
|
|
self.remote = config.connStrAccess[2]
|
|
|
|
def tearDown(self):
|
|
try:
|
|
self.conn.rollback()
|
|
except:
|
|
pass
|
|
try:
|
|
self.conn.close()
|
|
except:
|
|
pass
|
|
self.conn=None
|
|
|
|
def getConnection(self):
|
|
return self.conn
|
|
|
|
def getAnotherConnection(self, addkeys=None):
|
|
raise NotImplementedError('Jet cannot use a second connection to the database')
|
|
|
|
def testOkConnect(self):
|
|
c = self.db(*config.connStrAccess[0], **config.connStrAccess[1])
|
|
assert c != None
|
|
c.close()
|
|
|
|
class TestADOwithMySql(CommonDBTests):
|
|
def setUp(self):
|
|
self.conn = config.dbMySqlconnect(*config.connStrMySql[0], **config.connStrMySql[1])
|
|
self.conn.timeout = 30 # turn timeout back up
|
|
self.engine = 'MySQL'
|
|
self.db = config.dbMySqlconnect
|
|
self.remote = config.connStrMySql[2]
|
|
|
|
def tearDown(self):
|
|
try:
|
|
self.conn.rollback()
|
|
except:
|
|
pass
|
|
try:
|
|
self.conn.close()
|
|
except:
|
|
pass
|
|
self.conn=None
|
|
|
|
def getConnection(self):
|
|
return self.conn
|
|
|
|
def getAnotherConnection(self, addkeys=None):
|
|
keys = dict(config.connStrMySql[1])
|
|
if addkeys:
|
|
keys.update(addkeys)
|
|
return config.dbMySqlconnect(*config.connStrMySql[0], **keys)
|
|
|
|
def testOkConnect(self):
|
|
c = self.db(*config.connStrMySql[0], **config.connStrMySql[1])
|
|
assert c != None
|
|
|
|
# def testStoredProcedure(self):
|
|
# crsr=self.conn.cursor()
|
|
# try:
|
|
# crsr.execute("DROP PROCEDURE DeleteMeOnlyForTesting")
|
|
# self.conn.commit()
|
|
# except: #Make sure it is empty
|
|
# pass
|
|
# spdef= """
|
|
# DELIMITER $$
|
|
# CREATE PROCEDURE DeleteMeOnlyForTesting (onein CHAR(10), twoin CHAR(10), OUT theout CHAR(20))
|
|
# DETERMINISTIC
|
|
# BEGIN
|
|
# SET theout = onein //|| twoin;
|
|
# /* (SELECT 'a small string' as result; */
|
|
# END $$
|
|
# """
|
|
#
|
|
# crsr.execute(spdef)
|
|
#
|
|
# retvalues=crsr.callproc('DeleteMeOnlyForTesting',('Dodsworth','Anne',' '))
|
|
# print 'return value (mysql)=',repr(crsr.returnValue) ###
|
|
# assert retvalues[0]=='Dodsworth', '%s is not "Dodsworth"'%repr(retvalues[0])
|
|
# assert retvalues[1]=='Anne','%s is not "Anne"'%repr(retvalues[1])
|
|
# assert retvalues[2]=='DodsworthAnne','%s is not "DodsworthAnne"'%repr(retvalues[2])
|
|
#
|
|
# try:
|
|
# crsr.execute("DROP PROCEDURE, DeleteMeOnlyForTesting")
|
|
# self.conn.commit()
|
|
# except: #Make sure it is empty
|
|
# pass
|
|
|
|
class TestADOwithPostgres(CommonDBTests):
|
|
def setUp(self):
|
|
self.conn = config.dbPostgresConnect(*config.connStrPostgres[0], **config.connStrPostgres[1])
|
|
self.conn.timeout = 30 # turn timeout back up
|
|
self.engine = 'PostgreSQL'
|
|
self.db = config.dbPostgresConnect
|
|
self.remote = config.connStrPostgres[2]
|
|
|
|
def tearDown(self):
|
|
try:
|
|
self.conn.rollback()
|
|
except:
|
|
pass
|
|
try:
|
|
self.conn.close()
|
|
except:
|
|
pass
|
|
self.conn=None
|
|
|
|
def getConnection(self):
|
|
return self.conn
|
|
|
|
def getAnotherConnection(self, addkeys=None):
|
|
keys = dict(config.connStrPostgres[1])
|
|
if addkeys:
|
|
keys.update(addkeys)
|
|
return config.dbPostgresConnect(*config.connStrPostgres[0], **keys)
|
|
|
|
def testOkConnect(self):
|
|
c = self.db(*config.connStrPostgres[0], **config.connStrPostgres[1])
|
|
assert c != None
|
|
|
|
# def testStoredProcedure(self):
|
|
# crsr=self.conn.cursor()
|
|
# spdef= """
|
|
# CREATE OR REPLACE FUNCTION DeleteMeOnlyForTesting (text, text)
|
|
# RETURNS text AS $funk$
|
|
# BEGIN
|
|
# RETURN $1 || $2;
|
|
# END;
|
|
# $funk$
|
|
# LANGUAGE SQL;
|
|
# """
|
|
#
|
|
# crsr.execute(spdef)
|
|
# retvalues = crsr.callproc('DeleteMeOnlyForTesting',('Dodsworth','Anne',' '))
|
|
# ### print 'return value (pg)=',repr(crsr.returnValue) ###
|
|
# assert retvalues[0]=='Dodsworth', '%s is not "Dodsworth"'%repr(retvalues[0])
|
|
# assert retvalues[1]=='Anne','%s is not "Anne"'%repr(retvalues[1])
|
|
# assert retvalues[2]=='Dodsworth Anne','%s is not "Dodsworth Anne"'%repr(retvalues[2])
|
|
# self.conn.rollback()
|
|
# try:
|
|
# crsr.execute("DROP PROCEDURE, DeleteMeOnlyForTesting")
|
|
# self.conn.commit()
|
|
# except: #Make sure it is empty
|
|
# pass
|
|
|
|
class TimeConverterInterfaceTest(unittest.TestCase):
|
|
def testIDate(self):
|
|
assert self.tc.Date(1990,2,2)
|
|
|
|
def testITime(self):
|
|
assert self.tc.Time(13,2,2)
|
|
|
|
def testITimestamp(self):
|
|
assert self.tc.Timestamp(1990,2,2,13,2,1)
|
|
|
|
def testIDateObjectFromCOMDate(self):
|
|
assert self.tc.DateObjectFromCOMDate(37435.7604282)
|
|
|
|
def testICOMDate(self):
|
|
assert hasattr(self.tc,'COMDate')
|
|
|
|
def testExactDate(self):
|
|
d=self.tc.Date(1994,11,15)
|
|
comDate=self.tc.COMDate(d)
|
|
correct=34653.0
|
|
assert comDate == correct,comDate
|
|
|
|
def testExactTimestamp(self):
|
|
d=self.tc.Timestamp(1994,11,15,12,0,0)
|
|
comDate=self.tc.COMDate(d)
|
|
correct=34653.5
|
|
self.assertEqual( comDate ,correct)
|
|
|
|
d=self.tc.Timestamp(2003,5,6,14,15,17)
|
|
comDate=self.tc.COMDate(d)
|
|
correct=37747.593946759262
|
|
self.assertEqual( comDate ,correct)
|
|
|
|
def testIsoFormat(self):
|
|
d=self.tc.Timestamp(1994,11,15,12,3,10)
|
|
iso=self.tc.DateObjectToIsoFormatString(d)
|
|
self.assertEqual(str(iso[:19]) , '1994-11-15 12:03:10')
|
|
|
|
dt=self.tc.Date(2003,5,2)
|
|
iso=self.tc.DateObjectToIsoFormatString(dt)
|
|
self.assertEqual(str(iso[:10]), '2003-05-02')
|
|
|
|
if config.doMxDateTimeTest:
|
|
import mx.DateTime
|
|
class TestMXDateTimeConverter(TimeConverterInterfaceTest):
|
|
def setUp(self):
|
|
self.tc = api.mxDateTimeConverter()
|
|
|
|
def testCOMDate(self):
|
|
t=mx.DateTime.DateTime(2002,6,28,18,15,2)
|
|
cmd=self.tc.COMDate(t)
|
|
assert cmd == t.COMDate()
|
|
|
|
def testDateObjectFromCOMDate(self):
|
|
cmd=self.tc.DateObjectFromCOMDate(37435.7604282)
|
|
t=mx.DateTime.DateTime(2002,6,28,18,15,0)
|
|
t2=mx.DateTime.DateTime(2002,6,28,18,15,2)
|
|
assert t2>cmd>t
|
|
|
|
def testDate(self):
|
|
assert mx.DateTime.Date(1980,11,4)==self.tc.Date(1980,11,4)
|
|
|
|
def testTime(self):
|
|
assert mx.DateTime.Time(13,11,4)==self.tc.Time(13,11,4)
|
|
|
|
def testTimestamp(self):
|
|
t=mx.DateTime.DateTime(2002,6,28,18,15,1)
|
|
obj=self.tc.Timestamp(2002,6,28,18,15,1)
|
|
assert t == obj
|
|
|
|
import time
|
|
class TestPythonTimeConverter(TimeConverterInterfaceTest):
|
|
def setUp(self):
|
|
self.tc=api.pythonTimeConverter()
|
|
|
|
def testCOMDate(self):
|
|
mk = time.mktime((2002,6,28,18,15,1, 4,31+28+31+30+31+28,-1))
|
|
t=time.localtime(mk)
|
|
# Fri, 28 Jun 2002 18:15:01 +0000
|
|
cmd=self.tc.COMDate(t)
|
|
assert abs(cmd - 37435.7604282) < 1.0/24,"%f more than an hour wrong" % cmd
|
|
|
|
def testDateObjectFromCOMDate(self):
|
|
cmd=self.tc.DateObjectFromCOMDate(37435.7604282)
|
|
t1=time.gmtime(time.mktime((2002,6,28,0,14,1, 4,31+28+31+30+31+28,-1)))
|
|
#there are errors in the implementation of gmtime which we ignore
|
|
t2=time.gmtime(time.mktime((2002,6,29,12,14,2, 4,31+28+31+30+31+28,-1)))
|
|
assert t1<cmd<t2, '"%s" should be about 2002-6-28 12:15:01'%repr(cmd)
|
|
|
|
def testDate(self):
|
|
t1=time.mktime((2002,6,28,18,15,1, 4,31+28+31+30+31+30,0))
|
|
t2=time.mktime((2002,6,30,18,15,1, 4,31+28+31+30+31+28,0))
|
|
obj=self.tc.Date(2002,6,29)
|
|
assert t1< time.mktime(obj)<t2,obj
|
|
|
|
def testTime(self):
|
|
self.assertEqual( self.tc.Time(18,15,2),time.gmtime(18*60*60+15*60+2))
|
|
|
|
def testTimestamp(self):
|
|
t1=time.localtime(time.mktime((2002,6,28,18,14,1, 4,31+28+31+30+31+28,-1)))
|
|
t2=time.localtime(time.mktime((2002,6,28,18,16,1, 4,31+28+31+30+31+28,-1)))
|
|
obj=self.tc.Timestamp(2002,6,28,18,15,2)
|
|
assert t1< obj <t2,obj
|
|
|
|
class TestPythonDateTimeConverter(TimeConverterInterfaceTest):
|
|
def setUp(self):
|
|
self.tc = api.pythonDateTimeConverter()
|
|
|
|
def testCOMDate(self):
|
|
t=datetime.datetime( 2002,6,28,18,15,1)
|
|
# Fri, 28 Jun 2002 18:15:01 +0000
|
|
cmd=self.tc.COMDate(t)
|
|
assert abs(cmd - 37435.7604282) < 1.0/24,"more than an hour wrong"
|
|
|
|
def testDateObjectFromCOMDate(self):
|
|
cmd = self.tc.DateObjectFromCOMDate(37435.7604282)
|
|
t1 = datetime.datetime(2002,6,28,18,14,1)
|
|
t2 = datetime.datetime(2002,6,28,18,16,1)
|
|
assert t1 < cmd < t2, cmd
|
|
|
|
tx = datetime.datetime(2002,6,28,18,14,1,900000) # testing that microseconds don't become milliseconds
|
|
c1 = self.tc.DateObjectFromCOMDate(self.tc.COMDate(tx))
|
|
assert t1 < c1 < t2, c1
|
|
|
|
def testDate(self):
|
|
t1=datetime.date(2002,6,28)
|
|
t2=datetime.date(2002,6,30)
|
|
obj=self.tc.Date(2002,6,29)
|
|
assert t1< obj <t2,obj
|
|
|
|
def testTime(self):
|
|
self.assertEqual( self.tc.Time(18,15,2).isoformat()[:8],'18:15:02')
|
|
|
|
def testTimestamp(self):
|
|
t1=datetime.datetime(2002,6,28,18,14,1)
|
|
t2=datetime.datetime(2002,6,28,18,16,1)
|
|
obj=self.tc.Timestamp(2002,6,28,18,15,2)
|
|
assert t1< obj <t2,obj
|
|
|
|
suites=[]
|
|
suites.append( unittest.makeSuite(TestPythonDateTimeConverter,'test'))
|
|
if config.doMxDateTimeTest:
|
|
suites.append( unittest.makeSuite(TestMXDateTimeConverter,'test'))
|
|
if config.doTimeTest:
|
|
suites.append( unittest.makeSuite(TestPythonTimeConverter,'test'))
|
|
|
|
if config.doAccessTest:
|
|
suites.append( unittest.makeSuite(TestADOwithAccessDB,'test'))
|
|
if config.doSqlServerTest:
|
|
suites.append( unittest.makeSuite(TestADOwithSQLServer,'test'))
|
|
if config.doMySqlTest:
|
|
suites.append( unittest.makeSuite(TestADOwithMySql,'test'))
|
|
if config.doPostgresTest:
|
|
suites.append( unittest.makeSuite(TestADOwithPostgres,'test'))
|
|
|
|
class cleanup_manager(object):
|
|
def __enter__(self):
|
|
pass
|
|
def __exit__(self, exc_type, exc_val, exc_tb):
|
|
config.cleanup(config.testfolder, config.mdb_name)
|
|
|
|
suite=unittest.TestSuite(suites)
|
|
if __name__ == '__main__':
|
|
mysuite = copy.deepcopy(suite)
|
|
with cleanup_manager():
|
|
defaultDateConverter = adodbapi.dateconverter
|
|
print(__doc__)
|
|
print("Default Date Converter is %s" %(defaultDateConverter,))
|
|
dateconverter = defaultDateConverter
|
|
tag = 'datetime'
|
|
unittest.TextTestRunner().run(mysuite)
|
|
|
|
if config.iterateOverTimeTests:
|
|
for test, dateconverter, tag in (
|
|
(config.doTimeTest,api.pythonTimeConverter, 'pythontime'),
|
|
(config.doMxDateTimeTest, api.mxDateTimeConverter, 'mx')):
|
|
if test:
|
|
mysuite = copy.deepcopy(suite) # work around a side effect of unittest.TextTestRunner
|
|
adodbapi.adodbapi.dateconverter = dateconverter()
|
|
print("Changed dateconverter to ")
|
|
print(adodbapi.adodbapi.dateconverter)
|
|
unittest.TextTestRunner().run(mysuite)
|