(self, aconn_cls, aconn, dsn, tpc)
| 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") |
| 127 | assert aconn.info.transaction_status == TransactionStatus.IDLE |
| 128 | |
| 129 | await aconn.tpc_begin(xid) |
| 130 | assert aconn.info.transaction_status == TransactionStatus.INTRANS |
| 131 | |
| 132 | cur = aconn.cursor() |
| 133 | await cur.execute("insert into test_tpc values ('test_tpc_commit_rec')") |
| 134 | assert tpc.count_xacts() == 0 |
| 135 | assert tpc.count_test_records() == 0 |
| 136 | |
| 137 | await aconn.tpc_prepare() |
| 138 | await aconn.close() |
| 139 | assert tpc.count_xacts() == 1 |
| 140 | assert tpc.count_test_records() == 0 |
| 141 | |
| 142 | async with await aconn_cls.connect(dsn) as aconn: |
| 143 | xid = aconn.xid(1, "gtrid", "bqual") |
| 144 | await aconn.tpc_rollback(xid) |
| 145 | assert aconn.info.transaction_status == TransactionStatus.IDLE |
| 146 | |
| 147 | assert tpc.count_xacts() == 0 |
| 148 | assert tpc.count_test_records() == 0 |
| 149 | |
| 150 | async def test_status_after_recover(self, aconn, tpc): |
| 151 | assert aconn.info.transaction_status == TransactionStatus.IDLE |
nothing calls this directly
no test coverage detected