test_adodbapi_dbapi20.py 5.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186
  1. print("This module depends on the dbapi20 compliance tests created by Stuart Bishop")
  2. print("(see db-sig mailing list history for info)")
  3. import platform
  4. import unittest
  5. import sys
  6. import dbapi20
  7. import setuptestframework
  8. testfolder = setuptestframework.maketemp()
  9. if '--package' in sys.argv:
  10. pth = setuptestframework.makeadopackage(testfolder)
  11. sys.argv.remove('--package')
  12. else:
  13. pth = setuptestframework.find_ado_path()
  14. if pth not in sys.path:
  15. sys.path.insert(1,pth)
  16. # function to clean up the temporary folder -- calling program must run this function before exit.
  17. cleanup = setuptestframework.getcleanupfunction()
  18. import adodbapi
  19. import adodbapi.is64bit as is64bit
  20. db = adodbapi
  21. if '--verbose' in sys.argv:
  22. db.adodbapi.verbose = 3
  23. print(adodbapi.version)
  24. print("Tested with dbapi20 %s" % dbapi20.__version__)
  25. try:
  26. onWindows = bool(sys.getwindowsversion()) # seems to work on all versions of Python
  27. except:
  28. onWindows = False
  29. node = platform.node()
  30. conn_kws = {}
  31. host = 'testsql.2txt.us,1430' # if None, will use macro to fill in node name
  32. instance = r'%s\SQLEXPRESS'
  33. conn_kws['name'] = 'adotest'
  34. conn_kws['user'] = 'adotestuser' # None implies Windows security
  35. conn_kws['password'] = "Sq1234567"
  36. # macro definition for keyword "security" using macro "auto_security"
  37. conn_kws['macro_auto_security'] = 'security'
  38. if host is None:
  39. conn_kws['macro_getnode'] = ['host', instance]
  40. else:
  41. conn_kws['host'] = host
  42. conn_kws['provider'] = 'Provider=MSOLEDBSQL;DataTypeCompatibility=80;MARS Connection=True;'
  43. connStr = "%(provider)s; %(security)s; Initial Catalog=%(name)s;Data Source=%(host)s"
  44. if onWindows and node != "z-PC":
  45. pass # default should make a local SQL Server connection
  46. elif node == "xxx": # try Postgres database
  47. _computername = "25.223.161.222"
  48. _databasename='adotest'
  49. _username = 'adotestuser'
  50. _password = '12345678'
  51. _driver="PostgreSQL Unicode"
  52. _provider = ''
  53. connStr = '%sDriver={%s};Server=%s;Database=%s;uid=%s;pwd=%s;' % \
  54. (_provider,_driver,_computername,_databasename,_username,_password)
  55. elif node == "yyy": # ACCESS data base is known to fail some tests.
  56. if is64bit.Python():
  57. driver = "Microsoft.ACE.OLEDB.12.0"
  58. else:
  59. driver = "Microsoft.Jet.OLEDB.4.0"
  60. testmdb = setuptestframework.makemdb(testfolder)
  61. connStr = r"Provider=%s;Data Source=%s" % (driver, testmdb)
  62. else: # try a remote connection to an SQL server
  63. conn_kws['proxy_host'] = '25.44.77.176'
  64. import adodbapi.remote
  65. db = adodbapi.remote
  66. print('Using Connection String like=%s' % connStr)
  67. print('Keywords=%s' % repr(conn_kws))
  68. class test_adodbapi(dbapi20.DatabaseAPI20Test):
  69. driver = db
  70. connect_args = (connStr,)
  71. connect_kw_args = conn_kws
  72. def __init__(self,arg):
  73. dbapi20.DatabaseAPI20Test.__init__(self,arg)
  74. def testMethodName(self):
  75. return self.id().split('.')[-1]
  76. def setUp(self):
  77. # Call superclass setUp In case this does something in the
  78. # future
  79. dbapi20.DatabaseAPI20Test.setUp(self)
  80. if self.testMethodName()=='test_callproc':
  81. con = self._connect()
  82. engine = con.dbms_name
  83. ## print('Using database Engine=%s' % engine) ##
  84. if engine != 'MS Jet':
  85. sql="""
  86. create procedure templower
  87. @theData varchar(50)
  88. as
  89. select lower(@theData)
  90. """
  91. else: # Jet
  92. sql="""
  93. create procedure templower
  94. (theData varchar(50))
  95. as
  96. select lower(theData);
  97. """
  98. cur = con.cursor()
  99. try:
  100. cur.execute(sql)
  101. con.commit()
  102. except:
  103. pass
  104. cur.close()
  105. con.close()
  106. self.lower_func='templower'
  107. def tearDown(self):
  108. if self.testMethodName()=='test_callproc':
  109. con = self._connect()
  110. cur = con.cursor()
  111. try:
  112. cur.execute("drop procedure templower")
  113. except:
  114. pass
  115. con.commit()
  116. dbapi20.DatabaseAPI20Test.tearDown(self)
  117. def help_nextset_setUp(self,cur):
  118. 'Should create a procedure called deleteme '
  119. 'that returns two result sets, first the number of rows in booze then "name from booze"'
  120. sql="""
  121. create procedure deleteme as
  122. begin
  123. select count(*) from %sbooze
  124. select name from %sbooze
  125. end
  126. """ %(self.table_prefix,self.table_prefix)
  127. cur.execute(sql)
  128. def help_nextset_tearDown(self,cur):
  129. 'If cleaning up is needed after nextSetTest'
  130. try:
  131. cur.execute("drop procedure deleteme")
  132. except:
  133. pass
  134. def test_nextset(self):
  135. con = self._connect()
  136. try:
  137. cur = con.cursor()
  138. stmts=[self.ddl1] + self._populate()
  139. for sql in stmts:
  140. cur.execute(sql)
  141. self.help_nextset_setUp(cur)
  142. cur.callproc('deleteme')
  143. numberofrows=cur.fetchone()
  144. assert numberofrows[0]== 6
  145. assert cur.nextset()
  146. names=cur.fetchall()
  147. assert len(names) == len(self.samples)
  148. s=cur.nextset()
  149. assert s == None,'No more return sets, should return None'
  150. finally:
  151. try:
  152. self.help_nextset_tearDown(cur)
  153. finally:
  154. con.close()
  155. def test_setoutputsize(self): pass
  156. if __name__ == '__main__':
  157. unittest.main()
  158. cleanup(testfolder, None)