Detect open transaction at end of migration w/ transactional_ddl=False

A :class:`.CommandError` is now raised when a migration file opens
a database transaction and does not close/commit/rollback, when
the backend database or environment options also specify transactional_ddl
is False.   When transactional_ddl is not in use, Alembic doesn't
close any transaction so a transaction opened by a migration file
will cause the following migrations to fail to apply.

Change-Id: I7a9baf18837deb193d9ddc6813955909484d4945
Fixes: #369

3 files changed, 133 insertions, 3 deletions

diff --git a/alembic/runtime/migration.py b/alembic/runtime/migration.py
index 33cd74f..0f1d88e 100644
--- a/alembic/runtime/migration.py
+++ b/alembic/runtime/migration.py
@@ -67,7 +67,6 @@ class MigrationContext(object):
         self.script = opts.get('script')
         as_sql = opts.get('as_sql', False)
         transactional_ddl = opts.get("transactional_ddl")
-
         self._transaction_per_migration = opts.get(
             "transaction_per_migration", False)
 
@@ -327,6 +326,14 @@ class MigrationContext(object):
                 # just to run the operations on every version
                 head_maintainer.update_to_step(step)
 
+            if not self.as_sql and not self.impl.transactional_ddl and \
+                    self.connection.in_transaction():
+                raise util.CommandError(
+                    "Migration \"%s\" has left an uncommitted "
+                    "transaction opened; transactional_ddl is False so "
+                    "Alembic is not committing transactions"
+                    % step)
+
         if self.as_sql and not head_maintainer.heads:
             self._version.drop(self.connection)
 
diff --git a/docs/build/changelog.rst b/docs/build/changelog.rst
index 5552448..a739ff3 100644
--- a/docs/build/changelog.rst
+++ b/docs/build/changelog.rst
@@ -7,6 +7,17 @@ Changelog
     :version: 0.9.0
     :released:
 
+    .. change:: 369
+      :tags: bug, commands
+      :tickets: 369
+
+      A :class:`.CommandError` is now raised when a migration file opens
+      a database transaction and does not close/commit/rollback, when
+      the backend database or environment options also specify transactional_ddl
+      is False.   When transactional_ddl is not in use, Alembic doesn't
+      close any transaction so a transaction opened by a migration file
+      will cause the following migrations to fail to apply.
+
     .. change:: 413
       :tags: bug, autogenerate, mysql
       :tickets: 413
diff --git a/tests/test_script_consumption.py b/tests/test_script_consumption.py
index a86430f..6a22958 100644
--- a/tests/test_script_consumption.py
+++ b/tests/test_script_consumption.py
@@ -11,7 +11,9 @@ from alembic.testing.env import clear_staging_env, staging_env, \
     three_rev_fixture, _no_sql_testing_config
 from alembic.testing import eq_, assert_raises_message
 from alembic.testing.fixtures import TestBase, capture_context_buffer
-
+from alembic.environment import EnvironmentContext
+from contextlib import contextmanager
+from alembic.testing import mock
 
 class ApplyVersionsFunctionalTest(TestBase):
     __only_on__ = 'sqlite'
@@ -131,7 +133,7 @@ class SourcelessApplyVersionsTest(ApplyVersionsFunctionalTest):
     sourceless = True
 
 
-class TransactionalDDLTest(TestBase):
+class OfflineTransactionalDDLTest(TestBase):
     def setUp(self):
         self.env = staging_env()
         self.cfg = cfg = _no_sql_testing_config()
@@ -170,6 +172,116 @@ class TransactionalDDLTest(TestBase):
             buf.getvalue(), re.S)
 
 
+class OnlineTransactionalDDLTest(TestBase):
+    def tearDown(self):
+        clear_staging_env()
+
+    def _opened_transaction_fixture(self):
+        self.env = staging_env()
+        self.cfg = _sqlite_testing_config()
+
+        script = ScriptDirectory.from_config(self.cfg)
+        a = util.rev_id()
+        b = util.rev_id()
+        c = util.rev_id()
+        script.generate_revision(a, "revision a", refresh=True)
+        write_script(script, a, """
+"rev a"
+
+revision = '%s'
+down_revision = None
+
+def upgrade():
+    pass
+
+def downgrade():
+    pass
+
+""" % (a, ))
+        script.generate_revision(b, "revision b", refresh=True)
+        write_script(script, b, """
+"rev b"
+revision = '%s'
+down_revision = '%s'
+
+from alembic import op
+
+
+def upgrade():
+    conn = op.get_bind()
+    trans = conn.begin()
+
+
+def downgrade():
+    pass
+
+""" % (b, a))
+        script.generate_revision(c, "revision c", refresh=True)
+        write_script(script, c, """
+"rev c"
+revision = '%s'
+down_revision = '%s'
+
+from alembic import op
+
+
+def upgrade():
+    pass
+
+
+def downgrade():
+    pass
+
+""" % (c, b))
+        return a, b, c
+
+    @contextmanager
+    def _patch_environment(self, transactional_ddl, transaction_per_migration):
+        conf = EnvironmentContext.configure
+
+        def configure(*arg, **opt):
+            opt.update(
+                transactional_ddl=transactional_ddl,
+                transaction_per_migration=transaction_per_migration)
+            return conf(*arg, **opt)
+
+        with mock.patch.object(EnvironmentContext, "configure", configure):
+            yield
+
+    def test_raise_when_rev_leaves_open_transaction(self):
+        a, b, c = self._opened_transaction_fixture()
+
+        with self._patch_environment(
+                transactional_ddl=False, transaction_per_migration=False):
+            assert_raises_message(
+                util.CommandError,
+                r'Migration "upgrade .*, rev b" has left an uncommitted '
+                r'transaction opened; transactional_ddl is False so Alembic '
+                r'is not committing transactions',
+                command.upgrade, self.cfg, c
+            )
+
+    def test_raise_when_rev_leaves_open_transaction_tpm(self):
+        a, b, c = self._opened_transaction_fixture()
+
+        with self._patch_environment(
+                transactional_ddl=False, transaction_per_migration=True):
+            assert_raises_message(
+                util.CommandError,
+                r'Migration "upgrade .*, rev b" has left an uncommitted '
+                r'transaction opened; transactional_ddl is False so Alembic '
+                r'is not committing transactions',
+                command.upgrade, self.cfg, c
+            )
+
+    def test_noerr_rev_leaves_open_transaction_transactional_ddl(self):
+        a, b, c = self._opened_transaction_fixture()
+
+        with self._patch_environment(
+                transactional_ddl=True, transaction_per_migration=False):
+            command.upgrade(self.cfg, c)
+
+
 class EncodingTest(TestBase):
 
     def setUp(self):