(self, aconn_cls, aconn, dsn, tpc)
| 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") |
| 63 | assert aconn.info.transaction_status == TransactionStatus.IDLE |
| 64 | |
| 65 | await aconn.tpc_begin(xid) |
| 66 | assert aconn.info.transaction_status == TransactionStatus.INTRANS |
| 67 | |
| 68 | cur = aconn.cursor() |
| 69 | await cur.execute("insert into test_tpc values ('test_tpc_commit_rec')") |
| 70 | assert tpc.count_xacts() == 0 |
| 71 | assert tpc.count_test_records() == 0 |
| 72 | |
| 73 | await aconn.tpc_prepare() |
| 74 | await aconn.close() |
| 75 | assert tpc.count_xacts() == 1 |
| 76 | assert tpc.count_test_records() == 0 |
| 77 | |
| 78 | async with await aconn_cls.connect(dsn) as aconn: |
| 79 | xid = aconn.xid(1, "gtrid", "bqual") |
| 80 | await aconn.tpc_commit(xid) |
| 81 | assert aconn.info.transaction_status == TransactionStatus.IDLE |
| 82 | |
| 83 | assert tpc.count_xacts() == 0 |
| 84 | assert tpc.count_test_records() == 1 |
| 85 | |
| 86 | async def test_tpc_rollback(self, aconn, tpc): |
| 87 | xid = aconn.xid(1, "gtrid", "bqual") |
nothing calls this directly
no test coverage detected