You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
ORPA-pyOpenRPA/Resources/WPy64-3720/python-3.7.2.amd64/Lib/site-packages/adodbapi/test/test_adodbapi_dbapi20.py

182 lines
5.5 KiB

print("This module depends on the dbapi20 compliance tests created by Stuart Bishop")
print("(see db-sig mailing list history for info)")
import platform
import unittest
import sys
import dbapi20
import setuptestframework
testfolder = setuptestframework.maketemp()
if '--package' in sys.argv:
pth = setuptestframework.makeadopackage(testfolder)
sys.argv.remove('--package')
else:
pth = setuptestframework.find_ado_path()
if pth not in sys.path:
sys.path.insert(1,pth)
# function to clean up the temporary folder -- calling program must run this function before exit.
cleanup = setuptestframework.getcleanupfunction()
import adodbapi
import adodbapi.is64bit as is64bit
db = adodbapi
if '--verbose' in sys.argv:
db.adodbapi.verbose = 3
print((adodbapi.version))
print(("Tested with dbapi20 %s" % dbapi20.__version__))
try:
onWindows = bool(sys.getwindowsversion()) # seems to work on all versions of Python
except:
onWindows = False
node = platform.node()
conn_kws = {}
host = None # if None, will use macro to fill in node name
instance = r'%s\SQLExpress'
conn_kws['name'] = 'adotest'
if host is None:
conn_kws['macro_getnode'] = ['host', instance]
else:
conn_kws['host'] = host
conn_kws['provider'] = 'Provider=SQLNCLI11;DataTypeCompatibility=80;MARS Connection=True;'
connStr = "%(provider)s; Integrated Security=SSPI; Initial Catalog=%(name)s;Data Source=%(host)s"
if onWindows and node != "z-PC":
pass # default should make a local SQL Server connection
elif node == "xxx": # try Postgres database
_computername = "25.223.161.222"
_databasename='adotest'
_username = 'adotestuser'
_password = '12345678'
_driver="PostgreSQL Unicode"
_provider = ''
connStr = '%sDriver={%s};Server=%s;Database=%s;uid=%s;pwd=%s;' % \
(_provider,_driver,_computername,_databasename,_username,_password)
elif node == "yyy": # ACCESS data base is known to fail some tests.
if is64bit.Python():
driver = "Microsoft.ACE.OLEDB.12.0"
else:
driver = "Microsoft.Jet.OLEDB.4.0"
testmdb = setuptestframework.makemdb(testfolder)
connStr = r"Provider=%s;Data Source=%s" % (driver, testmdb)
else: # try a remote connection to an SQL server
conn_kws['proxy_host'] = '25.44.77.176'
import adodbapi.remote
db = adodbapi.remote
print(('Using Connection String like=%s' % connStr))
print(('Keywords=%s' % repr(conn_kws)))
class test_adodbapi(dbapi20.DatabaseAPI20Test):
driver = db
connect_args = (connStr,)
connect_kw_args = conn_kws
def __init__(self,arg):
dbapi20.DatabaseAPI20Test.__init__(self,arg)
def testMethodName(self):
return self.id().split('.')[-1]
def setUp(self):
# Call superclass setUp In case this does something in the
# future
dbapi20.DatabaseAPI20Test.setUp(self)
if self.testMethodName()=='test_callproc':
con = self._connect()
engine = con.dbms_name
## print('Using database Engine=%s' % engine) ##
if engine != 'MS Jet':
sql="""
create procedure templower
@theData varchar(50)
as
select lower(@theData)
"""
else: # Jet
sql="""
create procedure templower
(theData varchar(50))
as
select lower(theData);
"""
cur = con.cursor()
try:
cur.execute(sql)
con.commit()
except:
pass
cur.close()
con.close()
self.lower_func='templower'
def tearDown(self):
if self.testMethodName()=='test_callproc':
con = self._connect()
cur = con.cursor()
try:
cur.execute("drop procedure templower")
except:
pass
con.commit()
dbapi20.DatabaseAPI20Test.tearDown(self)
def help_nextset_setUp(self,cur):
'Should create a procedure called deleteme '
'that returns two result sets, first the number of rows in booze then "name from booze"'
sql="""
create procedure deleteme as
begin
select count(*) from %sbooze
select name from %sbooze
end
""" %(self.table_prefix,self.table_prefix)
cur.execute(sql)
def help_nextset_tearDown(self,cur):
'If cleaning up is needed after nextSetTest'
try:
cur.execute("drop procedure deleteme")
except:
pass
def test_nextset(self):
con = self._connect()
try:
cur = con.cursor()
stmts=[self.ddl1] + self._populate()
for sql in stmts:
cur.execute(sql)
self.help_nextset_setUp(cur)
cur.callproc('deleteme')
numberofrows=cur.fetchone()
assert numberofrows[0]== 6
assert cur.nextset()
names=cur.fetchall()
assert len(names) == len(self.samples)
s=cur.nextset()
assert s == None,'No more return sets, should return None'
finally:
try:
self.help_nextset_tearDown(cur)
finally:
con.close()
def test_setoutputsize(self): pass
if __name__ == '__main__':
unittest.main()
cleanup(testfolder, None)