openstack
/
neutron
Code Issues Proposed changes

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
This commit is contained in:
Ann Kamyshnikova 2015-09-09 14:32:36 +03:00
parent 4d831a462e
commit bbaa4abdd5
2 changed files with 55 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
neutron/db/migration/__init__.py
View File

@ -15,12 +15,24 @@
import contextlib import contextlib
import functools import functools
import alembic
from alembic import context from alembic import context
from alembic import op from alembic import op
import sqlalchemy as sa import sqlalchemy as sa
from sqlalchemy.engine import reflection 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): def skip_if_offline(func):
"""Decorator for skipping migrations in offline mode.""" """Decorator for skipping migrations in offline mode."""
@functools.wraps(func) @functools.wraps(func)

43
neutron/tests/functional/db/test_migrations.py
View File

@ -29,6 +29,7 @@ from oslo_db.sqlalchemy import test_migrations
import sqlalchemy import sqlalchemy
from sqlalchemy import event from sqlalchemy import event
import neutron.db.migration as migration_help
from neutron.db.migration.alembic_migrations import external from neutron.db.migration.alembic_migrations import external
from neutron.db.migration import cli as migration from neutron.db.migration import cli as migration
from neutron.db.migration.models import head as head_models 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', migration.do_alembic_command(self.alembic_config, 'upgrade',
'heads') '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, class TestModelsMigrationsPsql(_TestModelsMigrations,
base.PostgreSQLTestCase): base.PostgreSQLTestCase):