diff options
author | Mike Bayer <mike_mp@zzzcomputing.com> | 2021-04-17 12:53:22 -0400 |
---|---|---|
committer | Mike Bayer <mike_mp@zzzcomputing.com> | 2021-04-17 12:54:58 -0400 |
commit | 7ed092df5c4a16990942333a8a43743d02ce18ff (patch) | |
tree | 1625716d7dd68c80fd019eaadfc187b1c0054a8c /tests/test_environment.py | |
parent | dd9c4d695c4463c872b25f59b6a8742bbc047150 (diff) | |
download | alembic-7ed092df5c4a16990942333a8a43743d02ce18ff.tar.gz |
Ensure proxy transaction still present on exit before closing
Fixed regression caused by the SQLAlchemy 1.4/2.0 compatibility switch
where calling ``.rollback()`` or ``.commit()`` explicitly within the
``context.begin_transaction()`` context manager would cause it to fail when
the block ended, as it did not expect that the transaction was manually
closed.
Change-Id: I4c5ba368bfd480d186276cd75edaac5019130bc6
Fixes: #829
Diffstat (limited to 'tests/test_environment.py')
-rw-r--r-- | tests/test_environment.py | 28 |
1 files changed, 28 insertions, 0 deletions
diff --git a/tests/test_environment.py b/tests/test_environment.py index e90d6fb..23e23b3 100644 --- a/tests/test_environment.py +++ b/tests/test_environment.py @@ -257,6 +257,34 @@ class MigrationTransactionTest(TestBase): assert_raises(Exception, go) is_false(self.conn.in_transaction()) + def test_proxy_transaction_contextmanager_explicit_rollback(self): + context = self._fixture( + {"transaction_per_migration": True, "transactional_ddl": True} + ) + proxy = context.begin_transaction(_per_migration=True) + is_true(self.conn.in_transaction()) + + with proxy: + is_true(self.conn.in_transaction()) + proxy.rollback() + is_false(self.conn.in_transaction()) + + is_false(self.conn.in_transaction()) + + def test_proxy_transaction_contextmanager_explicit_commit(self): + context = self._fixture( + {"transaction_per_migration": True, "transactional_ddl": True} + ) + proxy = context.begin_transaction(_per_migration=True) + is_true(self.conn.in_transaction()) + + with proxy: + is_true(self.conn.in_transaction()) + proxy.commit() + is_false(self.conn.in_transaction()) + + is_false(self.conn.in_transaction()) + def test_transaction_per_migration_transactional_ddl(self): context = self._fixture( {"transaction_per_migration": True, "transactional_ddl": True} |