Add test to verify errors in row factories are handled ok

diff --git a/tests/test_cursor.py b/tests/test_cursor.py
index b88080292ec059a9435b52fad315fd41248ac09f..e1faa6a564ddec67bb7d1152fa92c14328340bc6 100644 (file)
--- a/tests/test_cursor.py
+++ b/tests/test_cursor.py
@@ -304,6 +304,26 @@ def test_row_factory(conn):
     assert cur.fetchone() == {"y": "y", "z": "z"}
 
 
+def test_bad_row_factory(conn):
+    def broken_factory(cur):
+        1 / 0
+
+    cur = conn.cursor(row_factory=broken_factory)
+    with pytest.raises(ZeroDivisionError):
+        cur.execute("select 1")
+
+    def broken_maker(cur):
+        def make_row(seq):
+            1 / 0
+
+        return make_row
+
+    cur = conn.cursor(row_factory=broken_maker)
+    cur.execute("select 1")
+    with pytest.raises(ZeroDivisionError):
+        cur.fetchone()
+
+
 def test_scroll(conn):
     cur = conn.cursor()
     with pytest.raises(psycopg.ProgrammingError):

diff --git a/tests/test_cursor_async.py b/tests/test_cursor_async.py
index 1e0ec046d90096671949c41c9cd7547e0f060682..b29070f02280f43309a05a6071e0c48d4d4ec81a 100644 (file)
--- a/tests/test_cursor_async.py
+++ b/tests/test_cursor_async.py
@@ -310,6 +310,26 @@ async def test_row_factory(aconn):
     assert await cur.fetchone() == {"y": "y", "z": "z"}
 
 
+async def test_bad_row_factory(aconn):
+    def broken_factory(cur):
+        1 / 0
+
+    cur = aconn.cursor(row_factory=broken_factory)
+    with pytest.raises(ZeroDivisionError):
+        await cur.execute("select 1")
+
+    def broken_maker(cur):
+        def make_row(seq):
+            1 / 0
+
+        return make_row
+
+    cur = aconn.cursor(row_factory=broken_maker)
+    await cur.execute("select 1")
+    with pytest.raises(ZeroDivisionError):
+        await cur.fetchone()
+
+
 async def test_scroll(aconn):
     cur = aconn.cursor()
     with pytest.raises(psycopg.ProgrammingError):