(self, conn, tpc)
| 152 | assert tpc.count_test_records() == 0 |
| 153 | |
| 154 | def test_status_after_recover(self, conn, tpc): |
| 155 | assert conn.info.transaction_status == TransactionStatus.IDLE |
| 156 | conn.tpc_recover() |
| 157 | assert conn.info.transaction_status == TransactionStatus.IDLE |
| 158 | |
| 159 | cur = conn.cursor() |
| 160 | cur.execute("select 1") |
| 161 | assert conn.info.transaction_status == TransactionStatus.INTRANS |
| 162 | conn.tpc_recover() |
| 163 | assert conn.info.transaction_status == TransactionStatus.INTRANS |
| 164 | |
| 165 | def test_recovered_xids(self, conn, tpc): |
| 166 | # insert a few test xns |
nothing calls this directly
no test coverage detected