- Fixed bug where a "branched" connection, that is the kind you get

when you call :meth:`.Connection.connect`, would not share transaction
status with the parent.  The architecture of branching has been tweaked
a bit so that the branched connection defers to the parent for
all transactional status and operations.
fixes #3190

3 files changed, 110 insertions, 10 deletions

diff --git a/test/engine/test_execute.py b/test/engine/test_execute.py
index e14a4fd2a..219a145c6 100644
--- a/test/engine/test_execute.py
+++ b/test/engine/test_execute.py
@@ -982,6 +982,17 @@ class ExecutionOptionsTest(fixtures.TestBase):
         eq_(c1._execution_options, {"foo": "bar"})
         eq_(c2._execution_options, {"foo": "bar", "bat": "hoho"})
 
+    def test_branched_connection_execution_options(self):
+        engine = testing_engine("sqlite://")
+
+        conn = engine.connect()
+        c2 = conn.execution_options(foo="bar")
+        c2_branch = c2.connect()
+        eq_(
+            c2_branch._execution_options,
+            {"foo": "bar"}
+        )
+
 
 class AlternateResultProxyTest(fixtures.TestBase):
     __requires__ = ('sqlite', )
diff --git a/test/engine/test_reconnect.py b/test/engine/test_reconnect.py
index 26a607301..4500ada6a 100644
--- a/test/engine/test_reconnect.py
+++ b/test/engine/test_reconnect.py
@@ -8,7 +8,7 @@ from sqlalchemy import testing
 from sqlalchemy.testing import engines
 from sqlalchemy.testing import fixtures
 from sqlalchemy.testing.engines import testing_engine
-from sqlalchemy.testing.mock import Mock, call
+from sqlalchemy.testing.mock import Mock, call, patch
 
 
 class MockError(Exception):
@@ -507,18 +507,21 @@ class RealReconnectTest(fixtures.TestBase):
     def test_branched_invalidate_branch_to_parent(self):
         c1 = self.engine.connect()
 
-        c1_branch = c1.connect()
-        eq_(c1_branch.execute(select([1])).scalar(), 1)
+        with patch.object(self.engine.pool, "logger") as logger:
+            c1_branch = c1.connect()
+            eq_(c1_branch.execute(select([1])).scalar(), 1)
 
-        self.engine.test_shutdown()
+            self.engine.test_shutdown()
 
-        _assert_invalidated(c1_branch.execute, select([1]))
-        assert c1.invalidated
-        assert c1_branch.invalidated
+            _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
+            c1_branch._revalidate_connection()
+            assert not c1.invalidated
+            assert not c1_branch.invalidated
+
+        assert "Invalidate connection" in logger.mock_calls[0][1][0]
 
     def test_branched_invalidate_parent_to_branch(self):
         c1 = self.engine.connect()
@@ -536,6 +539,19 @@ class RealReconnectTest(fixtures.TestBase):
         assert not c1.invalidated
         assert not c1_branch.invalidated
 
+    def test_branch_invalidate_state(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 not c1_branch.closed
+        assert not c1_branch._connection_is_valid
+
     def test_ensure_is_disconnect_gets_connection(self):
         def is_disconnect(e, conn, cursor):
             # connection is still present
diff --git a/test/engine/test_transaction.py b/test/engine/test_transaction.py
index d921e9ead..fbaf01db7 100644
--- a/test/engine/test_transaction.py
+++ b/test/engine/test_transaction.py
@@ -133,6 +133,79 @@ class TransactionTest(fixtures.TestBase):
             finally:
                 connection.close()
 
+    def test_branch_nested_rollback(self):
+        connection = testing.db.connect()
+        try:
+            connection.begin()
+            branched = connection.connect()
+            assert branched.in_transaction()
+            branched.execute(users.insert(), user_id=1, user_name='user1')
+            nested = branched.begin()
+            branched.execute(users.insert(), user_id=2, user_name='user2')
+            nested.rollback()
+            assert not connection.in_transaction()
+            eq_(connection.scalar("select count(*) from query_users"), 0)
+
+        finally:
+            connection.close()
+
+    def test_branch_orig_rollback(self):
+        connection = testing.db.connect()
+        try:
+            branched = connection.connect()
+            branched.execute(users.insert(), user_id=1, user_name='user1')
+            nested = branched.begin()
+            assert branched.in_transaction()
+            branched.execute(users.insert(), user_id=2, user_name='user2')
+            nested.rollback()
+            eq_(connection.scalar("select count(*) from query_users"), 1)
+
+        finally:
+            connection.close()
+
+    def test_branch_autocommit(self):
+        connection = testing.db.connect()
+        try:
+            branched = connection.connect()
+            branched.execute(users.insert(), user_id=1, user_name='user1')
+        finally:
+            connection.close()
+        eq_(testing.db.scalar("select count(*) from query_users"), 1)
+
+    @testing.requires.savepoints
+    def test_branch_savepoint_rollback(self):
+        connection = testing.db.connect()
+        try:
+            trans = connection.begin()
+            branched = connection.connect()
+            assert branched.in_transaction()
+            branched.execute(users.insert(), user_id=1, user_name='user1')
+            nested = branched.begin_nested()
+            branched.execute(users.insert(), user_id=2, user_name='user2')
+            nested.rollback()
+            assert connection.in_transaction()
+            trans.commit()
+            eq_(connection.scalar("select count(*) from query_users"), 1)
+
+        finally:
+            connection.close()
+
+    @testing.requires.two_phase_transactions
+    def test_branch_twophase_rollback(self):
+        connection = testing.db.connect()
+        try:
+            branched = connection.connect()
+            assert not branched.in_transaction()
+            branched.execute(users.insert(), user_id=1, user_name='user1')
+            nested = branched.begin_twophase()
+            branched.execute(users.insert(), user_id=2, user_name='user2')
+            nested.rollback()
+            assert not connection.in_transaction()
+            eq_(connection.scalar("select count(*) from query_users"), 1)
+
+        finally:
+            connection.close()
+
     def test_retains_through_options(self):
         connection = testing.db.connect()
         try: