Add test to check that correct functions is used in expand/contract

This test will check that expand branch does not contain drop SQLAlchemy
operations and contract branch does not contain create/add SQLAlchemy
operations.

Partially-Implements: blueprint online-schema-migrations

Change-Id: Ifda31c0599651931c1a98f673f3b10e64538f18b
Related-bug: #1490767

neutron/db/migration/__init__.py

@@ -15,12 +15,24 @@
 import contextlib
 import functools
 
+import alembic
 from alembic import context
 from alembic import op
 import sqlalchemy as sa
 from sqlalchemy.engine import reflection
 
 
+CREATION_OPERATIONS = (sa.sql.ddl.AddConstraint,
+                       sa.sql.ddl.CreateIndex,
+                       sa.sql.ddl.CreateTable,
+                       sa.sql.ddl.CreateColumn,
+                       )
+DROP_OPERATIONS = (sa.sql.ddl.DropConstraint,
+                   sa.sql.ddl.DropIndex,
+                   sa.sql.ddl.DropTable,
+                   alembic.ddl.base.DropColumn)
+
+
 def skip_if_offline(func):
     """Decorator for skipping migrations in offline mode."""
     @functools.wraps(func)

neutron/tests/functional/db/test_migrations.py

@@ -29,6 +29,7 @@ from oslo_db.sqlalchemy import test_migrations
 import sqlalchemy
 from sqlalchemy import event
 
+import neutron.db.migration as migration_help
 from neutron.db.migration.alembic_migrations import external
 from neutron.db.migration import cli as migration
 from neutron.db.migration.models import head as head_models
@@ -256,6 +257,48 @@ class TestModelsMigrationsMysql(_TestModelsMigrations,
                 migration.do_alembic_command(self.alembic_config, 'upgrade',
                                              'heads')
 
+    def test_branches(self):
+
+        def check_expand_branch(conn, clauseelement, multiparams, params):
+            if isinstance(clauseelement, migration_help.DROP_OPERATIONS):
+                self.fail("Migration from expand branch contains drop command")
+
+        def check_contract_branch(conn, clauseelement, multiparams, params):
+            if isinstance(clauseelement, migration_help.CREATION_OPERATIONS):
+                # Skip tables that were created by mistake in contract branch
+                if hasattr(clauseelement, 'element'):
+                    element = clauseelement.element
+                    if any([
+                        isinstance(element, sqlalchemy.Table) and
+                        element.name in ['ml2_geneve_allocations',
+                                         'ml2_geneve_endpoints'],
+                        isinstance(element, sqlalchemy.ForeignKeyConstraint)
+                        and
+                        element.table.name == 'flavorserviceprofilebindings',
+                        isinstance(element, sqlalchemy.Index) and
+                        element.table.name == 'ml2_geneve_allocations'
+                    ]):
+                        return
+                self.fail("Migration from contract branch contains create "
+                          "command")
+
+        engine = self.get_engine()
+        cfg.CONF.set_override('connection', engine.url, group='database')
+        with engine.begin() as connection:
+            self.alembic_config.attributes['connection'] = connection
+            migration.do_alembic_command(self.alembic_config, 'upgrade',
+                                         'kilo')
+
+            with self._listener(engine, check_expand_branch):
+                migration.do_alembic_command(
+                    self.alembic_config, 'upgrade',
+                    '%s@head' % migration.EXPAND_BRANCH)
+
+            with self._listener(engine, check_contract_branch):
+                migration.do_alembic_command(
+                    self.alembic_config, 'upgrade',
+                    '%s@head' % migration.CONTRACT_BRANCH)
+
 
 class TestModelsMigrationsPsql(_TestModelsMigrations,
                                base.PostgreSQLTestCase):