(self, aconn, tpc)
| 42 | assert tpc.count_test_records() == 1 |
| 43 | |
| 44 | async def test_tpc_commit_one_phase(self, aconn, tpc): |
| 45 | xid = aconn.xid(1, "gtrid", "bqual") |
| 46 | assert aconn.info.transaction_status == TransactionStatus.IDLE |
| 47 | |
| 48 | await aconn.tpc_begin(xid) |
| 49 | assert aconn.info.transaction_status == TransactionStatus.INTRANS |
| 50 | |
| 51 | cur = aconn.cursor() |
| 52 | await cur.execute("insert into test_tpc values ('test_tpc_commit_1p')") |
| 53 | assert tpc.count_xacts() == 0 |
| 54 | assert tpc.count_test_records() == 0 |
| 55 | |
| 56 | await aconn.tpc_commit() |
| 57 | assert aconn.info.transaction_status == TransactionStatus.IDLE |
| 58 | assert tpc.count_xacts() == 0 |
| 59 | assert tpc.count_test_records() == 1 |
| 60 | |
| 61 | async def test_tpc_commit_recovered(self, aconn_cls, aconn, dsn, tpc): |
| 62 | xid = aconn.xid(1, "gtrid", "bqual") |
nothing calls this directly
no test coverage detected