db: Don't rely on branched connections

We were previously calling 'connect()' on the 'connectable' object in
'run_migrations_online', regardless of whether it was an 'Engine' or
'Connection' object. This worked because, as noted in an inline comment,
"when connectable is already a Connection object, calling 'connect()'
gives us a *branched connection*." This is no longer the case. From the
SQLAlchemy docs [1]:

  The Connection object does not support "branching", which was a
  pattern by which a sub "connection" would be used that refers to this
  connection as a parent.

Update our code to reflect this change, using the newly updated example
from the SQLAlchemy cookbook doc [2] as inspiration.

[1] https://docs.sqlalchemy.org/en/14/core/future.html#sqlalchemy.future.Connection
[2] https://alembic.sqlalchemy.org/en/latest/cookbook.html#connection-sharing

Signed-off-by: Stephen Finucane <stephenfin@redhat.com>
Change-Id: Id41ae6d3fa3f343efdb727aae9bdef8ad2a098fd

heat/db/migrations/env.py

@@ -74,13 +74,17 @@ def run_migrations_online() -> None:
             prefix="sqlalchemy.",
             poolclass=pool.NullPool,
         )
+        with connectable.connect() as connection:
+            context.configure(
+                connection=connection,
+                target_metadata=target_metadata,
+            )
 
-    # when connectable is already a Connection object, calling connect() gives
-    # us a *branched connection*
-
-    with connectable.connect() as connection:
+            with context.begin_transaction():
+                context.run_migrations()
+    else:
         context.configure(
-            connection=connection,
+            connection=connectable,
             target_metadata=target_metadata,
         )
 

heat/tests/db/test_migrations.py

@@ -62,7 +62,7 @@ class TestBannedDBSchemaOperations(testtools.TestCase):
             self.assertRaises(DBNotAllowed, column.alter)
 
     def test_table(self):
-        table = sqlalchemy.Table()
+        table = sqlalchemy.Table("foo", sqlalchemy.MetaData())
         with BannedDBSchemaOperations(['Table']):
             self.assertRaises(DBNotAllowed, table.drop)
             self.assertRaises(DBNotAllowed, table.alter)