From: Mike Bayer <mike_mp@zzzcomputing.com>
Date: Fri, 11 Dec 2020 20:33:22 +0000 (-0500)
Subject: Emit 2.0 deprecation warning for sub-transactions
X-Git-Tag: rel_1_4_0b2~102^2
X-Git-Url: http://git.ipfire.org/?a=commitdiff_plain;h=628db9581c5bd43c01cbfcc16753477f206de431;p=thirdparty%2Fsqlalchemy%2Fsqlalchemy.git

Emit 2.0 deprecation warning for sub-transactions

The nesting pattern will be removed in 2.0, so the use of the
MarkerTransaction should emit a 2.0 deprecation warning
unconditionally.

Change-Id: I96aed22c4e5db9b59e9b28a7f2d1283cd99a9cb6
---

diff --git a/lib/sqlalchemy/engine/base.py b/lib/sqlalchemy/engine/base.py
index 028af9fbb7..d36cf30e9a 100644
--- a/lib/sqlalchemy/engine/base.py
+++ b/lib/sqlalchemy/engine/base.py
@@ -489,6 +489,7 @@ class Connection(Connectable):
                 code="8s2b",
             )
         else:
+            assert not self._is_future
             raise exc.PendingRollbackError(
                 "This connection is on an inactive %stransaction.  "
                 "Please rollback() fully before proceeding."
@@ -2190,6 +2191,15 @@ class MarkerTransaction(Transaction):
                 "Please issue a rollback first."
             )
 
+        assert not connection._is_future
+        util.warn_deprecated_20(
+            "Calling .begin() when a transaction is already begun, creating "
+            "a 'sub' transaction, is deprecated "
+            "and will be removed in 2.0.  See the documentation section "
+            "'Migrating from the nesting pattern' for background on how "
+            "to migrate from this pattern."
+        )
+
         self.connection = connection
         if connection._nested_transaction is not None:
             self._transaction = connection._nested_transaction
diff --git a/test/dialect/test_sqlite.py b/test/dialect/test_sqlite.py
index 12200f832f..16969467ed 100644
--- a/test/dialect/test_sqlite.py
+++ b/test/dialect/test_sqlite.py
@@ -2459,28 +2459,6 @@ class SavepointTest(fixtures.TablesTest):
         )
         connection.close()
 
-    def test_rollback_to_subtransaction(self):
-        users = self.tables.users
-        connection = self.bind.connect()
-        transaction = connection.begin()
-        connection.execute(users.insert(), user_id=1, user_name="user1")
-        trans2 = connection.begin_nested()
-        connection.execute(users.insert(), user_id=2, user_name="user2")
-        trans3 = connection.begin()
-        connection.execute(users.insert(), user_id=3, user_name="user3")
-        trans3.rollback()
-
-        trans2.rollback()
-        connection.execute(users.insert(), user_id=4, user_name="user4")
-        transaction.commit()
-        eq_(
-            connection.execute(
-                select(users.c.user_id).order_by(users.c.user_id)
-            ).fetchall(),
-            [(1,), (4,)],
-        )
-        connection.close()
-
 
 class TypeReflectionTest(fixtures.TestBase):
 
diff --git a/test/engine/test_deprecations.py b/test/engine/test_deprecations.py
index 47e59b55da..6acc2967d8 100644
--- a/test/engine/test_deprecations.py
+++ b/test/engine/test_deprecations.py
@@ -425,6 +425,357 @@ class TransactionTest(fixtures.TablesTest):
         with testing.db.connect() as conn:
             eq_(conn.execute(users.select()).fetchall(), [(1, "user1")])
 
+    def test_begin_begin_rollback_rollback(self):
+        with testing.db.connect() as connection:
+            trans = connection.begin()
+            with testing.expect_deprecated_20(
+                r"Calling .begin\(\) when a transaction is already "
+                "begun, creating a 'sub' transaction"
+            ):
+                trans2 = connection.begin()
+            assert connection.connection._reset_agent is trans
+            trans2.rollback()
+            assert connection.connection._reset_agent is None
+            trans.rollback()
+            assert connection.connection._reset_agent is None
+
+    def test_begin_begin_commit_commit(self):
+        with testing.db.connect() as connection:
+            trans = connection.begin()
+            with testing.expect_deprecated_20(
+                r"Calling .begin\(\) when a transaction is already "
+                "begun, creating a 'sub' transaction"
+            ):
+                trans2 = connection.begin()
+            assert connection.connection._reset_agent is trans
+            trans2.commit()
+            assert connection.connection._reset_agent is trans
+            trans.commit()
+            assert connection.connection._reset_agent is None
+
+    def test_branch_nested_rollback(self, local_connection):
+        connection = local_connection
+        users = self.tables.users
+        connection.begin()
+        branched = connection.connect()
+        assert branched.in_transaction()
+        branched.execute(users.insert(), user_id=1, user_name="user1")
+        with testing.expect_deprecated_20(
+            r"Calling .begin\(\) when a transaction is already "
+            "begun, creating a 'sub' transaction"
+        ):
+            nested = branched.begin()
+        branched.execute(users.insert(), user_id=2, user_name="user2")
+        nested.rollback()
+        assert not connection.in_transaction()
+
+        assert_raises_message(
+            exc.InvalidRequestError,
+            "This connection is on an inactive transaction.  Please",
+            connection.exec_driver_sql,
+            "select 1",
+        )
+
+    @testing.requires.savepoints
+    def test_savepoint_cancelled_by_toplevel_marker(self, local_connection):
+        conn = local_connection
+        users = self.tables.users
+        trans = conn.begin()
+        conn.execute(users.insert(), {"user_id": 1, "user_name": "name"})
+
+        with testing.expect_deprecated_20(
+            r"Calling .begin\(\) when a transaction is already "
+            "begun, creating a 'sub' transaction"
+        ):
+            mk1 = conn.begin()
+
+        sp1 = conn.begin_nested()
+        conn.execute(users.insert(), {"user_id": 2, "user_name": "name2"})
+
+        mk1.rollback()
+
+        assert not sp1.is_active
+        assert not trans.is_active
+        assert conn._transaction is trans
+        assert conn._nested_transaction is None
+
+        with testing.db.connect() as conn:
+            eq_(
+                conn.scalar(select(func.count(1)).select_from(users)),
+                0,
+            )
+
+    @testing.requires.savepoints
+    def test_rollback_to_subtransaction(self, local_connection):
+        connection = local_connection
+        users = self.tables.users
+        transaction = connection.begin()
+        connection.execute(users.insert(), user_id=1, user_name="user1")
+        trans2 = connection.begin_nested()
+        connection.execute(users.insert(), user_id=2, user_name="user2")
+
+        with testing.expect_deprecated_20(
+            r"Calling .begin\(\) when a transaction is already "
+            "begun, creating a 'sub' transaction"
+        ):
+            trans3 = connection.begin()
+        connection.execute(users.insert(), user_id=3, user_name="user3")
+        trans3.rollback()
+
+        assert_raises_message(
+            exc.InvalidRequestError,
+            "This connection is on an inactive savepoint transaction.",
+            connection.exec_driver_sql,
+            "select 1",
+        )
+        trans2.rollback()
+        assert connection._nested_transaction is None
+
+        connection.execute(users.insert(), user_id=4, user_name="user4")
+        transaction.commit()
+        eq_(
+            connection.execute(
+                select(users.c.user_id).order_by(users.c.user_id)
+            ).fetchall(),
+            [(1,), (4,)],
+        )
+
+    # PG emergency shutdown:
+    # select * from pg_prepared_xacts
+    # ROLLBACK PREPARED '<xid>'
+    # MySQL emergency shutdown:
+    # for arg in `mysql -u root -e "xa recover" | cut -c 8-100 |
+    #     grep sa`; do mysql -u root -e "xa rollback '$arg'"; done
+    @testing.requires.skip_mysql_on_windows
+    @testing.requires.two_phase_transactions
+    @testing.requires.savepoints
+    def test_mixed_two_phase_transaction(self, local_connection):
+        connection = local_connection
+        users = self.tables.users
+        transaction = connection.begin_twophase()
+        connection.execute(users.insert(), user_id=1, user_name="user1")
+        with testing.expect_deprecated_20(
+            r"Calling .begin\(\) when a transaction is already "
+            "begun, creating a 'sub' transaction"
+        ):
+            transaction2 = connection.begin()
+        connection.execute(users.insert(), user_id=2, user_name="user2")
+        transaction3 = connection.begin_nested()
+        connection.execute(users.insert(), user_id=3, user_name="user3")
+        with testing.expect_deprecated_20(
+            r"Calling .begin\(\) when a transaction is already "
+            "begun, creating a 'sub' transaction"
+        ):
+            transaction4 = connection.begin()
+        connection.execute(users.insert(), user_id=4, user_name="user4")
+        transaction4.commit()
+        transaction3.rollback()
+        connection.execute(users.insert(), user_id=5, user_name="user5")
+        transaction2.commit()
+        transaction.prepare()
+        transaction.commit()
+        eq_(
+            connection.execute(
+                select(users.c.user_id).order_by(users.c.user_id)
+            ).fetchall(),
+            [(1,), (2,), (5,)],
+        )
+
+    @testing.requires.savepoints
+    def test_inactive_due_to_subtransaction_on_nested_no_commit(
+        self, local_connection
+    ):
+        connection = local_connection
+        trans = connection.begin()
+
+        nested = connection.begin_nested()
+
+        with testing.expect_deprecated_20(
+            r"Calling .begin\(\) when a transaction is already "
+            "begun, creating a 'sub' transaction"
+        ):
+            trans2 = connection.begin()
+        trans2.rollback()
+
+        assert_raises_message(
+            exc.InvalidRequestError,
+            "This connection is on an inactive savepoint transaction.  "
+            "Please rollback",
+            nested.commit,
+        )
+        trans.commit()
+
+        assert_raises_message(
+            exc.InvalidRequestError,
+            "This nested transaction is inactive",
+            nested.commit,
+        )
+
+    def test_close(self, local_connection):
+        connection = local_connection
+        users = self.tables.users
+        transaction = connection.begin()
+        connection.execute(users.insert(), user_id=1, user_name="user1")
+        connection.execute(users.insert(), user_id=2, user_name="user2")
+        connection.execute(users.insert(), user_id=3, user_name="user3")
+        with testing.expect_deprecated_20(
+            r"Calling .begin\(\) when a transaction is already "
+            "begun, creating a 'sub' transaction"
+        ):
+            trans2 = connection.begin()
+        connection.execute(users.insert(), user_id=4, user_name="user4")
+        connection.execute(users.insert(), user_id=5, user_name="user5")
+        assert connection.in_transaction()
+        trans2.close()
+        assert connection.in_transaction()
+        transaction.commit()
+        assert not connection.in_transaction()
+        self.assert_(
+            connection.exec_driver_sql(
+                "select count(*) from " "users"
+            ).scalar()
+            == 5
+        )
+        result = connection.exec_driver_sql("select * from users")
+        assert len(result.fetchall()) == 5
+
+    def test_close2(self, local_connection):
+        connection = local_connection
+        users = self.tables.users
+        transaction = connection.begin()
+        connection.execute(users.insert(), user_id=1, user_name="user1")
+        connection.execute(users.insert(), user_id=2, user_name="user2")
+        connection.execute(users.insert(), user_id=3, user_name="user3")
+        with testing.expect_deprecated_20(
+            r"Calling .begin\(\) when a transaction is already "
+            "begun, creating a 'sub' transaction"
+        ):
+            trans2 = connection.begin()
+        connection.execute(users.insert(), user_id=4, user_name="user4")
+        connection.execute(users.insert(), user_id=5, user_name="user5")
+        assert connection.in_transaction()
+        trans2.close()
+        assert connection.in_transaction()
+        transaction.close()
+        assert not connection.in_transaction()
+        self.assert_(
+            connection.exec_driver_sql(
+                "select count(*) from " "users"
+            ).scalar()
+            == 0
+        )
+        result = connection.exec_driver_sql("select * from users")
+        assert len(result.fetchall()) == 0
+
+    def test_inactive_due_to_subtransaction_no_commit(self, local_connection):
+        connection = local_connection
+        trans = connection.begin()
+        with testing.expect_deprecated_20(
+            r"Calling .begin\(\) when a transaction is already "
+            "begun, creating a 'sub' transaction"
+        ):
+            trans2 = connection.begin()
+        trans2.rollback()
+        assert_raises_message(
+            exc.InvalidRequestError,
+            "This connection is on an inactive transaction.  Please rollback",
+            trans.commit,
+        )
+
+        trans.rollback()
+
+        assert_raises_message(
+            exc.InvalidRequestError,
+            "This transaction is inactive",
+            trans.commit,
+        )
+
+    def test_nested_rollback(self, local_connection):
+        connection = local_connection
+        users = self.tables.users
+        try:
+            transaction = connection.begin()
+            try:
+                connection.execute(
+                    users.insert(), user_id=1, user_name="user1"
+                )
+                connection.execute(
+                    users.insert(), user_id=2, user_name="user2"
+                )
+                connection.execute(
+                    users.insert(), user_id=3, user_name="user3"
+                )
+                with testing.expect_deprecated_20(
+                    r"Calling .begin\(\) when a transaction is already "
+                    "begun, creating a 'sub' transaction"
+                ):
+                    trans2 = connection.begin()
+                try:
+                    connection.execute(
+                        users.insert(), user_id=4, user_name="user4"
+                    )
+                    connection.execute(
+                        users.insert(), user_id=5, user_name="user5"
+                    )
+                    raise Exception("uh oh")
+                    trans2.commit()
+                except Exception:
+                    trans2.rollback()
+                    raise
+                transaction.rollback()
+            except Exception:
+                transaction.rollback()
+                raise
+        except Exception as e:
+            # and not "This transaction is inactive"
+            # comment moved here to fix pep8
+            assert str(e) == "uh oh"
+        else:
+            assert False
+
+    def test_nesting(self, local_connection):
+        connection = local_connection
+        users = self.tables.users
+        transaction = connection.begin()
+        connection.execute(users.insert(), user_id=1, user_name="user1")
+        connection.execute(users.insert(), user_id=2, user_name="user2")
+        connection.execute(users.insert(), user_id=3, user_name="user3")
+        with testing.expect_deprecated_20(
+            r"Calling .begin\(\) when a transaction is already "
+            "begun, creating a 'sub' transaction"
+        ):
+            trans2 = connection.begin()
+        connection.execute(users.insert(), user_id=4, user_name="user4")
+        connection.execute(users.insert(), user_id=5, user_name="user5")
+        trans2.commit()
+        transaction.rollback()
+        self.assert_(
+            connection.exec_driver_sql(
+                "select count(*) from " "users"
+            ).scalar()
+            == 0
+        )
+        result = connection.exec_driver_sql("select * from users")
+        assert len(result.fetchall()) == 0
+
+    def test_no_marker_on_inactive_trans(self, local_connection):
+        conn = local_connection
+        conn.begin()
+
+        with testing.expect_deprecated_20(
+            r"Calling .begin\(\) when a transaction is already "
+            "begun, creating a 'sub' transaction"
+        ):
+            mk1 = conn.begin()
+
+        mk1.rollback()
+
+        assert_raises_message(
+            exc.InvalidRequestError,
+            "the current transaction on this connection is inactive.",
+            conn.begin,
+        )
+
     def test_implicit_autocommit_compiled(self):
         users = self.tables.users
 
diff --git a/test/engine/test_transaction.py b/test/engine/test_transaction.py
index 4db5a745ad..79126fc5bb 100644
--- a/test/engine/test_transaction.py
+++ b/test/engine/test_transaction.py
@@ -93,148 +93,6 @@ class TransactionTest(fixtures.TablesTest):
         result = connection.exec_driver_sql("select * from users")
         assert len(result.fetchall()) == 0
 
-    def test_nested_rollback(self, local_connection):
-        connection = local_connection
-        users = self.tables.users
-        try:
-            transaction = connection.begin()
-            try:
-                connection.execute(
-                    users.insert(), user_id=1, user_name="user1"
-                )
-                connection.execute(
-                    users.insert(), user_id=2, user_name="user2"
-                )
-                connection.execute(
-                    users.insert(), user_id=3, user_name="user3"
-                )
-                trans2 = connection.begin()
-                try:
-                    connection.execute(
-                        users.insert(), user_id=4, user_name="user4"
-                    )
-                    connection.execute(
-                        users.insert(), user_id=5, user_name="user5"
-                    )
-                    raise Exception("uh oh")
-                    trans2.commit()
-                except Exception:
-                    trans2.rollback()
-                    raise
-                transaction.rollback()
-            except Exception:
-                transaction.rollback()
-                raise
-        except Exception as e:
-            # and not "This transaction is inactive"
-            # comment moved here to fix pep8
-            assert str(e) == "uh oh"
-        else:
-            assert False
-
-    def test_branch_nested_rollback(self, local_connection):
-        connection = local_connection
-        users = self.tables.users
-        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()
-
-        assert_raises_message(
-            exc.InvalidRequestError,
-            "This connection is on an inactive transaction.  Please",
-            connection.exec_driver_sql,
-            "select 1",
-        )
-
-    def test_no_marker_on_inactive_trans(self, local_connection):
-        conn = local_connection
-        conn.begin()
-
-        mk1 = conn.begin()
-
-        mk1.rollback()
-
-        assert_raises_message(
-            exc.InvalidRequestError,
-            "the current transaction on this connection is inactive.",
-            conn.begin,
-        )
-
-    @testing.requires.savepoints
-    def test_savepoint_cancelled_by_toplevel_marker(self, local_connection):
-        conn = local_connection
-        users = self.tables.users
-        trans = conn.begin()
-        conn.execute(users.insert(), {"user_id": 1, "user_name": "name"})
-
-        mk1 = conn.begin()
-
-        sp1 = conn.begin_nested()
-        conn.execute(users.insert(), {"user_id": 2, "user_name": "name2"})
-
-        mk1.rollback()
-
-        assert not sp1.is_active
-        assert not trans.is_active
-        assert conn._transaction is trans
-        assert conn._nested_transaction is None
-
-        with testing.db.connect() as conn:
-            eq_(
-                conn.scalar(select(func.count(1)).select_from(users)),
-                0,
-            )
-
-    def test_inactive_due_to_subtransaction_no_commit(self, local_connection):
-        connection = local_connection
-        trans = connection.begin()
-        trans2 = connection.begin()
-        trans2.rollback()
-        assert_raises_message(
-            exc.InvalidRequestError,
-            "This connection is on an inactive transaction.  Please rollback",
-            trans.commit,
-        )
-
-        trans.rollback()
-
-        assert_raises_message(
-            exc.InvalidRequestError,
-            "This transaction is inactive",
-            trans.commit,
-        )
-
-    @testing.requires.savepoints
-    def test_inactive_due_to_subtransaction_on_nested_no_commit(
-        self, local_connection
-    ):
-        connection = local_connection
-        trans = connection.begin()
-
-        nested = connection.begin_nested()
-
-        trans2 = connection.begin()
-        trans2.rollback()
-
-        assert_raises_message(
-            exc.InvalidRequestError,
-            "This connection is on an inactive savepoint transaction.  "
-            "Please rollback",
-            nested.commit,
-        )
-        trans.commit()
-
-        assert_raises_message(
-            exc.InvalidRequestError,
-            "This nested transaction is inactive",
-            nested.commit,
-        )
-
     def test_deactivated_warning_ctxmanager(self, local_connection):
         with expect_warnings(
             "transaction already deassociated from connection"
@@ -394,27 +252,6 @@ class TransactionTest(fixtures.TablesTest):
             0,
         )
 
-    def test_nesting(self, local_connection):
-        connection = local_connection
-        users = self.tables.users
-        transaction = connection.begin()
-        connection.execute(users.insert(), user_id=1, user_name="user1")
-        connection.execute(users.insert(), user_id=2, user_name="user2")
-        connection.execute(users.insert(), user_id=3, user_name="user3")
-        trans2 = connection.begin()
-        connection.execute(users.insert(), user_id=4, user_name="user4")
-        connection.execute(users.insert(), user_id=5, user_name="user5")
-        trans2.commit()
-        transaction.rollback()
-        self.assert_(
-            connection.exec_driver_sql(
-                "select count(*) from " "users"
-            ).scalar()
-            == 0
-        )
-        result = connection.exec_driver_sql("select * from users")
-        assert len(result.fetchall()) == 0
-
     def test_with_interface(self, local_connection):
         connection = local_connection
         users = self.tables.users
@@ -452,22 +289,11 @@ class TransactionTest(fixtures.TablesTest):
         connection.execute(users.insert(), user_id=1, user_name="user1")
         connection.execute(users.insert(), user_id=2, user_name="user2")
         connection.execute(users.insert(), user_id=3, user_name="user3")
-        trans2 = connection.begin()
-        connection.execute(users.insert(), user_id=4, user_name="user4")
-        connection.execute(users.insert(), user_id=5, user_name="user5")
-        assert connection.in_transaction()
-        trans2.close()
         assert connection.in_transaction()
         transaction.commit()
         assert not connection.in_transaction()
-        self.assert_(
-            connection.exec_driver_sql(
-                "select count(*) from " "users"
-            ).scalar()
-            == 5
-        )
         result = connection.exec_driver_sql("select * from users")
-        assert len(result.fetchall()) == 5
+        eq_(len(result.fetchall()), 3)
 
     def test_close2(self, local_connection):
         connection = local_connection
@@ -476,20 +302,9 @@ class TransactionTest(fixtures.TablesTest):
         connection.execute(users.insert(), user_id=1, user_name="user1")
         connection.execute(users.insert(), user_id=2, user_name="user2")
         connection.execute(users.insert(), user_id=3, user_name="user3")
-        trans2 = connection.begin()
-        connection.execute(users.insert(), user_id=4, user_name="user4")
-        connection.execute(users.insert(), user_id=5, user_name="user5")
-        assert connection.in_transaction()
-        trans2.close()
         assert connection.in_transaction()
         transaction.close()
         assert not connection.in_transaction()
-        self.assert_(
-            connection.exec_driver_sql(
-                "select count(*) from " "users"
-            ).scalar()
-            == 0
-        )
         result = connection.exec_driver_sql("select * from users")
         assert len(result.fetchall()) == 0
 
@@ -529,37 +344,6 @@ class TransactionTest(fixtures.TablesTest):
             [(1,), (2,), (3,)],
         )
 
-    @testing.requires.savepoints
-    def test_rollback_to_subtransaction(self, local_connection):
-        connection = local_connection
-        users = self.tables.users
-        transaction = connection.begin()
-        connection.execute(users.insert(), user_id=1, user_name="user1")
-        trans2 = connection.begin_nested()
-        connection.execute(users.insert(), user_id=2, user_name="user2")
-
-        trans3 = connection.begin()
-        connection.execute(users.insert(), user_id=3, user_name="user3")
-        trans3.rollback()
-
-        assert_raises_message(
-            exc.InvalidRequestError,
-            "This connection is on an inactive savepoint transaction.",
-            connection.exec_driver_sql,
-            "select 1",
-        )
-        trans2.rollback()
-        assert connection._nested_transaction is None
-
-        connection.execute(users.insert(), user_id=4, user_name="user4")
-        transaction.commit()
-        eq_(
-            connection.execute(
-                select(users.c.user_id).order_by(users.c.user_id)
-            ).fetchall(),
-            [(1,), (4,)],
-        )
-
     @testing.requires.two_phase_transactions
     def test_two_phase_transaction(self, local_connection):
         connection = local_connection
@@ -587,39 +371,6 @@ class TransactionTest(fixtures.TablesTest):
             [(1,), (2,)],
         )
 
-    # PG emergency shutdown:
-    # select * from pg_prepared_xacts
-    # ROLLBACK PREPARED '<xid>'
-    # MySQL emergency shutdown:
-    # for arg in `mysql -u root -e "xa recover" | cut -c 8-100 |
-    #     grep sa`; do mysql -u root -e "xa rollback '$arg'"; done
-    @testing.requires.skip_mysql_on_windows
-    @testing.requires.two_phase_transactions
-    @testing.requires.savepoints
-    def test_mixed_two_phase_transaction(self, local_connection):
-        connection = local_connection
-        users = self.tables.users
-        transaction = connection.begin_twophase()
-        connection.execute(users.insert(), user_id=1, user_name="user1")
-        transaction2 = connection.begin()
-        connection.execute(users.insert(), user_id=2, user_name="user2")
-        transaction3 = connection.begin_nested()
-        connection.execute(users.insert(), user_id=3, user_name="user3")
-        transaction4 = connection.begin()
-        connection.execute(users.insert(), user_id=4, user_name="user4")
-        transaction4.commit()
-        transaction3.rollback()
-        connection.execute(users.insert(), user_id=5, user_name="user5")
-        transaction2.commit()
-        transaction.prepare()
-        transaction.commit()
-        eq_(
-            connection.execute(
-                select(users.c.user_id).order_by(users.c.user_id)
-            ).fetchall(),
-            [(1,), (2,), (5,)],
-        )
-
     @testing.requires.two_phase_transactions
     @testing.requires.two_phase_recovery
     def test_two_phase_recover(self):
@@ -862,26 +613,6 @@ class ResetAgentTest(fixtures.TestBase):
             trans.rollback()
             assert connection.connection._reset_agent is None
 
-    def test_begin_begin_rollback_rollback(self):
-        with testing.db.connect() as connection:
-            trans = connection.begin()
-            trans2 = connection.begin()
-            assert connection.connection._reset_agent is trans
-            trans2.rollback()
-            assert connection.connection._reset_agent is None
-            trans.rollback()
-            assert connection.connection._reset_agent is None
-
-    def test_begin_begin_commit_commit(self):
-        with testing.db.connect() as connection:
-            trans = connection.begin()
-            trans2 = connection.begin()
-            assert connection.connection._reset_agent is trans
-            trans2.commit()
-            assert connection.connection._reset_agent is trans
-            trans.commit()
-            assert connection.connection._reset_agent is None
-
     @testing.requires.two_phase_transactions
     def test_reset_via_agent_begin_twophase(self):
         with testing.db.connect() as connection:
diff --git a/test/ext/asyncio/test_engine_py3k.py b/test/ext/asyncio/test_engine_py3k.py
index 6df8a0e006..cd1e16ed91 100644
--- a/test/ext/asyncio/test_engine_py3k.py
+++ b/test/ext/asyncio/test_engine_py3k.py
@@ -46,14 +46,10 @@ class EngineFixture(fixtures.TablesTest):
     @classmethod
     def insert_data(cls, connection):
         users = cls.tables.users
-        with connection.begin():
-            connection.execute(
-                users.insert(),
-                [
-                    {"user_id": i, "user_name": "name%d" % i}
-                    for i in range(1, 20)
-                ],
-            )
+        connection.execute(
+            users.insert(),
+            [{"user_id": i, "user_name": "name%d" % i} for i in range(1, 20)],
+        )
 
 
 class AsyncEngineTest(EngineFixture):