| 366 | |
| 367 | |
| 368 | async def test_rollback(aconn): |
| 369 | aconn.pgconn.exec_(b"drop table if exists foo") |
| 370 | aconn.pgconn.exec_(b"create table foo (id int primary key)") |
| 371 | aconn.pgconn.exec_(b"begin") |
| 372 | assert aconn.pgconn.transaction_status == pq.TransactionStatus.INTRANS |
| 373 | aconn.pgconn.exec_(b"insert into foo values (1)") |
| 374 | await aconn.rollback() |
| 375 | assert aconn.pgconn.transaction_status == pq.TransactionStatus.IDLE |
| 376 | res = aconn.pgconn.exec_(b"select id from foo where id = 1") |
| 377 | assert res.ntuples == 0 |
| 378 | |
| 379 | await aconn.close() |
| 380 | with pytest.raises(psycopg.OperationalError): |
| 381 | await aconn.rollback() |
| 382 | |
| 383 | |
| 384 | async def test_auto_transaction(aconn): |