| 76 | assert len(result.fetchall()) == 0 |
| 77 | |
| 78 | def test_raise(self, local_connection): |
| 79 | connection = local_connection |
| 80 | users = self.tables.users |
| 81 | |
| 82 | transaction = connection.begin() |
| 83 | try: |
| 84 | connection.execute( |
| 85 | users.insert(), dict(user_id=1, user_name="user1") |
| 86 | ) |
| 87 | connection.execute( |
| 88 | users.insert(), dict(user_id=2, user_name="user2") |
| 89 | ) |
| 90 | connection.execute( |
| 91 | users.insert(), dict(user_id=1, user_name="user3") |
| 92 | ) |
| 93 | transaction.commit() |
| 94 | assert False |
| 95 | except Exception as e: |
| 96 | print("Exception: ", e) |
| 97 | transaction.rollback() |
| 98 | |
| 99 | result = connection.exec_driver_sql("select * from users") |
| 100 | assert len(result.fetchall()) == 0 |
| 101 | |
| 102 | def test_rollback_end_ctx_manager_autocommit(self, local_connection): |
| 103 | m1 = mock.Mock() |