test: add test to verify the behaviour of failed commits

The exception is raised where expected; the transaction is left in IDLE
state. This is unlike what happens in pipeline mode.

diff --git a/tests/test_connection.py b/tests/test_connection.py
index 800c5a11a1eac06876d2d735ed85f0b861adc51d..c9eb3ba660139bd1993b775a264d8c89fb1a27aa 100644 (file)
--- a/tests/test_connection.py
+++ b/tests/test_connection.py
@@ -11,9 +11,8 @@ else:
     from typing_extensions import TypedDict
 
 import psycopg
-from psycopg import Connection, Notify
+from psycopg import Connection, Notify, errors as e
 from psycopg.rows import tuple_row
-from psycopg.errors import UndefinedTable
 from psycopg.conninfo import conninfo_to_dict, make_conninfo
 
 from .utils import gc_collect
@@ -151,7 +150,7 @@ def test_context_rollback(conn, dsn):
 
     with psycopg.connect(dsn) as conn:
         with conn.cursor() as cur:
-            with pytest.raises(UndefinedTable):
+            with pytest.raises(e.UndefinedTable):
                 cur.execute("select * from textctx")
 
 
@@ -221,6 +220,25 @@ def test_commit(conn):
         conn.commit()
 
 
+def test_commit_error(conn):
+    conn.execute(
+        """
+        drop table if exists selfref;
+        create table selfref (
+            x serial primary key,
+            y int references selfref (x) deferrable initially deferred)
+        """
+    )
+    conn.commit()
+
+    conn.execute("insert into selfref (y) values (-1)")
+    with pytest.raises(e.ForeignKeyViolation):
+        conn.commit()
+    assert conn.pgconn.transaction_status == conn.TransactionStatus.IDLE
+    cur = conn.execute("select 1")
+    assert cur.fetchone() == (1,)
+
+
 def test_rollback(conn):
     conn.pgconn.exec_(b"drop table if exists foo")
     conn.pgconn.exec_(b"create table foo (id int primary key)")

diff --git a/tests/test_connection_async.py b/tests/test_connection_async.py
index dd10ed6fc63ed30857625d44788b88b98dc3ad8f..16502a51cfe18087f6da7f27a590f64596e1fb9a 100644 (file)
--- a/tests/test_connection_async.py
+++ b/tests/test_connection_async.py
@@ -4,9 +4,8 @@ import logging
 import weakref
 
 import psycopg
-from psycopg import AsyncConnection, Notify
+from psycopg import AsyncConnection, Notify, errors as e
 from psycopg.rows import tuple_row
-from psycopg.errors import UndefinedTable
 from psycopg.conninfo import conninfo_to_dict, make_conninfo
 
 from .utils import gc_collect
@@ -152,7 +151,7 @@ async def test_context_rollback(aconn, dsn):
 
     async with await psycopg.AsyncConnection.connect(dsn) as aconn:
         async with aconn.cursor() as cur:
-            with pytest.raises(UndefinedTable):
+            with pytest.raises(e.UndefinedTable):
                 await cur.execute("select * from textctx")
 
 
@@ -220,6 +219,25 @@ async def test_commit(aconn):
         await aconn.commit()
 
 
+async def test_commit_error(aconn):
+    await aconn.execute(
+        """
+        drop table if exists selfref;
+        create table selfref (
+            x serial primary key,
+            y int references selfref (x) deferrable initially deferred)
+        """
+    )
+    await aconn.commit()
+
+    await aconn.execute("insert into selfref (y) values (-1)")
+    with pytest.raises(e.ForeignKeyViolation):
+        await aconn.commit()
+    assert aconn.pgconn.transaction_status == aconn.TransactionStatus.IDLE
+    cur = await aconn.execute("select 1")
+    assert await cur.fetchone() == (1,)
+
+
 async def test_rollback(aconn):
     aconn.pgconn.exec_(b"drop table if exists foo")
     aconn.pgconn.exec_(b"create table foo (id int primary key)")