Add tests to verify broken queries or other statements in server-side cursors

Enforce passing a single statement by making sure to use the advanced
query protocol.

diff --git a/psycopg3/psycopg3/server_cursor.py b/psycopg3/psycopg3/server_cursor.py
index 058a0392141fedf620bc3adba89b7d4dd9ba9d52..cadd2ca7de4afc922ff41795f7ac026bc82edd8a 100644 (file)
--- a/psycopg3/psycopg3/server_cursor.py
+++ b/psycopg3/psycopg3/server_cursor.py
@@ -61,7 +61,7 @@ class ServerCursorHelper(Generic[ConnectionType]):
 
         yield from cur._start_query(query)
         pgq = cur._convert_query(query, params)
-        cur._execute_send(pgq)
+        cur._execute_send(pgq, no_pqexec=True)
         results = yield from execute(conn.pgconn)
         cur._execute_results(results)
 

diff --git a/tests/test_server_cursor.py b/tests/test_server_cursor.py
index 551579e95ffedebc58a0dd62ab5d814a4f98190a..29cb2921cd443e8c0a2bed552f7ee3e78cba54e1 100644 (file)
--- a/tests/test_server_cursor.py
+++ b/tests/test_server_cursor.py
@@ -108,6 +108,15 @@ def test_execute_reuse(conn):
         assert cur.description[1].name == "baz"
 
 
+@pytest.mark.parametrize(
+    "stmt", ["", "wat", "create table ssc ()", "select 1; select 2"]
+)
+def test_execute_error(conn, stmt):
+    cur = conn.cursor("foo")
+    with pytest.raises(e.ProgrammingError):
+        cur.execute(stmt)
+
+
 def test_executemany(conn):
     cur = conn.cursor("foo")
     with pytest.raises(e.NotSupportedError):

diff --git a/tests/test_server_cursor_async.py b/tests/test_server_cursor_async.py
index fa2d9425cdcc4f67f35c9630550a7e480d3165d6..d7d4805ac62dafafc28de3bfcd2f9cbf989cec71 100644 (file)
--- a/tests/test_server_cursor_async.py
+++ b/tests/test_server_cursor_async.py
@@ -110,6 +110,15 @@ async def test_execute_reuse(aconn):
         assert cur.description[1].name == "baz"
 
 
+@pytest.mark.parametrize(
+    "stmt", ["", "wat", "create table ssc ()", "select 1; select 2"]
+)
+async def test_execute_error(aconn, stmt):
+    cur = aconn.cursor("foo")
+    with pytest.raises(e.ProgrammingError):
+        await cur.execute(stmt)
+
+
 async def test_executemany(aconn):
     cur = aconn.cursor("foo")
     with pytest.raises(e.NotSupportedError):