- Fixed bug in :class:`.Connection` and pool where the

:meth:`.Connection.invalidate` method, or an invalidation due
to a database disconnect, would fail if the
``isolation_level`` parameter had been used with
:meth:`.Connection.execution_options`; the "finalizer" that resets
the isolation level would be called on the no longer opened connection.
fixes #3302

2 files changed, 22 insertions, 0 deletions

diff --git a/test/engine/test_reconnect.py b/test/engine/test_reconnect.py
index 4500ada6a..619319693 100644
--- a/test/engine/test_reconnect.py
+++ b/test/engine/test_reconnect.py
@@ -5,6 +5,7 @@ from sqlalchemy import (
 from sqlalchemy.testing.schema import Table, Column
 import sqlalchemy as tsa
 from sqlalchemy import testing
+from sqlalchemy.testing import mock
 from sqlalchemy.testing import engines
 from sqlalchemy.testing import fixtures
 from sqlalchemy.testing.engines import testing_engine
@@ -211,6 +212,15 @@ class MockReconnectTest(fixtures.TestBase):
             [[call()], []]
         )
 
+    def test_invalidate_dont_call_finalizer(self):
+        conn = self.db.connect()
+        finalizer = mock.Mock()
+        conn.connection._connection_record.\
+            finalize_callback.append(finalizer)
+        conn.invalidate()
+        assert conn.invalidated
+        eq_(finalizer.call_count, 0)
+
     def test_conn_reusable(self):
         conn = self.db.connect()
 
diff --git a/test/engine/test_transaction.py b/test/engine/test_transaction.py
index b7d900917..b662c7fcd 100644
--- a/test/engine/test_transaction.py
+++ b/test/engine/test_transaction.py
@@ -1323,6 +1323,18 @@ class IsolationLevelTest(fixtures.TestBase):
             eng.connect
         )
 
+    def test_connection_invalidated(self):
+        eng = testing_engine()
+        conn = eng.connect()
+        c2 = conn.execution_options(
+            isolation_level=self._non_default_isolation_level())
+        c2.invalidate()
+        c2.connection
+
+        # TODO: do we want to rebuild the previous isolation?
+        # for now, this is current behavior so we will leave it.
+        eq_(c2.get_isolation_level(), self._default_isolation_level())
+
     def test_per_connection(self):
         from sqlalchemy.pool import QueuePool
         eng = testing_engine(