Merge "fully support isolation_level parameter in base dialect" into main

3 files changed, 104 insertions, 3 deletions

diff --git a/lib/sqlalchemy/testing/assertions.py b/lib/sqlalchemy/testing/assertions.py
index c30fdf823..3aa2649f4 100644
--- a/lib/sqlalchemy/testing/assertions.py
+++ b/lib/sqlalchemy/testing/assertions.py
@@ -384,9 +384,13 @@ def _expect_raises(except_cls, msg=None, check_context=False):
         ec.error = err
         success = True
         if msg is not None:
-            assert re.search(
-                msg, util.text_type(err), re.UNICODE
-            ), "%r !~ %s" % (msg, err)
+            # I'm often pdbing here, and "err" above isn't
+            # in scope, so assign the string explicitly
+            error_as_string = util.text_type(err)
+            assert re.search(msg, error_as_string, re.UNICODE), "%r !~ %s" % (
+                msg,
+                error_as_string,
+            )
         if check_context and not are_we_already_in_a_traceback:
             _assert_proper_exception_context(err)
         print(util.text_type(err).encode("utf-8"))
diff --git a/lib/sqlalchemy/testing/requirements.py b/lib/sqlalchemy/testing/requirements.py
index 56df452a5..43849efca 100644
--- a/lib/sqlalchemy/testing/requirements.py
+++ b/lib/sqlalchemy/testing/requirements.py
@@ -20,6 +20,7 @@ import sys
 
 from . import exclusions
 from . import only_on
+from .. import create_engine
 from .. import util
 from ..pool import QueuePool
 
@@ -873,6 +874,59 @@ class SuiteRequirements(Requirements):
                 ]
             }
         """
+        with config.db.connect() as conn:
+
+            try:
+                supported = conn.dialect.get_isolation_level_values(
+                    conn.connection.dbapi_connection
+                )
+            except NotImplementedError:
+                return None
+            else:
+                return {
+                    "default": conn.dialect.default_isolation_level,
+                    "supported": supported,
+                }
+
+    @property
+    def get_isolation_level_values(self):
+        """target dialect supports the
+        :meth:`_engine.Dialect.get_isolation_level_values`
+        method added in SQLAlchemy 2.0.
+
+        """
+
+        def go(config):
+            with config.db.connect() as conn:
+                try:
+                    conn.dialect.get_isolation_level_values(
+                        conn.connection.dbapi_connection
+                    )
+                except NotImplementedError:
+                    return False
+                else:
+                    return True
+
+        return exclusions.only_if(go)
+
+    @property
+    def dialect_level_isolation_level_param(self):
+        """test that the dialect allows the 'isolation_level' argument
+        to be handled by DefaultDialect"""
+
+        def go(config):
+            try:
+                e = create_engine(
+                    config.db.url, isolation_level="READ COMMITTED"
+                )
+            except:
+                return False
+            else:
+                return (
+                    e.dialect._on_connect_isolation_level == "READ COMMITTED"
+                )
+
+        return exclusions.only_if(go)
 
     @property
     def json_type(self):
diff --git a/lib/sqlalchemy/testing/suite/test_dialect.py b/lib/sqlalchemy/testing/suite/test_dialect.py
index b3e43aad0..28fd99876 100644
--- a/lib/sqlalchemy/testing/suite/test_dialect.py
+++ b/lib/sqlalchemy/testing/suite/test_dialect.py
@@ -8,6 +8,7 @@ from .. import eq_
 from .. import fixtures
 from .. import ne_
 from .. import provide_metadata
+from ..assertions import expect_raises_message
 from ..config import requirements
 from ..provision import set_default_schema_on_connection
 from ..schema import Column
@@ -140,6 +141,48 @@ class IsolationLevelTest(fixtures.TestBase):
                     levels["default"],
                 )
 
+    @testing.requires.get_isolation_level_values
+    def test_invalid_level_execution_option(self, connection_no_trans):
+        """test for the new get_isolation_level_values() method"""
+
+        connection = connection_no_trans
+        with expect_raises_message(
+            exc.ArgumentError,
+            "Invalid value '%s' for isolation_level. "
+            "Valid isolation levels for '%s' are %s"
+            % (
+                "FOO",
+                connection.dialect.name,
+                ", ".join(
+                    requirements.get_isolation_levels(config)["supported"]
+                ),
+            ),
+        ):
+            connection.execution_options(isolation_level="FOO")
+
+    @testing.requires.get_isolation_level_values
+    @testing.requires.dialect_level_isolation_level_param
+    def test_invalid_level_engine_param(self, testing_engine):
+        """test for the new get_isolation_level_values() method
+        and support for the dialect-level 'isolation_level' parameter.
+
+        """
+
+        eng = testing_engine(options=dict(isolation_level="FOO"))
+        with expect_raises_message(
+            exc.ArgumentError,
+            "Invalid value '%s' for isolation_level. "
+            "Valid isolation levels for '%s' are %s"
+            % (
+                "FOO",
+                eng.dialect.name,
+                ", ".join(
+                    requirements.get_isolation_levels(config)["supported"]
+                ),
+            ),
+        ):
+            eng.connect()
+
 
 class AutocommitIsolationTest(fixtures.TablesTest):