(self, conn, tpc)
| 24 | class TestTPC: |
| 25 | |
| 26 | def test_tpc_commit(self, conn, tpc): |
| 27 | xid = conn.xid(1, "gtrid", "bqual") |
| 28 | assert conn.info.transaction_status == TransactionStatus.IDLE |
| 29 | |
| 30 | conn.tpc_begin(xid) |
| 31 | assert conn.info.transaction_status == TransactionStatus.INTRANS |
| 32 | |
| 33 | cur = conn.cursor() |
| 34 | cur.execute("insert into test_tpc values ('test_tpc_commit')") |
| 35 | assert tpc.count_xacts() == 0 |
| 36 | assert tpc.count_test_records() == 0 |
| 37 | |
| 38 | conn.tpc_prepare() |
| 39 | assert conn.info.transaction_status == TransactionStatus.IDLE |
| 40 | assert tpc.count_xacts() == 1 |
| 41 | assert tpc.count_test_records() == 0 |
| 42 | |
| 43 | conn.tpc_commit() |
| 44 | assert conn.info.transaction_status == TransactionStatus.IDLE |
| 45 | assert tpc.count_xacts() == 0 |
| 46 | assert tpc.count_test_records() == 1 |
| 47 | |
| 48 | def test_tpc_commit_one_phase(self, conn, tpc): |
| 49 | xid = conn.xid(1, "gtrid", "bqual") |
nothing calls this directly
no test coverage detected