(self, aconn, tpc)
| 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") |
| 110 | assert aconn.info.transaction_status == TransactionStatus.IDLE |
| 111 | |
| 112 | await aconn.tpc_begin(xid) |
| 113 | assert aconn.info.transaction_status == TransactionStatus.INTRANS |
| 114 | |
| 115 | cur = aconn.cursor() |
| 116 | await cur.execute("insert into test_tpc values ('test_tpc_rollback_1p')") |
| 117 | assert tpc.count_xacts() == 0 |
| 118 | assert tpc.count_test_records() == 0 |
| 119 | |
| 120 | await aconn.tpc_rollback() |
| 121 | assert aconn.info.transaction_status == TransactionStatus.IDLE |
| 122 | assert tpc.count_xacts() == 0 |
| 123 | assert tpc.count_test_records() == 0 |
| 124 | |
| 125 | async def test_tpc_rollback_recovered(self, aconn_cls, aconn, dsn, tpc): |
| 126 | xid = aconn.xid(1, "gtrid", "bqual") |
nothing calls this directly
no test coverage detected