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