db: Add replacement test for walking migrations

Previously, this was provided by oslo.db. However, the oslo.db variant
only works with sqlalchemy-migrate. We have to provide our own one here.
Fortunately, this has already been done in Cinder, Nova etc. and we can
duplicate the effort from there.

We also add some sanity check logic, while we're here.

Signed-off-by: Stephen Finucane <stephenfin@redhat.com>
Change-Id: I1ab3fd46d6564b71f4c9c81940943d11e944eb68

1 files changed, 103 insertions, 0 deletions

diff --git a/heat/tests/db/test_migrations.py b/heat/tests/db/test_migrations.py
index 31485c1b0..b627e963e 100644
--- a/heat/tests/db/test_migrations.py
+++ b/heat/tests/db/test_migrations.py
@@ -12,6 +12,8 @@
 
 """Tests for database migrations."""
 
+from alembic import command as alembic_api
+from alembic import script as alembic_script
 import fixtures
 from oslo_db.sqlalchemy import enginefacade
 from oslo_db.sqlalchemy import test_fixtures
@@ -109,3 +111,104 @@ class ModelsMigrationsSyncSQLite(
     test_base.BaseTestCase,
 ):
     pass
+
+
+class DatabaseSanityChecks(
+    test_fixtures.OpportunisticDBTestMixin,
+    test_base.BaseTestCase,
+):
+    def setUp(self):
+        super().setUp()
+        self.engine = enginefacade.writer.get_engine()
+        # self.patch(api, 'get_engine', lambda: self.engine)
+        self.config = migration._find_alembic_conf()
+        self.init_version = migration.ALEMBIC_INIT_VERSION
+
+    def test_single_base_revision(self):
+        """Ensure we only have a single base revision.
+
+        There's no good reason for us to have diverging history, so validate
+        that only one base revision exists. This will prevent simple errors
+        where people forget to specify the base revision. If this fail for your
+        change, look for migrations that do not have a 'revises' line in them.
+        """
+        script = alembic_script.ScriptDirectory.from_config(self.config)
+        self.assertEqual(1, len(script.get_bases()))
+
+    def test_single_head_revision(self):
+        """Ensure we only have a single head revision.
+
+        There's no good reason for us to have diverging history, so validate
+        that only one head revision exists. This will prevent merge conflicts
+        adding additional head revision points. If this fail for your change,
+        look for migrations with the same 'revises' line in them.
+        """
+        script = alembic_script.ScriptDirectory.from_config(self.config)
+        self.assertEqual(1, len(script.get_heads()))
+
+
+class MigrationsWalk(
+    test_fixtures.OpportunisticDBTestMixin,
+    test_base.BaseTestCase,
+):
+    # Migrations can take a long time, particularly on underpowered CI nodes.
+    # Give them some breathing room.
+    TIMEOUT_SCALING_FACTOR = 4
+
+    def setUp(self):
+        super().setUp()
+        self.engine = enginefacade.writer.get_engine()
+        # self.patch(api, 'get_engine', lambda: self.engine)
+        self.config = migration._find_alembic_conf()
+        self.init_version = migration.ALEMBIC_INIT_VERSION
+
+    def _migrate_up(self, revision, connection):
+        check_method = getattr(self, f'_check_{revision}', None)
+        if revision != self.init_version:  # no tests for the initial revision
+            self.assertIsNotNone(
+                check_method,
+                f"DB Migration {revision} doesn't have a test; add one"
+            )
+
+        pre_upgrade = getattr(self, f'_pre_upgrade_{revision}', None)
+        if pre_upgrade:
+            pre_upgrade(connection)
+
+        alembic_api.upgrade(self.config, revision)
+
+        if check_method:
+            check_method(connection)
+
+    def test_walk_versions(self):
+        with self.engine.begin() as connection:
+            self.config.attributes['connection'] = connection
+            script = alembic_script.ScriptDirectory.from_config(self.config)
+            revisions = list(script.walk_revisions())
+            # Need revisions from older to newer so the walk works as intended
+            revisions.reverse()
+            for revision_script in revisions:
+                self._migrate_up(revision_script.revision, connection)
+
+
+class TestMigrationsWalkSQLite(
+    MigrationsWalk,
+    test_fixtures.OpportunisticDBTestMixin,
+    test_base.BaseTestCase,
+):
+    pass
+
+
+class TestMigrationsWalkMySQL(
+    MigrationsWalk,
+    test_fixtures.OpportunisticDBTestMixin,
+    test_base.BaseTestCase,
+):
+    FIXTURE = test_fixtures.MySQLOpportunisticFixture
+
+
+class TestMigrationsWalkPostgreSQL(
+    MigrationsWalk,
+    test_fixtures.OpportunisticDBTestMixin,
+    test_base.BaseTestCase,
+):
+    FIXTURE = test_fixtures.PostgresqlOpportunisticFixture