(self)
| 104 | self.cursor.execute("drop table %s" % (self.table)) |
| 105 | |
| 106 | def test_transactions(self): |
| 107 | columndefs = ("col1 INT", "col2 VARCHAR(255)") |
| 108 | |
| 109 | def generator(row, col): |
| 110 | if col == 0: |
| 111 | return row |
| 112 | else: |
| 113 | return ("%i" % (row % 10)) * 255 |
| 114 | |
| 115 | self.create_table(columndefs) |
| 116 | insert_statement = "INSERT INTO {} VALUES ({})".format( |
| 117 | self.table, |
| 118 | ",".join(["%s"] * len(columndefs)), |
| 119 | ) |
| 120 | data = [ |
| 121 | [generator(i, j) for j in range(len(columndefs))] for i in range(self.rows) |
| 122 | ] |
| 123 | self.cursor.executemany(insert_statement, data) |
| 124 | # verify |
| 125 | self.connection.commit() |
| 126 | self.cursor.execute("select * from %s" % self.table) |
| 127 | res = self.cursor.fetchall() |
| 128 | self.assertEqual(len(res), self.rows) |
| 129 | for i in range(self.rows): |
| 130 | for j in range(len(columndefs)): |
| 131 | self.assertEqual(res[i][j], generator(i, j)) |
| 132 | delete_statement = "delete from %s where col1=%%s" % self.table |
| 133 | self.cursor.execute(delete_statement, (0,)) |
| 134 | self.cursor.execute(f"select col1 from {self.table} where col1=%s", (0,)) |
| 135 | res = self.cursor.fetchall() |
| 136 | self.assertFalse(res, "DELETE didn't work") |
| 137 | self.connection.rollback() |
| 138 | self.cursor.execute(f"select col1 from {self.table} where col1=%s", (0,)) |
| 139 | res = self.cursor.fetchall() |
| 140 | self.assertTrue(len(res) == 1, "ROLLBACK didn't work") |
| 141 | self.cursor.execute("drop table %s" % (self.table)) |
| 142 | |
| 143 | def test_truncation(self): |
| 144 | columndefs = ("col1 INT", "col2 VARCHAR(255)") |
nothing calls this directly
no test coverage detected