| 79 | self.create_table_extra)) |
| 80 | |
| 81 | def check_data_integrity(self, columndefs, generator): |
| 82 | # insert |
| 83 | self.create_table(columndefs) |
| 84 | insert_statement = ('INSERT INTO %s VALUES (%s)' % |
| 85 | (self.table, |
| 86 | ','.join(['{!s}'] * len(columndefs)))) |
| 87 | data = [ [ generator(i,j) for j in range(len(columndefs)) ] |
| 88 | for i in range(self.rows) ] |
| 89 | self.cursor.executemany(insert_statement, data) |
| 90 | self.connection.commit() |
| 91 | # verify |
| 92 | self.cursor.execute('select * from %s' % self.table) |
| 93 | l = self.cursor.fetchall() |
| 94 | self.assertEqual(len(l), self.rows) |
| 95 | try: |
| 96 | for i in range(self.rows): |
| 97 | for j in range(len(columndefs)): |
| 98 | self.assertEqual(l[i][j], generator(i,j)) |
| 99 | finally: |
| 100 | if not self.debug: |
| 101 | self.cursor.execute('drop table %s' % (self.table)) |
| 102 | |
| 103 | def test_transactions(self): |
| 104 | columndefs = ( 'col1 INT', 'col2 VARCHAR(255)') |