capabilities.py 9.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281
  1. #!/usr/bin/env python -O
  2. """ Script to test database capabilities and the DB-API interface
  3. for functionality and memory leaks.
  4. Adapted from a script by M-A Lemburg.
  5. """
  6. from time import time
  7. import array
  8. import unittest
  9. from configdb import connection_factory
  10. class DatabaseTest(unittest.TestCase):
  11. db_module = None
  12. connect_args = ()
  13. connect_kwargs = dict()
  14. create_table_extra = ''
  15. rows = 10
  16. debug = False
  17. def setUp(self):
  18. import gc
  19. db = connection_factory(**self.connect_kwargs)
  20. self.connection = db
  21. self.cursor = db.cursor()
  22. # TODO: this needs to be re-evaluated for Python 3
  23. self.BLOBText = ''.join([chr(i) for i in range(256)] * 100);
  24. self.BLOBUText = u''.join([unichr(i) for i in range(16384)])
  25. self.BLOBBinary = self.db_module.Binary(''.join([chr(i) for i in range(256)] * 16))
  26. leak_test = True
  27. def tearDown(self):
  28. if self.leak_test:
  29. import gc
  30. del self.cursor
  31. orphans = gc.collect()
  32. self.failIf(orphans, "%d orphaned objects found after deleting cursor" % orphans)
  33. del self.connection
  34. orphans = gc.collect()
  35. self.failIf(orphans, "%d orphaned objects found after deleting connection" % orphans)
  36. def table_exists(self, name):
  37. try:
  38. self.cursor.execute('select * from %s where 1=0' % name)
  39. except:
  40. return False
  41. else:
  42. return True
  43. def quote_identifier(self, ident):
  44. return '"%s"' % ident
  45. def new_table_name(self):
  46. i = id(self.cursor)
  47. while True:
  48. name = self.quote_identifier('tb%08x' % i)
  49. if not self.table_exists(name):
  50. return name
  51. i = i + 1
  52. def create_table(self, columndefs):
  53. """ Create a table using a list of column definitions given in
  54. columndefs.
  55. generator must be a function taking arguments (row_number,
  56. col_number) returning a suitable data object for insertion
  57. into the table.
  58. """
  59. self.table = self.new_table_name()
  60. self.cursor.execute('CREATE TABLE %s (%s) %s' %
  61. (self.table,
  62. ',\n'.join(columndefs),
  63. self.create_table_extra))
  64. def check_data_integrity(self, columndefs, generator):
  65. # insert
  66. self.create_table(columndefs)
  67. insert_statement = ('INSERT INTO %s VALUES (%s)' %
  68. (self.table,
  69. ','.join(['%s'] * len(columndefs))))
  70. data = [ [ generator(i,j) for j in range(len(columndefs)) ]
  71. for i in range(self.rows) ]
  72. self.cursor.executemany(insert_statement, data)
  73. self.connection.commit()
  74. # verify
  75. self.cursor.execute('select * from %s' % self.table)
  76. l = self.cursor.fetchall()
  77. self.assertEquals(len(l), self.rows)
  78. try:
  79. for i in range(self.rows):
  80. for j in range(len(columndefs)):
  81. self.assertEquals(l[i][j], generator(i,j))
  82. finally:
  83. if not self.debug:
  84. self.cursor.execute('drop table %s' % (self.table))
  85. def test_transactions(self):
  86. columndefs = ( 'col1 INT', 'col2 VARCHAR(255)')
  87. def generator(row, col):
  88. if col == 0: return row
  89. else: return ('%i' % (row%10))*255
  90. self.create_table(columndefs)
  91. insert_statement = ('INSERT INTO %s VALUES (%s)' %
  92. (self.table,
  93. ','.join(['%s'] * len(columndefs))))
  94. data = [ [ generator(i,j) for j in range(len(columndefs)) ]
  95. for i in range(self.rows) ]
  96. self.cursor.executemany(insert_statement, data)
  97. # verify
  98. self.connection.commit()
  99. self.cursor.execute('select * from %s' % self.table)
  100. l = self.cursor.fetchall()
  101. self.assertEquals(len(l), self.rows)
  102. for i in range(self.rows):
  103. for j in range(len(columndefs)):
  104. self.assertEquals(l[i][j], generator(i,j))
  105. delete_statement = 'delete from %s where col1=%%s' % self.table
  106. self.cursor.execute(delete_statement, (0,))
  107. self.cursor.execute('select col1 from %s where col1=%s' % \
  108. (self.table, 0))
  109. l = self.cursor.fetchall()
  110. self.assertFalse(l, "DELETE didn't work")
  111. self.connection.rollback()
  112. self.cursor.execute('select col1 from %s where col1=%s' % \
  113. (self.table, 0))
  114. l = self.cursor.fetchall()
  115. self.assertTrue(len(l) == 1, "ROLLBACK didn't work")
  116. self.cursor.execute('drop table %s' % (self.table))
  117. def test_truncation(self):
  118. columndefs = ( 'col1 INT', 'col2 VARCHAR(255)')
  119. def generator(row, col):
  120. if col == 0: return row
  121. else: return ('%i' % (row%10))*((255-self.rows/2)+row)
  122. self.create_table(columndefs)
  123. insert_statement = ('INSERT INTO %s VALUES (%s)' %
  124. (self.table,
  125. ','.join(['%s'] * len(columndefs))))
  126. try:
  127. self.cursor.execute(insert_statement, (0, '0'*256))
  128. except self.connection.DataError:
  129. pass
  130. else:
  131. self.fail("Over-long column did not generate warnings/exception with single insert")
  132. self.connection.rollback()
  133. try:
  134. for i in range(self.rows):
  135. data = []
  136. for j in range(len(columndefs)):
  137. data.append(generator(i,j))
  138. self.cursor.execute(insert_statement,tuple(data))
  139. except self.connection.DataError:
  140. pass
  141. else:
  142. self.fail("Over-long columns did not generate warnings/exception with execute()")
  143. self.connection.rollback()
  144. try:
  145. data = [ [ generator(i,j) for j in range(len(columndefs)) ]
  146. for i in range(self.rows) ]
  147. self.cursor.executemany(insert_statement, data)
  148. except self.connection.DataError:
  149. pass
  150. else:
  151. self.fail("Over-long columns did not generate warnings/exception with executemany()")
  152. self.connection.rollback()
  153. self.cursor.execute('drop table %s' % (self.table))
  154. def test_CHAR(self):
  155. # Character data
  156. def generator(row,col):
  157. return ('%i' % ((row+col) % 10)) * 255
  158. self.check_data_integrity(
  159. ('col1 char(255)','col2 char(255)'),
  160. generator)
  161. def test_INT(self):
  162. # Number data
  163. def generator(row,col):
  164. return row*row
  165. self.check_data_integrity(
  166. ('col1 INT',),
  167. generator)
  168. def test_DECIMAL(self):
  169. # DECIMAL
  170. def generator(row,col):
  171. from decimal import Decimal
  172. return Decimal("%d.%02d" % (row, col))
  173. self.check_data_integrity(
  174. ('col1 DECIMAL(5,2)',),
  175. generator)
  176. def test_DATE(self):
  177. ticks = time()
  178. def generator(row,col):
  179. return self.db_module.DateFromTicks(ticks+row*86400-col*1313)
  180. self.check_data_integrity(
  181. ('col1 DATE',),
  182. generator)
  183. def test_TIME(self):
  184. ticks = time()
  185. def generator(row,col):
  186. return self.db_module.TimeFromTicks(ticks+row*86400-col*1313)
  187. self.check_data_integrity(
  188. ('col1 TIME',),
  189. generator)
  190. def test_DATETIME(self):
  191. ticks = time()
  192. def generator(row,col):
  193. return self.db_module.TimestampFromTicks(ticks+row*86400-col*1313)
  194. self.check_data_integrity(
  195. ('col1 DATETIME',),
  196. generator)
  197. def test_TIMESTAMP(self):
  198. ticks = time()
  199. def generator(row,col):
  200. return self.db_module.TimestampFromTicks(ticks+row*86400-col*1313)
  201. self.check_data_integrity(
  202. ('col1 TIMESTAMP',),
  203. generator)
  204. def test_fractional_TIMESTAMP(self):
  205. ticks = time()
  206. def generator(row,col):
  207. return self.db_module.TimestampFromTicks(ticks+row*86400-col*1313+row*0.7*col/3.0)
  208. self.check_data_integrity(
  209. ('col1 TIMESTAMP',),
  210. generator)
  211. def test_LONG(self):
  212. def generator(row,col):
  213. if col == 0:
  214. return row
  215. else:
  216. return self.BLOBUText # 'BLOB Text ' * 1024
  217. self.check_data_integrity(
  218. ('col1 INT','col2 LONG'),
  219. generator)
  220. def test_TEXT(self):
  221. def generator(row,col):
  222. return self.BLOBUText # 'BLOB Text ' * 1024
  223. self.check_data_integrity(
  224. ('col2 TEXT',),
  225. generator)
  226. def test_LONG_BYTE(self):
  227. def generator(row,col):
  228. if col == 0:
  229. return row
  230. else:
  231. return self.BLOBBinary # 'BLOB\000Binary ' * 1024
  232. self.check_data_integrity(
  233. ('col1 INT','col2 LONG BYTE'),
  234. generator)
  235. def test_BLOB(self):
  236. def generator(row,col):
  237. if col == 0:
  238. return row
  239. else:
  240. return self.BLOBBinary # 'BLOB\000Binary ' * 1024
  241. self.check_data_integrity(
  242. ('col1 INT','col2 BLOB'),
  243. generator)