(self, aconn, tpc)
| 84 | assert tpc.count_test_records() == 1 |
| 85 | |
| 86 | async def test_tpc_rollback(self, aconn, tpc): |
| 87 | xid = aconn.xid(1, "gtrid", "bqual") |
| 88 | assert aconn.info.transaction_status == TransactionStatus.IDLE |
| 89 | |
| 90 | await aconn.tpc_begin(xid) |
| 91 | assert aconn.info.transaction_status == TransactionStatus.INTRANS |
| 92 | |
| 93 | cur = aconn.cursor() |
| 94 | await cur.execute("insert into test_tpc values ('test_tpc_rollback')") |
| 95 | assert tpc.count_xacts() == 0 |
| 96 | assert tpc.count_test_records() == 0 |
| 97 | |
| 98 | await aconn.tpc_prepare() |
| 99 | assert aconn.info.transaction_status == TransactionStatus.IDLE |
| 100 | assert tpc.count_xacts() == 1 |
| 101 | assert tpc.count_test_records() == 0 |
| 102 | |
| 103 | await aconn.tpc_rollback() |
| 104 | assert aconn.info.transaction_status == TransactionStatus.IDLE |
| 105 | assert tpc.count_xacts() == 0 |
| 106 | assert tpc.count_test_records() == 0 |
| 107 | |
| 108 | async def test_tpc_rollback_one_phase(self, aconn, tpc): |
| 109 | xid = aconn.xid(1, "gtrid", "bqual") |
nothing calls this directly
no test coverage detected