fully support isolation_level parameter in base dialect

Generalized the :paramref:`_sa.create_engine.isolation_level` parameter to
the base dialect so that it is no longer dependent on individual dialects
to be present. This parameter sets up the "isolation level" setting to
occur for all new database connections as soon as they are created by the
connection pool, where the value then stays set without being reset on
every checkin.

The :paramref:`_sa.create_engine.isolation_level` parameter is essentially
equivalent in functionality to using the
:paramref:`_engine.Engine.execution_options.isolation_level` parameter via
:meth:`_engine.Engine.execution_options` for an engine-wide setting. The
difference is in that the former setting assigns the isolation level just
once when a connection is created, the latter sets and resets the given
level on each connection checkout.

Fixes: #6342
Change-Id: Id81d6b1c1a94371d901ada728a610696e09e9741

3 files changed, 71 insertions, 79 deletions

diff --git a/test/dialect/test_sqlite.py b/test/dialect/test_sqlite.py
index f1efaf7a6..ca203ad8c 100644
--- a/test/dialect/test_sqlite.py
+++ b/test/dialect/test_sqlite.py
@@ -54,7 +54,6 @@ from sqlalchemy.testing import expect_warnings
 from sqlalchemy.testing import fixtures
 from sqlalchemy.testing import is_
 from sqlalchemy.testing import mock
-from sqlalchemy.testing.assertions import expect_raises_message
 from sqlalchemy.types import Boolean
 from sqlalchemy.types import Date
 from sqlalchemy.types import DateTime
@@ -597,21 +596,6 @@ class DialectTest(
                 )
             )
 
-    @testing.only_on("sqlite+pysqlite")
-    def test_isolation_level_message(self):
-        # needs to test that all three words are present and we also
-        # dont want to default all isolation level messages to use
-        # sorted(), so rely on python 3.7 for ordering of keywords
-        # in the message
-        with expect_raises_message(
-            exc.ArgumentError,
-            "Invalid value 'invalid' for "
-            "isolation_level. Valid isolation levels for "
-            "sqlite are READ UNCOMMITTED, SERIALIZABLE, AUTOCOMMIT",
-        ):
-            with testing.db.connect() as conn:
-                conn.execution_options(isolation_level="invalid")
-
     @testing.only_on("sqlite+pysqlcipher")
     def test_pysqlcipher_connects(self):
         """test #6586"""
diff --git a/test/engine/test_transaction.py b/test/engine/test_transaction.py
index 50da425bd..e18f1c446 100644
--- a/test/engine/test_transaction.py
+++ b/test/engine/test_transaction.py
@@ -1102,6 +1102,8 @@ class AutoRollbackTest(fixtures.TestBase):
 
 
 class IsolationLevelTest(fixtures.TestBase):
+    """see also sqlalchemy/testing/suite/test_dialect.py::IsolationLevelTest"""
+
     __requires__ = (
         "isolation_level",
         "ad_hoc_engines",
@@ -1123,7 +1125,6 @@ class IsolationLevelTest(fixtures.TestBase):
         else:
             assert False, "no non-default isolation level available"
 
-    @testing.requires.legacy_isolation_level
     def test_engine_param_stays(self):
 
         eng = testing_engine()
@@ -1177,7 +1178,6 @@ class IsolationLevelTest(fixtures.TestBase):
 
         conn.close()
 
-    @testing.requires.legacy_isolation_level
     def test_reset_level_with_setting(self):
         eng = testing_engine(
             options=dict(isolation_level=self._non_default_isolation_level())
@@ -1201,45 +1201,88 @@ class IsolationLevelTest(fixtures.TestBase):
         )
         conn.close()
 
-    @testing.requires.legacy_isolation_level
+    def test_underscore_replacement(self, connection_no_trans):
+        conn = connection_no_trans
+        with mock.patch.object(
+            conn.dialect, "set_isolation_level"
+        ) as mock_sil, mock.patch.object(
+            conn.dialect,
+            "_gen_allowed_isolation_levels",
+            mock.Mock(return_value=["READ COMMITTED", "REPEATABLE READ"]),
+        ):
+            conn.execution_options(isolation_level="REPEATABLE_READ")
+            dbapi_conn = conn.connection.dbapi_connection
+
+        eq_(mock_sil.mock_calls, [mock.call(dbapi_conn, "REPEATABLE READ")])
+
+    def test_casing_replacement(self, connection_no_trans):
+        conn = connection_no_trans
+        with mock.patch.object(
+            conn.dialect, "set_isolation_level"
+        ) as mock_sil, mock.patch.object(
+            conn.dialect,
+            "_gen_allowed_isolation_levels",
+            mock.Mock(return_value=["READ COMMITTED", "REPEATABLE READ"]),
+        ):
+            conn.execution_options(isolation_level="repeatable_read")
+            dbapi_conn = conn.connection.dbapi_connection
+
+        eq_(mock_sil.mock_calls, [mock.call(dbapi_conn, "REPEATABLE READ")])
+
+    def test_dialect_doesnt_follow_naming_guidelines(
+        self, connection_no_trans
+    ):
+        conn = connection_no_trans
+
+        conn.dialect.__dict__.pop("_gen_allowed_isolation_levels", None)
+        with mock.patch.object(
+            conn.dialect,
+            "get_isolation_level_values",
+            mock.Mock(
+                return_value=[
+                    "READ COMMITTED",
+                    "REPEATABLE_READ",
+                    "serializable",
+                ]
+            ),
+        ):
+            with expect_raises_message(
+                ValueError,
+                f"Dialect {conn.dialect.name!r} "
+                r"get_isolation_level_values\(\) method "
+                r"should "
+                r"return names as UPPERCASE using spaces, not underscores; "
+                r"got \['REPEATABLE_READ', 'serializable'\]",
+            ):
+                conn.execution_options(isolation_level="READ COMMITTED")
+
     def test_invalid_level_engine_param(self):
         eng = testing_engine(options=dict(isolation_level="FOO"))
         assert_raises_message(
             exc.ArgumentError,
-            "Invalid value '%s' for isolation_level. "
-            "Valid isolation levels for %s are %s"
-            % (
-                "FOO",
-                eng.dialect.name,
-                ", ".join(eng.dialect._isolation_lookup),
-            ),
+            f"Invalid value 'FOO' for isolation_level. "
+            f"Valid isolation levels for {eng.dialect.name!r} are "
+            f"""{', '.join(
+                testing.requires.get_isolation_levels(
+                    testing.config
+                )['supported']
+            )}""",
             eng.connect,
         )
 
-    # TODO: all the dialects seem to be manually raising ArgumentError
-    # individually within their set_isolation_level() methods, when this
-    # should be a default dialect feature so that
-    # error messaging etc. is consistent, including that it works for 3rd
-    # party dialects.
-    # TODO: barring that, at least implement this for the Oracle dialect
-    @testing.fails_on(
-        "oracle",
-        "cx_oracle dialect doesnt have argument error here, "
-        "raises it via the DB rejecting it",
-    )
     def test_invalid_level_execution_option(self):
         eng = testing_engine(
             options=dict(execution_options=dict(isolation_level="FOO"))
         )
         assert_raises_message(
             exc.ArgumentError,
-            "Invalid value '%s' for isolation_level. "
-            "Valid isolation levels for %s are %s"
-            % (
-                "FOO",
-                eng.dialect.name,
-                ", ".join(eng.dialect._isolation_lookup),
-            ),
+            f"Invalid value 'FOO' for isolation_level. "
+            f"Valid isolation levels for {eng.dialect.name!r} are "
+            f"""{', '.join(
+                testing.requires.get_isolation_levels(
+                    testing.config
+                )['supported']
+            )}""",
             eng.connect,
         )
 
diff --git a/test/requirements.py b/test/requirements.py
index 28a283b12..378d54fbd 100644
--- a/test/requirements.py
+++ b/test/requirements.py
@@ -383,41 +383,6 @@ class DefaultRequirements(SuiteRequirements):
         )
 
     @property
-    def legacy_isolation_level(self):
-        # refers dialects where "isolation_level" can be passed to
-        # create_engine
-        return only_on(
-            ("postgresql", "sqlite", "mysql", "mariadb", "mssql"),
-            "DBAPI has no isolation level support",
-        )
-
-    def get_isolation_levels(self, config):
-        levels = set(config.db.dialect._isolation_lookup)
-
-        if against(config, "sqlite"):
-            default = "SERIALIZABLE"
-            levels.add("AUTOCOMMIT")
-        elif against(config, "postgresql"):
-            default = "READ COMMITTED"
-            levels.add("AUTOCOMMIT")
-        elif against(config, "mysql"):
-            default = "REPEATABLE READ"
-            levels.add("AUTOCOMMIT")
-        elif against(config, "mariadb"):
-            default = "REPEATABLE READ"
-            levels.add("AUTOCOMMIT")
-        elif against(config, "mssql"):
-            default = "READ COMMITTED"
-            levels.add("AUTOCOMMIT")
-        elif against(config, "oracle"):
-            default = "READ COMMITTED"
-            levels.add("AUTOCOMMIT")
-        else:
-            raise NotImplementedError()
-
-        return {"default": default, "supported": levels}
-
-    @property
     def autocommit(self):
         """target dialect supports 'AUTOCOMMIT' as an isolation_level"""