summaryrefslogtreecommitdiff
path: root/test/engine/test_reconnect.py
diff options
context:
space:
mode:
authorMike Bayer <mike_mp@zzzcomputing.com>2014-09-26 14:55:36 -0400
committerMike Bayer <mike_mp@zzzcomputing.com>2014-09-26 14:55:36 -0400
commitfbddf193a684ffe660c94c28e4c26e187111b21c (patch)
tree77e44889c1d9344ac84952bff6a18cc8e5e8dc1a /test/engine/test_reconnect.py
parentb1a956d4210c2bb06051a4a8b0d2e75d7c471ecd (diff)
downloadsqlalchemy-fbddf193a684ffe660c94c28e4c26e187111b21c.tar.gz
- Fixed bug where a "branched" connection, that is the kind you get
when you call :meth:`.Connection.connect`, would not share invalidation status with the parent. The architecture of branching has been tweaked a bit so that the branched connection defers to the parent for all invalidation status and operations. fixes #3215
Diffstat (limited to 'test/engine/test_reconnect.py')
-rw-r--r--test/engine/test_reconnect.py32
1 files changed, 32 insertions, 0 deletions
diff --git a/test/engine/test_reconnect.py b/test/engine/test_reconnect.py
index c82cca5a1..26a607301 100644
--- a/test/engine/test_reconnect.py
+++ b/test/engine/test_reconnect.py
@@ -504,6 +504,38 @@ class RealReconnectTest(fixtures.TestBase):
# pool isn't replaced
assert self.engine.pool is p2
+ def test_branched_invalidate_branch_to_parent(self):
+ c1 = self.engine.connect()
+
+ c1_branch = c1.connect()
+ eq_(c1_branch.execute(select([1])).scalar(), 1)
+
+ self.engine.test_shutdown()
+
+ _assert_invalidated(c1_branch.execute, select([1]))
+ assert c1.invalidated
+ assert c1_branch.invalidated
+
+ c1_branch._revalidate_connection()
+ assert not c1.invalidated
+ assert not c1_branch.invalidated
+
+ def test_branched_invalidate_parent_to_branch(self):
+ c1 = self.engine.connect()
+
+ c1_branch = c1.connect()
+ eq_(c1_branch.execute(select([1])).scalar(), 1)
+
+ self.engine.test_shutdown()
+
+ _assert_invalidated(c1.execute, select([1]))
+ assert c1.invalidated
+ assert c1_branch.invalidated
+
+ c1._revalidate_connection()
+ assert not c1.invalidated
+ assert not c1_branch.invalidated
+
def test_ensure_is_disconnect_gets_connection(self):
def is_disconnect(e, conn, cursor):
# connection is still present