- Added new user-space accessors for viewing transaction isolation

levels; :meth:`.Connection.get_isolation_level`,
:attr:`.Connection.default_isolation_level`.
- enhance documentation inter-linkage between new accessors,
existing isolation_level parameters, as well as in
the dialect-level methods which should be fully covered
by Engine/Connection level APIs now.

2 files changed, 43 insertions, 0 deletions

diff --git a/test/engine/test_execute.py b/test/engine/test_execute.py
index 8e58d202d..725dcebe0 100644
--- a/test/engine/test_execute.py
+++ b/test/engine/test_execute.py
@@ -1900,6 +1900,27 @@ class HandleErrorTest(fixtures.TestBase):
         self._test_alter_disconnect(True, False)
         self._test_alter_disconnect(False, False)
 
+    def test_handle_error_event_connect_isolation_level(self):
+        engine = engines.testing_engine()
+
+        class MySpecialException(Exception):
+            pass
+
+        @event.listens_for(engine, "handle_error")
+        def handle_error(ctx):
+            raise MySpecialException("failed operation")
+
+        ProgrammingError = engine.dialect.dbapi.ProgrammingError
+        with engine.connect() as conn:
+            with patch.object(
+                conn.dialect, "get_isolation_level",
+                Mock(side_effect=ProgrammingError("random error"))
+            ):
+                assert_raises(
+                    MySpecialException,
+                    conn.get_isolation_level
+                )
+
 
 class HandleInvalidatedOnConnectTest(fixtures.TestBase):
     __requires__ = ('sqlite', )
diff --git a/test/engine/test_transaction.py b/test/engine/test_transaction.py
index b7ad01408..0f5bb4cb5 100644
--- a/test/engine/test_transaction.py
+++ b/test/engine/test_transaction.py
@@ -1385,3 +1385,25 @@ class IsolationLevelTest(fixtures.TestBase):
             eng.dialect.get_isolation_level(conn.connection),
             self._non_default_isolation_level()
         )
+
+    def test_isolation_level_accessors_connection_default(self):
+        eng = create_engine(
+            testing.db.url
+        )
+        with eng.connect() as conn:
+            eq_(conn.default_isolation_level, self._default_isolation_level())
+        with eng.connect() as conn:
+            eq_(conn.get_isolation_level(), self._default_isolation_level())
+
+    def test_isolation_level_accessors_connection_option_modified(self):
+        eng = create_engine(
+            testing.db.url
+        )
+        with eng.connect() as conn:
+            c2 = conn.execution_options(
+                isolation_level=self._non_default_isolation_level())
+            eq_(conn.default_isolation_level, self._default_isolation_level())
+            eq_(conn.get_isolation_level(),
+                self._non_default_isolation_level())
+            eq_(c2.get_isolation_level(), self._non_default_isolation_level())
+