openstack
/
neutron
Code Issues Proposed changes

Remove functional migration script about external tables 

This test has run out of its useful purpose. It was meant to protect
accidental schema changes involving external tables. After a few cycles
now, it is very unlikely that human error will not be spotted during
code review, where external tables are referenced/modified during Neutron
core schema migrations.

Putting effort into fixing the BGP removal corner case is not worth it,
and it is probably best to get rid of the test entirely.

Change-Id: I71da8ea5aa5513056a0dc62b6ac9d8c0394583dd
Related-bug: 1466704
This commit is contained in:
Armando Migliaccio 2016-04-22 12:42:05 -07:00
parent ae5bad49cc
commit 24e04ddaa2
1 changed files with 0 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -171,45 +171,6 @@ class TestModelsMigrationsMysql(_TestModelsMigrations,
event.remove(engine, 'before_execute',
listener_func)
# There is no use to run this against both dialects, so add this test just
# for MySQL tests
def test_external_tables_not_changed(self):
def block_external_tables(conn, clauseelement, multiparams, params):
if isinstance(clauseelement, sqlalchemy.sql.selectable.Select):
return
if (isinstance(clauseelement, six.string_types) and
any(name in clauseelement for name in external.TABLES)):
self.fail("External table referenced by neutron core "
"migration.")
if hasattr(clauseelement, 'element'):
element = clauseelement.element
if (element.name in external.TABLES or
(hasattr(clauseelement, 'table') and
element.table.name in external.TABLES)):
# Table 'nsxv_vdr_dhcp_bindings' was created in liberty,
# before NSXV has moved to separate repo.
if ((isinstance(clauseelement,
sqlalchemy.sql.ddl.CreateTable) and
element.name == 'nsxv_vdr_dhcp_bindings')):
return
self.fail("External table referenced by neutron core "
"migration.")
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,
block_external_tables):
migration.do_alembic_command(self.alembic_config, 'upgrade',
'heads')
def test_branches(self):
def check_expand_branch(conn, clauseelement, multiparams, params):