test: add tests for close pq wrappers

diff --git a/tests/pq/test_async.py b/tests/pq/test_async.py
index 2c3de9814c6b2c7b89a760086b3a110762b0692f..c47729c5a85cec04f01fd5a8cc2d316d14ead17a 100644 (file)
--- a/tests/pq/test_async.py
+++ b/tests/pq/test_async.py
@@ -189,6 +189,22 @@ def test_send_describe_prepared(pgconn):
         pgconn.send_describe_prepared(b"prep")
 
 
+@pytest.mark.libpq(">= 17")
+def test_send_close_prepared(pgconn):
+    pgconn.send_prepare(b"prep", b"select $1::int8 + $2::int8 as fld")
+    (res,) = execute_wait(pgconn)
+    assert res.status == pq.ExecStatus.COMMAND_OK, res.error_message
+
+    pgconn.send_close_prepared(b"prep")
+    (res,) = execute_wait(pgconn)
+    assert res.status == pq.ExecStatus.COMMAND_OK, res.error_message
+
+    # Because we closed it, describing should not work
+    pgconn.send_describe_prepared(b"prep")
+    (res,) = execute_wait(pgconn)
+    assert res.status == pq.ExecStatus.FATAL_ERROR
+
+
 @pytest.mark.crdb_skip("server-side cursor")
 def test_send_describe_portal(pgconn):
     res = pgconn.exec_(
@@ -208,3 +224,24 @@ def test_send_describe_portal(pgconn):
     pgconn.finish()
     with pytest.raises(psycopg.OperationalError):
         pgconn.send_describe_portal(b"cur")
+
+
+@pytest.mark.crdb_skip("server-side cursor")
+@pytest.mark.libpq(">= 17")
+def test_send_close_portal(pgconn):
+    res = pgconn.exec_(
+        b"""
+        begin;
+        declare cur cursor for select * from generate_series(1,10) foo;
+        """
+    )
+    assert res.status == pq.ExecStatus.COMMAND_OK, res.error_message
+
+    pgconn.send_close_portal(b"cur")
+    (res,) = execute_wait(pgconn)
+    assert res.status == pq.ExecStatus.COMMAND_OK, res.error_message
+
+    # Because we closed it, describing should not work
+    pgconn.send_describe_portal(b"cur")
+    (res,) = execute_wait(pgconn)
+    assert res.status == pq.ExecStatus.FATAL_ERROR

diff --git a/tests/pq/test_exec.py b/tests/pq/test_exec.py
index 86c30c099e905f2fb360d112446000ab74e28311..24bbe7087ede8fd1d812449fbd70f975c7c8b34c 100644 (file)
--- a/tests/pq/test_exec.py
+++ b/tests/pq/test_exec.py
@@ -126,6 +126,19 @@ def test_exec_prepared_binary_out(pgconn, fmt, out):
     assert res.get_value(0, 0) == out
 
 
+@pytest.mark.libpq(">= 17")
+def test_close_prepared(pgconn):
+    res = pgconn.prepare(b"prep", b"select $1::int + $2::int")
+    assert res.status == pq.ExecStatus.COMMAND_OK, res.error_message
+
+    res = pgconn.close_prepared(b"prep")
+    assert res.status == pq.ExecStatus.COMMAND_OK, res.error_message
+
+    # Because we closed it, executing should not work
+    res = pgconn.exec_prepared(b"prep", [b"3", b"5"])
+    assert res.status == pq.ExecStatus.FATAL_ERROR
+
+
 @pytest.mark.crdb_skip("server-side cursor")
 def test_describe_portal(pgconn):
     res = pgconn.exec_(
@@ -144,3 +157,22 @@ def test_describe_portal(pgconn):
     pgconn.finish()
     with pytest.raises(psycopg.OperationalError):
         pgconn.describe_portal(b"cur")
+
+
+@pytest.mark.crdb_skip("server-side cursor")
+@pytest.mark.libpq(">= 17")
+def test_close_portal(pgconn):
+    res = pgconn.exec_(
+        b"""
+        begin;
+        declare cur cursor for select * from generate_series(1,10) foo;
+        """
+    )
+    assert res.status == pq.ExecStatus.COMMAND_OK, res.error_message
+
+    res = pgconn.close_portal(b"cur")
+    assert res.status == pq.ExecStatus.COMMAND_OK, res.error_message
+
+    # Because we closed it, describing should not work
+    res = pgconn.describe_portal(b"cur")
+    assert res.status == pq.ExecStatus.FATAL_ERROR