Added support for ``psycopg`` dialect.

Both sync and async versions are supported.

Fixes: #6842
Change-Id: I57751c5028acebfc6f9c43572562405453a2f2a4

8 files changed, 390 insertions, 82 deletions

diff --git a/test/dialect/postgresql/test_dialect.py b/test/dialect/postgresql/test_dialect.py
index 57682686c..02d7ad483 100644
--- a/test/dialect/postgresql/test_dialect.py
+++ b/test/dialect/postgresql/test_dialect.py
@@ -8,6 +8,7 @@ from sqlalchemy import BigInteger
 from sqlalchemy import bindparam
 from sqlalchemy import cast
 from sqlalchemy import Column
+from sqlalchemy import create_engine
 from sqlalchemy import DateTime
 from sqlalchemy import DDL
 from sqlalchemy import event
@@ -30,6 +31,9 @@ from sqlalchemy import text
 from sqlalchemy import TypeDecorator
 from sqlalchemy import util
 from sqlalchemy.dialects.postgresql import base as postgresql
+from sqlalchemy.dialects.postgresql import HSTORE
+from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.dialects.postgresql import psycopg as psycopg_dialect
 from sqlalchemy.dialects.postgresql import psycopg2 as psycopg2_dialect
 from sqlalchemy.dialects.postgresql.psycopg2 import EXECUTEMANY_BATCH
 from sqlalchemy.dialects.postgresql.psycopg2 import EXECUTEMANY_PLAIN
@@ -269,10 +273,12 @@ class PGCodeTest(fixtures.TestBase):
         if testing.against("postgresql+pg8000"):
             # TODO: is there another way we're supposed to see this?
             eq_(errmsg.orig.args[0]["C"], "23505")
-        else:
+        elif not testing.against("postgresql+psycopg"):
             eq_(errmsg.orig.pgcode, "23505")
 
-        if testing.against("postgresql+asyncpg"):
+        if testing.against("postgresql+asyncpg") or testing.against(
+            "postgresql+psycopg"
+        ):
             eq_(errmsg.orig.sqlstate, "23505")
 
 
@@ -858,6 +864,13 @@ class MiscBackendTest(
             ".".join(str(x) for x in v)
         )
 
+    @testing.only_on("postgresql+psycopg")
+    def test_psycopg_version(self):
+        v = testing.db.dialect.psycopg_version
+        assert testing.db.dialect.dbapi.__version__.startswith(
+            ".".join(str(x) for x in v)
+        )
+
     @testing.combinations(
         ((8, 1), False, False),
         ((8, 1), None, False),
@@ -902,6 +915,7 @@ class MiscBackendTest(
         with testing.db.connect().execution_options(
             isolation_level="SERIALIZABLE"
         ) as conn:
+
             dbapi_conn = conn.connection.dbapi_connection
 
             is_false(dbapi_conn.autocommit)
@@ -1069,25 +1083,30 @@ class MiscBackendTest(
                 dbapi_conn.rollback()
             eq_(val, "off")
 
-    @testing.requires.psycopg2_compatibility
-    def test_psycopg2_non_standard_err(self):
+    @testing.requires.psycopg_compatibility
+    def test_psycopg_non_standard_err(self):
         # note that psycopg2 is sometimes called psycopg2cffi
         # depending on platform
-        psycopg2 = testing.db.dialect.dbapi
-        TransactionRollbackError = __import__(
-            "%s.extensions" % psycopg2.__name__
-        ).extensions.TransactionRollbackError
+        psycopg = testing.db.dialect.dbapi
+        if psycopg.__version__.startswith("3"):
+            TransactionRollbackError = __import__(
+                "%s.errors" % psycopg.__name__
+            ).errors.TransactionRollback
+        else:
+            TransactionRollbackError = __import__(
+                "%s.extensions" % psycopg.__name__
+            ).extensions.TransactionRollbackError
 
         exception = exc.DBAPIError.instance(
             "some statement",
             {},
             TransactionRollbackError("foo"),
-            psycopg2.Error,
+            psycopg.Error,
         )
         assert isinstance(exception, exc.OperationalError)
 
     @testing.requires.no_coverage
-    @testing.requires.psycopg2_compatibility
+    @testing.requires.psycopg_compatibility
     def test_notice_logging(self):
         log = logging.getLogger("sqlalchemy.dialects.postgresql")
         buf = logging.handlers.BufferingHandler(100)
@@ -1115,14 +1134,14 @@ $$ LANGUAGE plpgsql;
         finally:
             log.removeHandler(buf)
             log.setLevel(lev)
-        msgs = " ".join(b.msg for b in buf.buffer)
+        msgs = " ".join(b.getMessage() for b in buf.buffer)
         eq_regex(
             msgs,
-            "NOTICE:  notice: hi there(\nCONTEXT: .*?)? "
-            "NOTICE:  notice: another note(\nCONTEXT: .*?)?",
+            "NOTICE: [ ]?notice: hi there(\nCONTEXT: .*?)? "
+            "NOTICE: [ ]?notice: another note(\nCONTEXT: .*?)?",
         )
 
-    @testing.requires.psycopg2_or_pg8000_compatibility
+    @testing.requires.psycopg_or_pg8000_compatibility
     @engines.close_open_connections
     def test_client_encoding(self):
         c = testing.db.connect()
@@ -1143,7 +1162,7 @@ $$ LANGUAGE plpgsql;
         new_encoding = c.exec_driver_sql("show client_encoding").fetchone()[0]
         eq_(new_encoding, test_encoding)
 
-    @testing.requires.psycopg2_or_pg8000_compatibility
+    @testing.requires.psycopg_or_pg8000_compatibility
     @engines.close_open_connections
     def test_autocommit_isolation_level(self):
         c = testing.db.connect().execution_options(
@@ -1302,7 +1321,7 @@ $$ LANGUAGE plpgsql;
         assert result == [(1, "user", "lala")]
         connection.execute(text("DROP TABLE speedy_users"))
 
-    @testing.requires.psycopg2_or_pg8000_compatibility
+    @testing.requires.psycopg_or_pg8000_compatibility
     def test_numeric_raise(self, connection):
         stmt = text("select cast('hi' as char) as hi").columns(hi=Numeric)
         assert_raises(exc.InvalidRequestError, connection.execute, stmt)
@@ -1364,9 +1383,90 @@ $$ LANGUAGE plpgsql;
             )
 
     @testing.requires.psycopg2_compatibility
-    def test_initial_transaction_state(self):
+    def test_initial_transaction_state_psycopg2(self):
         from psycopg2.extensions import STATUS_IN_TRANSACTION
 
         engine = engines.testing_engine()
         with engine.connect() as conn:
             ne_(conn.connection.status, STATUS_IN_TRANSACTION)
+
+    @testing.only_on("postgresql+psycopg")
+    def test_initial_transaction_state_psycopg(self):
+        from psycopg.pq import TransactionStatus
+
+        engine = engines.testing_engine()
+        with engine.connect() as conn:
+            ne_(
+                conn.connection.dbapi_connection.info.transaction_status,
+                TransactionStatus.INTRANS,
+            )
+
+
+class Psycopg3Test(fixtures.TestBase):
+    __only_on__ = ("postgresql+psycopg",)
+
+    def test_json_correctly_registered(self, testing_engine):
+        import json
+
+        def loads(value):
+            value = json.loads(value)
+            value["x"] = value["x"] + "_loads"
+            return value
+
+        def dumps(value):
+            value = dict(value)
+            value["x"] = "dumps_y"
+            return json.dumps(value)
+
+        engine = testing_engine(
+            options=dict(json_serializer=dumps, json_deserializer=loads)
+        )
+        engine2 = testing_engine(
+            options=dict(
+                json_serializer=json.dumps, json_deserializer=json.loads
+            )
+        )
+
+        s = select(cast({"key": "value", "x": "q"}, JSONB))
+        with engine.begin() as conn:
+            eq_(conn.scalar(s), {"key": "value", "x": "dumps_y_loads"})
+            with engine.begin() as conn:
+                eq_(conn.scalar(s), {"key": "value", "x": "dumps_y_loads"})
+                with engine2.begin() as conn:
+                    eq_(conn.scalar(s), {"key": "value", "x": "q"})
+                with engine.begin() as conn:
+                    eq_(conn.scalar(s), {"key": "value", "x": "dumps_y_loads"})
+
+    @testing.requires.hstore
+    def test_hstore_correctly_registered(self, testing_engine):
+        engine = testing_engine(options=dict(use_native_hstore=True))
+        engine2 = testing_engine(options=dict(use_native_hstore=False))
+
+        def rp(self, *a):
+            return lambda a: {"a": "b"}
+
+        with mock.patch.object(HSTORE, "result_processor", side_effect=rp):
+            s = select(cast({"key": "value", "x": "q"}, HSTORE))
+            with engine.begin() as conn:
+                eq_(conn.scalar(s), {"key": "value", "x": "q"})
+                with engine.begin() as conn:
+                    eq_(conn.scalar(s), {"key": "value", "x": "q"})
+                    with engine2.begin() as conn:
+                        eq_(conn.scalar(s), {"a": "b"})
+                    with engine.begin() as conn:
+                        eq_(conn.scalar(s), {"key": "value", "x": "q"})
+
+    def test_get_dialect(self):
+        u = url.URL.create("postgresql://")
+        d = psycopg_dialect.PGDialect_psycopg.get_dialect_cls(u)
+        is_(d, psycopg_dialect.PGDialect_psycopg)
+        d = psycopg_dialect.PGDialect_psycopg.get_async_dialect_cls(u)
+        is_(d, psycopg_dialect.PGDialectAsync_psycopg)
+        d = psycopg_dialect.PGDialectAsync_psycopg.get_dialect_cls(u)
+        is_(d, psycopg_dialect.PGDialectAsync_psycopg)
+        d = psycopg_dialect.PGDialectAsync_psycopg.get_dialect_cls(u)
+        is_(d, psycopg_dialect.PGDialectAsync_psycopg)
+
+    def test_async_version(self):
+        e = create_engine("postgresql+psycopg_async://")
+        is_true(isinstance(e.dialect, psycopg_dialect.PGDialectAsync_psycopg))
diff --git a/test/dialect/postgresql/test_query.py b/test/dialect/postgresql/test_query.py
index b488b146c..fdce643f8 100644
--- a/test/dialect/postgresql/test_query.py
+++ b/test/dialect/postgresql/test_query.py
@@ -14,6 +14,7 @@ from sqlalchemy import Float
 from sqlalchemy import ForeignKey
 from sqlalchemy import func
 from sqlalchemy import Integer
+from sqlalchemy import JSON
 from sqlalchemy import literal
 from sqlalchemy import literal_column
 from sqlalchemy import MetaData
@@ -29,6 +30,7 @@ from sqlalchemy import true
 from sqlalchemy import tuple_
 from sqlalchemy.dialects import postgresql
 from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.sql.expression import type_coerce
 from sqlalchemy.testing import assert_raises
 from sqlalchemy.testing import AssertsCompiledSQL
 from sqlalchemy.testing import AssertsExecutionResults
@@ -40,6 +42,17 @@ from sqlalchemy.testing.assertsql import CursorSQL
 from sqlalchemy.testing.assertsql import DialectSQL
 
 
+class FunctionTypingTest(fixtures.TestBase, AssertsExecutionResults):
+    __only_on__ = "postgresql"
+    __backend__ = True
+
+    def test_count_star(self, connection):
+        eq_(connection.scalar(func.count("*")), 1)
+
+    def test_count_int(self, connection):
+        eq_(connection.scalar(func.count(1)), 1)
+
+
 class InsertTest(fixtures.TestBase, AssertsExecutionResults):
 
     __only_on__ = "postgresql"
@@ -956,23 +969,42 @@ class MatchTest(fixtures.TablesTest, AssertsCompiledSQL):
             ],
         )
 
+    def _strs_render_bind_casts(self, connection):
+
+        return (
+            connection.dialect._bind_typing_render_casts
+            and String().dialect_impl(connection.dialect).render_bind_cast
+        )
+
     @testing.requires.pyformat_paramstyle
-    def test_expression_pyformat(self):
+    def test_expression_pyformat(self, connection):
         matchtable = self.tables.matchtable
-        self.assert_compile(
-            matchtable.c.title.match("somstr"),
-            "matchtable.title @@ to_tsquery(%(title_1)s" ")",
-        )
+
+        if self._strs_render_bind_casts(connection):
+            self.assert_compile(
+                matchtable.c.title.match("somstr"),
+                "matchtable.title @@ to_tsquery(%(title_1)s::VARCHAR(200))",
+            )
+        else:
+            self.assert_compile(
+                matchtable.c.title.match("somstr"),
+                "matchtable.title @@ to_tsquery(%(title_1)s)",
+            )
 
     @testing.requires.format_paramstyle
-    def test_expression_positional(self):
+    def test_expression_positional(self, connection):
         matchtable = self.tables.matchtable
-        self.assert_compile(
-            matchtable.c.title.match("somstr"),
-            # note we assume current tested DBAPIs use emulated setinputsizes
-            # here, the cast is not strictly necessary
-            "matchtable.title @@ to_tsquery(%s::VARCHAR(200))",
-        )
+
+        if self._strs_render_bind_casts(connection):
+            self.assert_compile(
+                matchtable.c.title.match("somstr"),
+                "matchtable.title @@ to_tsquery(%s::VARCHAR(200))",
+            )
+        else:
+            self.assert_compile(
+                matchtable.c.title.match("somstr"),
+                "matchtable.title @@ to_tsquery(%s)",
+            )
 
     def test_simple_match(self, connection):
         matchtable = self.tables.matchtable
@@ -1551,17 +1583,106 @@ class TableValuedRoundTripTest(fixtures.TestBase):
             [(14, 1), (41, 2), (7, 3), (54, 4), (9, 5), (49, 6)],
         )
 
-    @testing.only_on(
-        "postgresql+psycopg2",
-        "I cannot get this to run at all on other drivers, "
-        "even selecting from a table",
+    @testing.combinations(
+        (
+            type_coerce,
+            testing.fails("fails on all drivers"),
+        ),
+        (
+            cast,
+            testing.fails("fails on all drivers"),
+        ),
+        (
+            None,
+            testing.fails_on_everything_except(
+                ["postgresql+psycopg2"],
+                "I cannot get this to run at all on other drivers, "
+                "even selecting from a table",
+            ),
+        ),
+        argnames="cast_fn",
     )
-    def test_render_derived_quoting(self, connection):
+    def test_render_derived_quoting_text(self, connection, cast_fn):
+
+        value = (
+            '[{"CaseSensitive":1,"the % value":"foo"}, '
+            '{"CaseSensitive":"2","the % value":"bar"}]'
+        )
+
+        if cast_fn:
+            value = cast_fn(value, JSON)
+
         fn = (
-            func.json_to_recordset(  # noqa
-                '[{"CaseSensitive":1,"the % value":"foo"}, '
-                '{"CaseSensitive":"2","the % value":"bar"}]'
+            func.json_to_recordset(value)
+            .table_valued(
+                column("CaseSensitive", Integer), column("the % value", String)
             )
+            .render_derived(with_types=True)
+        )
+
+        stmt = select(fn.c.CaseSensitive, fn.c["the % value"])
+
+        eq_(connection.execute(stmt).all(), [(1, "foo"), (2, "bar")])
+
+    @testing.combinations(
+        (
+            type_coerce,
+            testing.fails("fails on all drivers"),
+        ),
+        (
+            cast,
+            testing.fails("fails on all drivers"),
+        ),
+        (
+            None,
+            testing.fails("Fails on all drivers"),
+        ),
+        argnames="cast_fn",
+    )
+    def test_render_derived_quoting_text_to_json(self, connection, cast_fn):
+
+        value = (
+            '[{"CaseSensitive":1,"the % value":"foo"}, '
+            '{"CaseSensitive":"2","the % value":"bar"}]'
+        )
+
+        if cast_fn:
+            value = cast_fn(value, JSON)
+
+        # why wont this work?!?!?
+        # should be exactly json_to_recordset(to_json('string'::text))
+        #
+        fn = (
+            func.json_to_recordset(func.to_json(value))
+            .table_valued(
+                column("CaseSensitive", Integer), column("the % value", String)
+            )
+            .render_derived(with_types=True)
+        )
+
+        stmt = select(fn.c.CaseSensitive, fn.c["the % value"])
+
+        eq_(connection.execute(stmt).all(), [(1, "foo"), (2, "bar")])
+
+    @testing.combinations(
+        (type_coerce,),
+        (cast,),
+        (None, testing.fails("Fails on all PG backends")),
+        argnames="cast_fn",
+    )
+    def test_render_derived_quoting_straight_json(self, connection, cast_fn):
+        # these all work
+
+        value = [
+            {"CaseSensitive": 1, "the % value": "foo"},
+            {"CaseSensitive": "2", "the % value": "bar"},
+        ]
+
+        if cast_fn:
+            value = cast_fn(value, JSON)
+
+        fn = (
+            func.json_to_recordset(value)  # noqa
             .table_valued(
                 column("CaseSensitive", Integer), column("the % value", String)
             )
diff --git a/test/dialect/postgresql/test_types.py b/test/dialect/postgresql/test_types.py
index 4008881d2..5f8a41d1f 100644
--- a/test/dialect/postgresql/test_types.py
+++ b/test/dialect/postgresql/test_types.py
@@ -2360,7 +2360,7 @@ class ArrayEnum(fixtures.TestBase):
             testing.combinations(
                 sqltypes.ARRAY,
                 postgresql.ARRAY,
-                (_ArrayOfEnum, testing.only_on("postgresql+psycopg2")),
+                (_ArrayOfEnum, testing.requires.psycopg_compatibility),
                 argnames="array_cls",
             )(fn)
         )
@@ -3066,7 +3066,7 @@ class HStoreRoundTripTest(fixtures.TablesTest):
 
     @testing.fixture
     def non_native_hstore_connection(self, testing_engine):
-        local_engine = testing.requires.psycopg2_native_hstore.enabled
+        local_engine = testing.requires.native_hstore.enabled
 
         if local_engine:
             engine = testing_engine(options=dict(use_native_hstore=False))
@@ -3096,14 +3096,14 @@ class HStoreRoundTripTest(fixtures.TablesTest):
         )["1"]
         eq_(connection.scalar(select(expr)), "3")
 
-    @testing.requires.psycopg2_native_hstore
+    @testing.requires.native_hstore
     def test_insert_native(self, connection):
         self._test_insert(connection)
 
     def test_insert_python(self, non_native_hstore_connection):
         self._test_insert(non_native_hstore_connection)
 
-    @testing.requires.psycopg2_native_hstore
+    @testing.requires.native_hstore
     def test_criterion_native(self, connection):
         self._fixture_data(connection)
         self._test_criterion(connection)
@@ -3134,7 +3134,7 @@ class HStoreRoundTripTest(fixtures.TablesTest):
     def test_fixed_round_trip_python(self, non_native_hstore_connection):
         self._test_fixed_round_trip(non_native_hstore_connection)
 
-    @testing.requires.psycopg2_native_hstore
+    @testing.requires.native_hstore
     def test_fixed_round_trip_native(self, connection):
         self._test_fixed_round_trip(connection)
 
@@ -3154,11 +3154,11 @@ class HStoreRoundTripTest(fixtures.TablesTest):
             },
         )
 
-    @testing.requires.psycopg2_native_hstore
+    @testing.requires.native_hstore
     def test_unicode_round_trip_python(self, non_native_hstore_connection):
         self._test_unicode_round_trip(non_native_hstore_connection)
 
-    @testing.requires.psycopg2_native_hstore
+    @testing.requires.native_hstore
     def test_unicode_round_trip_native(self, connection):
         self._test_unicode_round_trip(connection)
 
@@ -3167,7 +3167,7 @@ class HStoreRoundTripTest(fixtures.TablesTest):
     ):
         self._test_escaped_quotes_round_trip(non_native_hstore_connection)
 
-    @testing.requires.psycopg2_native_hstore
+    @testing.requires.native_hstore
     def test_escaped_quotes_round_trip_native(self, connection):
         self._test_escaped_quotes_round_trip(connection)
 
@@ -3356,7 +3356,7 @@ class _RangeTypeCompilation(AssertsCompiledSQL, fixtures.TestBase):
 
 
 class _RangeTypeRoundTrip(fixtures.TablesTest):
-    __requires__ = "range_types", "psycopg2_compatibility"
+    __requires__ = "range_types", "psycopg_compatibility"
     __backend__ = True
 
     def extras(self):
@@ -3364,8 +3364,18 @@ class _RangeTypeRoundTrip(fixtures.TablesTest):
         # older psycopg2 versions.
         if testing.against("postgresql+psycopg2cffi"):
             from psycopg2cffi import extras
-        else:
+        elif testing.against("postgresql+psycopg2"):
             from psycopg2 import extras
+        elif testing.against("postgresql+psycopg"):
+            from psycopg.types.range import Range
+
+            class psycopg_extras:
+                def __getattr__(self, _):
+                    return Range
+
+            extras = psycopg_extras()
+        else:
+            assert False, "Unknonw dialect"
         return extras
 
     @classmethod
diff --git a/test/engine/test_parseconnect.py b/test/engine/test_parseconnect.py
index c69b332a8..f12d32d5d 100644
--- a/test/engine/test_parseconnect.py
+++ b/test/engine/test_parseconnect.py
@@ -1037,6 +1037,64 @@ class TestRegNewDBAPI(fixtures.TestBase):
         )
 
 
+class TestGetDialect(fixtures.TestBase):
+    @testing.requires.sqlite
+    @testing.combinations(True, False, None)
+    def test_is_async_to_create_engine(self, is_async):
+        def get_dialect_cls(url):
+            url = url.set(drivername="sqlite")
+            return url.get_dialect()
+
+        global MockDialectGetDialect
+        MockDialectGetDialect = Mock()
+        MockDialectGetDialect.get_dialect_cls.side_effect = get_dialect_cls
+        MockDialectGetDialect.get_async_dialect_cls.side_effect = (
+            get_dialect_cls
+        )
+
+        registry.register("mockdialect", __name__, "MockDialectGetDialect")
+
+        from sqlalchemy.dialects import sqlite
+
+        kw = {}
+        if is_async is not None:
+            kw["_is_async"] = is_async
+        e = create_engine("mockdialect://", **kw)
+
+        eq_(e.dialect.name, "sqlite")
+        assert isinstance(e.dialect, sqlite.dialect)
+
+        if is_async:
+            eq_(
+                MockDialectGetDialect.mock_calls,
+                [
+                    call.get_async_dialect_cls(url.make_url("mockdialect://")),
+                    call.engine_created(e),
+                ],
+            )
+        else:
+            eq_(
+                MockDialectGetDialect.mock_calls,
+                [
+                    call.get_dialect_cls(url.make_url("mockdialect://")),
+                    call.engine_created(e),
+                ],
+            )
+        MockDialectGetDialect.reset_mock()
+        u = url.make_url("mockdialect://")
+        u.get_dialect(**kw)
+        if is_async:
+            eq_(
+                MockDialectGetDialect.mock_calls,
+                [call.get_async_dialect_cls(u)],
+            )
+        else:
+            eq_(
+                MockDialectGetDialect.mock_calls,
+                [call.get_dialect_cls(u)],
+            )
+
+
 class MockDialect(DefaultDialect):
     @classmethod
     def dbapi(cls, **kw):
diff --git a/test/engine/test_reconnect.py b/test/engine/test_reconnect.py
index afd027698..f37d1893f 100644
--- a/test/engine/test_reconnect.py
+++ b/test/engine/test_reconnect.py
@@ -365,7 +365,7 @@ class MockReconnectTest(fixtures.TestBase):
         )
 
         self.mock_connect = call(
-            host="localhost", password="bar", user="foo", database="test"
+            host="localhost", password="bar", user="foo", dbname="test"
         )
         # monkeypatch disconnect checker
         self.db.dialect.is_disconnect = lambda e, conn, cursor: isinstance(
@@ -1321,6 +1321,7 @@ class InvalidateDuringResultTest(fixtures.TestBase):
             "+aiosqlite",
             "+aiomysql",
             "+asyncmy",
+            "+psycopg",
         ],
         "Buffers the result set and doesn't check for connection close",
     )
diff --git a/test/ext/asyncio/test_engine_py3k.py b/test/ext/asyncio/test_engine_py3k.py
index 44cf9388c..aee71f8d5 100644
--- a/test/ext/asyncio/test_engine_py3k.py
+++ b/test/ext/asyncio/test_engine_py3k.py
@@ -634,7 +634,11 @@ class AsyncEventTest(EngineFixture):
 
         eq_(
             canary.mock_calls,
-            [mock.call(sync_conn, mock.ANY, "select 1", (), mock.ANY, False)],
+            [
+                mock.call(
+                    sync_conn, mock.ANY, "select 1", mock.ANY, mock.ANY, False
+                )
+            ],
         )
 
     @async_test
@@ -651,7 +655,11 @@ class AsyncEventTest(EngineFixture):
 
         eq_(
             canary.mock_calls,
-            [mock.call(sync_conn, mock.ANY, "select 1", (), mock.ANY, False)],
+            [
+                mock.call(
+                    sync_conn, mock.ANY, "select 1", mock.ANY, mock.ANY, False
+                )
+            ],
         )
 
     @async_test
diff --git a/test/requirements.py b/test/requirements.py
index eeb71323b..3934dd23f 100644
--- a/test/requirements.py
+++ b/test/requirements.py
@@ -223,6 +223,7 @@ class DefaultRequirements(SuiteRequirements):
     def pyformat_paramstyle(self):
         return only_on(
             [
+                "postgresql+psycopg",
                 "postgresql+psycopg2",
                 "postgresql+psycopg2cffi",
                 "mysql+mysqlconnector",
@@ -1161,7 +1162,10 @@ class DefaultRequirements(SuiteRequirements):
     @property
     def infinity_floats(self):
         return fails_on_everything_except(
-            "sqlite", "postgresql+psycopg2", "postgresql+asyncpg"
+            "sqlite",
+            "postgresql+psycopg2",
+            "postgresql+asyncpg",
+            "postgresql+psycopg",
         ) + skip_if(
             "postgresql+pg8000", "seems to work on pg14 only, not earlier?"
         )
@@ -1241,9 +1245,7 @@ class DefaultRequirements(SuiteRequirements):
     @property
     def range_types(self):
         def check_range_types(config):
-            if not against(
-                config, ["postgresql+psycopg2", "postgresql+psycopg2cffi"]
-            ):
+            if not self.psycopg_compatibility.enabled:
                 return False
             try:
                 with config.db.connect() as conn:
@@ -1291,24 +1293,28 @@ class DefaultRequirements(SuiteRequirements):
         )
 
     @property
-    def psycopg2_native_hstore(self):
-        return self.psycopg2_compatibility
+    def native_hstore(self):
+        return self.psycopg_compatibility
 
     @property
     def psycopg2_compatibility(self):
         return only_on(["postgresql+psycopg2", "postgresql+psycopg2cffi"])
 
     @property
-    def psycopg2_or_pg8000_compatibility(self):
+    def psycopg_compatibility(self):
         return only_on(
             [
                 "postgresql+psycopg2",
                 "postgresql+psycopg2cffi",
-                "postgresql+pg8000",
+                "postgresql+psycopg",
             ]
         )
 
     @property
+    def psycopg_or_pg8000_compatibility(self):
+        return only_on([self.psycopg_compatibility, "postgresql+pg8000"])
+
+    @property
     def percent_schema_names(self):
         return skip_if(
             ["mysql+aiomysql", "mariadb+aiomysql"],
@@ -1690,3 +1696,8 @@ class DefaultRequirements(SuiteRequirements):
     def reflect_tables_no_columns(self):
         # so far sqlite, mariadb, mysql don't support this
         return only_on(["postgresql"])
+
+    @property
+    def json_deserializer_binary(self):
+        "indicates if the json_deserializer function is called with bytes"
+        return only_on(["postgresql+psycopg"])
diff --git a/test/sql/test_types.py b/test/sql/test_types.py
index f63e1e01b..35467de94 100644
--- a/test/sql/test_types.py
+++ b/test/sql/test_types.py
@@ -2459,15 +2459,16 @@ class EnumTest(AssertsCompiledSQL, fixtures.TablesTest):
     @testing.combinations(
         (True, "omit_alias"), (False, "with_alias"), id_="ai", argnames="omit"
     )
-    @testing.provide_metadata
     @testing.skip_if("mysql < 8")
-    def test_duplicate_values_accepted(self, native, omit):
+    def test_duplicate_values_accepted(
+        self, metadata, connection, native, omit
+    ):
         foo_enum = pep435_enum("foo_enum")
         foo_enum("one", 1, "two")
         foo_enum("three", 3, "four")
         tbl = sa.Table(
             "foo_table",
-            self.metadata,
+            metadata,
             sa.Column("id", sa.Integer),
             sa.Column(
                 "data",
@@ -2481,7 +2482,7 @@ class EnumTest(AssertsCompiledSQL, fixtures.TablesTest):
         )
         t = sa.table("foo_table", sa.column("id"), sa.column("data"))
 
-        self.metadata.create_all(testing.db)
+        metadata.create_all(connection)
         if omit:
             with expect_raises(
                 (
@@ -2491,29 +2492,27 @@ class EnumTest(AssertsCompiledSQL, fixtures.TablesTest):
                     exc.DBAPIError,
                 )
             ):
-                with testing.db.begin() as conn:
-                    conn.execute(
-                        t.insert(),
-                        [
-                            {"id": 1, "data": "four"},
-                            {"id": 2, "data": "three"},
-                        ],
-                    )
-        else:
-            with testing.db.begin() as conn:
-                conn.execute(
+                connection.execute(
                     t.insert(),
-                    [{"id": 1, "data": "four"}, {"id": 2, "data": "three"}],
+                    [
+                        {"id": 1, "data": "four"},
+                        {"id": 2, "data": "three"},
+                    ],
                 )
+        else:
+            connection.execute(
+                t.insert(),
+                [{"id": 1, "data": "four"}, {"id": 2, "data": "three"}],
+            )
 
-                eq_(
-                    conn.execute(t.select().order_by(t.c.id)).fetchall(),
-                    [(1, "four"), (2, "three")],
-                )
-                eq_(
-                    conn.execute(tbl.select().order_by(tbl.c.id)).fetchall(),
-                    [(1, foo_enum.three), (2, foo_enum.three)],
-                )
+            eq_(
+                connection.execute(t.select().order_by(t.c.id)).fetchall(),
+                [(1, "four"), (2, "three")],
+            )
+            eq_(
+                connection.execute(tbl.select().order_by(tbl.c.id)).fetchall(),
+                [(1, foo_enum.three), (2, foo_enum.three)],
+            )
 
 
 MyPickleType = None