From: Ann Kamyshnikova <akamyshnikova@mirantis.com>
Date: Fri, 7 Aug 2015 13:53:53 +0000 (+0300)
Subject: Add test that checks external tables are not changed
X-Git-Url: https://review.fuel-infra.org/gitweb?a=commitdiff_plain;h=43f3d7506121084e29d5084ea3da7cbd2e7e7902;p=openstack-build%2Fneutron-build.git

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 200b601ac..e0a1be078 100644
--- 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,