move BYTEA test to PG suite

we can't have any postgresql imports in suite/test_insert.py

Change-Id: I81f364169c775bef71dc77db0a6e9d6bc530e79d

diff --git a/lib/sqlalchemy/testing/suite/test_insert.py b/lib/sqlalchemy/testing/suite/test_insert.py
index 0faf591da4990a246d900aa1910c220b51cd48aa..09f24d356dae3c7a9c9460d160f0f92519572369 100644 (file)
--- a/lib/sqlalchemy/testing/suite/test_insert.py
+++ b/lib/sqlalchemy/testing/suite/test_insert.py
@@ -18,7 +18,6 @@ from ... import literal_column
 from ... import Numeric
 from ... import select
 from ... import String
-from ...dialects.postgresql import BYTEA
 from ...types import LargeBinary
 from ...types import UUID
 from ...types import Uuid
@@ -553,7 +552,6 @@ class ReturningTest(fixtures.TablesTest):
             b"this is binary",
         ),
         ("LargeBinary2", LargeBinary(), b"7\xe7\x9f"),
-        ("PG BYTEA", BYTEA(), b"7\xe7\x9f", testing.only_on("postgresql")),
         argnames="type_,value",
         id_="iaa",
     )

diff --git a/test/dialect/postgresql/test_types.py b/test/dialect/postgresql/test_types.py
index 73a66fff612befb567f61bfc66cb813f91dc9823..54a7c69cab6f9d16344f62e2de870a16f977122e 100644 (file)
--- a/test/dialect/postgresql/test_types.py
+++ b/test/dialect/postgresql/test_types.py
@@ -46,6 +46,7 @@ from sqlalchemy.dialects.postgresql import array
 from sqlalchemy.dialects.postgresql import array_agg
 from sqlalchemy.dialects.postgresql import asyncpg
 from sqlalchemy.dialects.postgresql import base
+from sqlalchemy.dialects.postgresql import BYTEA
 from sqlalchemy.dialects.postgresql import CITEXT
 from sqlalchemy.dialects.postgresql import DATEMULTIRANGE
 from sqlalchemy.dialects.postgresql import DATERANGE
@@ -6186,3 +6187,70 @@ class InetRoundTripTests(fixtures.TestBase):
             "ipaddress type handling",
         ):
             testing_engine(options={"native_inet_types": True})
+
+
+class PGInsertManyValuesTest(fixtures.TestBase):
+    """test pg-specific types for insertmanyvalues"""
+
+    __only_on__ = "postgresql"
+    __backend__ = True
+
+    @testing.combinations(
+        ("PG BYTEA", BYTEA(), b"7\xe7\x9f"),
+        argnames="type_,value",
+        id_="iaa",
+    )
+    @testing.variation("sort_by_parameter_order", [True, False])
+    @testing.variation("multiple_rows", [True, False])
+    @testing.requires.insert_returning
+    def test_imv_returning_datatypes(
+        self,
+        connection,
+        metadata,
+        sort_by_parameter_order,
+        type_,
+        value,
+        multiple_rows,
+    ):
+        """test #9739, #9808 (similar to #9701) for PG specific types
+
+        this tests insertmanyvalues in conjunction with various datatypes.
+
+        These tests are particularly for the asyncpg driver which needs
+        most types to be explicitly cast for the new IMV format
+
+        """
+        t = Table(
+            "d_t",
+            metadata,
+            Column("id", Integer, primary_key=True),
+            Column("value", type_),
+        )
+
+        t.create(connection)
+
+        result = connection.execute(
+            t.insert().returning(
+                t.c.id,
+                t.c.value,
+                sort_by_parameter_order=bool(sort_by_parameter_order),
+            ),
+            [{"value": value} for i in range(10)]
+            if multiple_rows
+            else {"value": value},
+        )
+
+        if multiple_rows:
+            i_range = range(1, 11)
+        else:
+            i_range = range(1, 2)
+
+        eq_(
+            set(result),
+            {(id_, value) for id_ in i_range},
+        )
+
+        eq_(
+            set(connection.scalars(select(t.c.value))),
+            {value},
+        )