(self, aconn, tpc)
| 20 | |
| 21 | class TestTPC: |
| 22 | async def test_tpc_commit(self, aconn, tpc): |
| 23 | xid = aconn.xid(1, "gtrid", "bqual") |
| 24 | assert aconn.info.transaction_status == TransactionStatus.IDLE |
| 25 | |
| 26 | await aconn.tpc_begin(xid) |
| 27 | assert aconn.info.transaction_status == TransactionStatus.INTRANS |
| 28 | |
| 29 | cur = aconn.cursor() |
| 30 | await cur.execute("insert into test_tpc values ('test_tpc_commit')") |
| 31 | assert tpc.count_xacts() == 0 |
| 32 | assert tpc.count_test_records() == 0 |
| 33 | |
| 34 | await aconn.tpc_prepare() |
| 35 | assert aconn.info.transaction_status == TransactionStatus.IDLE |
| 36 | assert tpc.count_xacts() == 1 |
| 37 | assert tpc.count_test_records() == 0 |
| 38 | |
| 39 | await aconn.tpc_commit() |
| 40 | assert aconn.info.transaction_status == TransactionStatus.IDLE |
| 41 | assert tpc.count_xacts() == 0 |
| 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") |
nothing calls this directly
no test coverage detected