(conn, tpc)
| 12 | |
| 13 | |
| 14 | def test_tpc_on_pg_connection(conn, tpc): |
| 15 | xid = conn.xid(1, "gtrid", "bqual") |
| 16 | assert conn.info.transaction_status == TransactionStatus.IDLE |
| 17 | |
| 18 | conn.tpc_begin(xid) |
| 19 | assert conn.info.transaction_status == TransactionStatus.INTRANS |
| 20 | |
| 21 | cur = conn.cursor() |
| 22 | cur.execute("insert into test_tpc values ('test_tpc_commit')") |
| 23 | assert tpc.count_xacts() == 0 |
| 24 | assert tpc.count_test_records() == 0 |
| 25 | |
| 26 | conn.tpc_prepare() |
| 27 | assert conn.info.transaction_status == TransactionStatus.IDLE |
| 28 | assert tpc.count_xacts() == 1 |
| 29 | assert tpc.count_test_records() == 0 |
| 30 | |
| 31 | conn.tpc_commit() |
| 32 | assert conn.info.transaction_status == TransactionStatus.IDLE |
| 33 | assert tpc.count_xacts() == 0 |
| 34 | assert tpc.count_test_records() == 1 |
nothing calls this directly
no test coverage detected