(self, aconn, tpc)
| 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 |
| 152 | await aconn.tpc_recover() |
| 153 | assert aconn.info.transaction_status == TransactionStatus.IDLE |
| 154 | |
| 155 | cur = aconn.cursor() |
| 156 | await cur.execute("select 1") |
| 157 | assert aconn.info.transaction_status == TransactionStatus.INTRANS |
| 158 | await aconn.tpc_recover() |
| 159 | assert aconn.info.transaction_status == TransactionStatus.INTRANS |
| 160 | |
| 161 | async def test_recovered_xids(self, aconn, tpc): |
| 162 | # insert a few test xns |
nothing calls this directly
no test coverage detected