From: Daniele Varrazzo <daniele.varrazzo@gmail.com>
Date: Fri, 24 Oct 2025 15:09:15 +0000 (+0100)
Subject: test(copy): cover the Python implementation in copy pinning tests
X-Git-Url: http://git.ipfire.org/?a=commitdiff_plain;h=c30c3cb98771c7ce3db55fff42500dd2a6be583b;p=thirdparty%2Fpsycopg.git

test(copy): cover the Python implementation in copy pinning tests
---

diff --git a/tests/test_copy.py b/tests/test_copy.py
index 4627fa23a..18975c99a 100644
--- a/tests/test_copy.py
+++ b/tests/test_copy.py
@@ -12,8 +12,9 @@ import pytest
 import psycopg
 from psycopg import errors as e
 from psycopg import pq, sql
+from psycopg.abc import Buffer
 from psycopg.copy import Copy, LibpqWriter, QueuedLibpqWriter
-from psycopg.adapt import PyFormat
+from psycopg.adapt import Dumper, PyFormat
 from psycopg.types import TypeInfo
 from psycopg.types.hstore import register_hstore
 from psycopg.types.numeric import Int4
@@ -487,8 +488,19 @@ def test_copy_in_records_binary(conn, format):
     assert data == [(1, None, "hello"), (2, None, "world")]
 
 
+class StrictIntDumper(Dumper):
+    oid = psycopg.adapters.types["int4"].oid
+
+    def dump(self, obj: int) -> Buffer:
+        if type(obj) is not int:
+            raise TypeError(f"bad type: {obj!r}")
+        return str(obj).encode()
+
+
 def test_copy_in_text_no_pinning(conn):
     cur = conn.cursor()
+    cur.adapters.register_dumper(int, StrictIntDumper)
+
     cols = [
         "col1 serial primary key",
         "col2 int",
@@ -513,14 +525,9 @@ def test_copy_in_text_no_pinning(conn):
 
 
 def test_copy_in_text_pinned(conn):
-    # FIXME: this test works currently only in c,
-    # as c/python dumpers differ in what they accept as valid input
-    # here: python int & float text dumpers always allow str as input
-    from psycopg._cmodule import _psycopg
-
-    if not _psycopg:
-        return
     cur = conn.cursor()
+    cur.adapters.register_dumper(int, StrictIntDumper)
+
     cols = [
         "col1 serial primary key",
         "col2 int",
@@ -543,8 +550,6 @@ def test_copy_in_text_pinned(conn):
             copy.write_row([1.0, 2, 3, 4.1])
         with pytest.raises((e.DataError, TypeError)):
             copy.write_row([1, "2", 3, 4.1])
-        with pytest.raises((e.DataError, TypeError)):
-            copy.write_row([1, 2, 3, "4.1"])
 
     cur.execute("select col2,col3,col4,col5 from copy_in order by 1")
     data = cur.fetchall()
diff --git a/tests/test_copy_async.py b/tests/test_copy_async.py
index cbf38843f..df3a7e6b7 100644
--- a/tests/test_copy_async.py
+++ b/tests/test_copy_async.py
@@ -9,8 +9,9 @@ import pytest
 import psycopg
 from psycopg import errors as e
 from psycopg import pq, sql
+from psycopg.abc import Buffer
 from psycopg.copy import AsyncCopy, AsyncLibpqWriter, AsyncQueuedLibpqWriter
-from psycopg.adapt import PyFormat
+from psycopg.adapt import Dumper, PyFormat
 from psycopg.types import TypeInfo
 from psycopg.types.hstore import register_hstore
 from psycopg.types.numeric import Int4
@@ -501,8 +502,19 @@ async def test_copy_in_records_binary(aconn, format):
     assert data == [(1, None, "hello"), (2, None, "world")]
 
 
+class StrictIntDumper(Dumper):
+    oid = psycopg.adapters.types["int4"].oid
+
+    def dump(self, obj: int) -> Buffer:
+        if type(obj) is not int:
+            raise TypeError(f"bad type: {obj!r}")
+        return str(obj).encode()
+
+
 async def test_copy_in_text_no_pinning(aconn):
     cur = aconn.cursor()
+    cur.adapters.register_dumper(int, StrictIntDumper)
+
     cols = [
         "col1 serial primary key",
         "col2 int",
@@ -527,14 +539,9 @@ async def test_copy_in_text_no_pinning(aconn):
 
 
 async def test_copy_in_text_pinned(aconn):
-    # FIXME: this test works currently only in c,
-    # as c/python dumpers differ in what they accept as valid input
-    # here: python int & float text dumpers always allow str as input
-    from psycopg._cmodule import _psycopg
-
-    if not _psycopg:
-        return
     cur = aconn.cursor()
+    cur.adapters.register_dumper(int, StrictIntDumper)
+
     cols = [
         "col1 serial primary key",
         "col2 int",
@@ -557,8 +564,6 @@ async def test_copy_in_text_pinned(aconn):
             await copy.write_row([1.0, 2, 3, 4.1])
         with pytest.raises((e.DataError, TypeError)):
             await copy.write_row([1, "2", 3, 4.1])
-        with pytest.raises((e.DataError, TypeError)):
-            await copy.write_row([1, 2, 3, "4.1"])
 
     await cur.execute("select col2,col3,col4,col5 from copy_in order by 1")
     data = await cur.fetchall()