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