Add test that checks external tables are not changed

Alembic migrations should not change tables which models were moved
out of Neutron. This change add check for this.

Also this change remove clear_override from db_sync in
_TestModelsMigrations which is not needed.

Closes-bug: #1466704

Change-Id: I587cd6fb2baa82fd7e452bb8597136efa5b8084e

diff --git a/neutron/tests/functional/db/test_migrations.py b/neutron/tests/functional/db/test_migrations.py
index 200b601ac49d5d1cc5fa23498c568184e5753d0d..e0a1be07848055b9f77f0af33a3ae7860c86ae25 100644 (file)
--- a/neutron/tests/functional/db/test_migrations.py
+++ b/neutron/tests/functional/db/test_migrations.py
@@ -15,6 +15,7 @@
 import functools
 import logging
 import pprint
+import six
 
 import alembic
 import alembic.autogenerate
@@ -26,6 +27,7 @@ from oslo_config import fixture as config_fixture
 from oslo_db.sqlalchemy import test_base
 from oslo_db.sqlalchemy import test_migrations
 import sqlalchemy
+from sqlalchemy import event
 
 from neutron.db.migration.alembic_migrations import external
 from neutron.db.migration import cli as migration
@@ -122,7 +124,6 @@ class _TestModelsMigrations(test_migrations.ModelsMigrationsSync):
     def db_sync(self, engine):
         cfg.CONF.set_override('connection', engine.url, group='database')
         migration.do_alembic_command(self.alembic_config, 'upgrade', 'heads')
-        cfg.CONF.clear_override('connection', group='database')
 
     def get_engine(self):
         return self.engine
@@ -210,7 +211,35 @@ class _TestModelsMigrations(test_migrations.ModelsMigrationsSync):
 
 class TestModelsMigrationsMysql(_TestModelsMigrations,
                                 base.MySQLTestCase):
-    pass
+
+    # 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'):
+                if (clauseelement.element.name in external.TABLES or
+                        (hasattr(clauseelement, 'table') and
+                         clauseelement.element.table.name in external.TABLES)):
+                    self.fail("External table referenced by neutron core "
+                              "migration.")
+
+        engine = self.get_engine()
+        cfg.CONF.set_override('connection', engine.url, group='database')
+        migration.do_alembic_command(self.alembic_config, 'upgrade', 'kilo')
+
+        event.listen(engine, 'before_execute', block_external_tables)
+        migration.do_alembic_command(self.alembic_config, 'upgrade', 'heads')
+
+        event.remove(engine, 'before_execute', block_external_tables)
 
 
 class TestModelsMigrationsPsql(_TestModelsMigrations,