Implement autocommit isolation level for cx_oracle

As part of this change Oracle also gets the concept of a
default isolation level, however since Oracle does not provide a
fixed method of knowing what the isolation level would be without a
server side transaction actually in progress, for now we hardcode
just to "READ COMMITTED".

Enhanced the test suite for isolation level testing in the dialect
test suite and added features to requirements so that the supported
isolation levels can be reported generically for dialects.

Co-authored-by: Mike Bayer <mike_mp@zzzcomputing.com>
Fixes: #5200
Change-Id: I2c4d49da9ff80ccc228c21e196ec9a961de53478

1 files changed, 56 insertions, 1 deletions

diff --git a/lib/sqlalchemy/testing/suite/test_dialect.py b/lib/sqlalchemy/testing/suite/test_dialect.py
index 1b3307042..8aa13a622 100644
--- a/lib/sqlalchemy/testing/suite/test_dialect.py
+++ b/lib/sqlalchemy/testing/suite/test_dialect.py
@@ -4,6 +4,7 @@ from .. import assert_raises
 from .. import config
 from .. import eq_
 from .. import fixtures
+from .. import ne_
 from .. import provide_metadata
 from ..config import requirements
 from ..schema import Column
@@ -79,6 +80,46 @@ class ExceptionTest(fixtures.TablesTest):
                 assert isinstance(err_str, str)
 
 
+class IsolationLevelTest(fixtures.TestBase):
+    __backend__ = True
+
+    __requires__ = ("isolation_level",)
+
+    def _get_non_default_isolation_level(self):
+        levels = requirements.get_isolation_levels(config)
+
+        default = levels["default"]
+        supported = levels["supported"]
+
+        s = set(supported).difference(["AUTOCOMMIT", default])
+        if s:
+            return s.pop()
+        else:
+            config.skip_test("no non-default isolation level available")
+
+    def test_default_isolation_level(self):
+        eq_(
+            config.db.dialect.default_isolation_level,
+            requirements.get_isolation_levels(config)["default"],
+        )
+
+    def test_non_default_isolation_level(self):
+        non_default = self._get_non_default_isolation_level()
+
+        with config.db.connect() as conn:
+            existing = conn.get_isolation_level()
+
+            ne_(existing, non_default)
+
+            conn.execution_options(isolation_level=non_default)
+
+            eq_(conn.get_isolation_level(), non_default)
+
+            conn.dialect.reset_isolation_level(conn.connection)
+
+            eq_(conn.get_isolation_level(), existing)
+
+
 class AutocommitTest(fixtures.TablesTest):
 
     run_deletes = "each"
@@ -115,13 +156,27 @@ class AutocommitTest(fixtures.TablesTest):
         conn = config.db.connect()
         c2 = conn.execution_options(isolation_level="AUTOCOMMIT")
         self._test_conn_autocommits(c2, True)
-        conn.invalidate()
+
+        c2.dialect.reset_isolation_level(c2.connection)
+
         self._test_conn_autocommits(conn, False)
 
     def test_autocommit_off(self):
         conn = config.db.connect()
         self._test_conn_autocommits(conn, False)
 
+    def test_turn_autocommit_off_via_default_iso_level(self):
+        conn = config.db.connect()
+        conn.execution_options(isolation_level="AUTOCOMMIT")
+        self._test_conn_autocommits(conn, True)
+
+        conn.execution_options(
+            isolation_level=requirements.get_isolation_levels(config)[
+                "default"
+            ]
+        )
+        self._test_conn_autocommits(conn, False)
+
 
 class EscapingTest(fixtures.TestBase):
     @provide_metadata