| 403 | |
| 404 | |
| 405 | def test_auto_transaction_fail(conn): |
| 406 | conn.pgconn.exec_(b"drop table if exists foo") |
| 407 | conn.pgconn.exec_(b"create table foo (id int primary key)") |
| 408 | |
| 409 | cur = conn.cursor() |
| 410 | assert conn.pgconn.transaction_status == pq.TransactionStatus.IDLE |
| 411 | |
| 412 | cur.execute("insert into foo values (1)") |
| 413 | assert conn.pgconn.transaction_status == pq.TransactionStatus.INTRANS |
| 414 | |
| 415 | with pytest.raises(psycopg.DatabaseError): |
| 416 | cur.execute("meh") |
| 417 | assert conn.pgconn.transaction_status == pq.TransactionStatus.INERROR |
| 418 | |
| 419 | with pytest.raises(psycopg.errors.InFailedSqlTransaction): |
| 420 | cur.execute("select 1") |
| 421 | |
| 422 | conn.commit() |
| 423 | assert conn.pgconn.transaction_status == pq.TransactionStatus.IDLE |
| 424 | cur.execute("select * from foo") |
| 425 | assert cur.fetchone() is None |
| 426 | assert conn.pgconn.transaction_status == pq.TransactionStatus.INTRANS |
| 427 | |
| 428 | |
| 429 | def test_autocommit(conn): |