summaryrefslogtreecommitdiff
path: root/tests/test_environment.py
diff options
context:
space:
mode:
authorMike Bayer <mike_mp@zzzcomputing.com>2021-04-17 12:53:22 -0400
committerMike Bayer <mike_mp@zzzcomputing.com>2021-04-17 12:54:58 -0400
commit7ed092df5c4a16990942333a8a43743d02ce18ff (patch)
tree1625716d7dd68c80fd019eaadfc187b1c0054a8c /tests/test_environment.py
parentdd9c4d695c4463c872b25f59b6a8742bbc047150 (diff)
downloadalembic-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.py28
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}