Upgrade and downgrade based on revision existence

Before there was no way to upgrade to a specific revision, because the
revision passed to the manager was passed to all the plugins, some of
which failed to process it.

Add a new method to every plugin `has_revision`, which returns
whether the plugin has the revision. Use it in upgrade and
downgrade.

Change-Id: I89b02d7ad479da6bff3c492c88edfee9c19abc22
Closes-Bug: 1474067

4 files changed, 63 insertions, 4 deletions

diff --git a/oslo_db/sqlalchemy/migration_cli/ext_alembic.py b/oslo_db/sqlalchemy/migration_cli/ext_alembic.py
index 1dbf88f..825c638 100644
--- a/oslo_db/sqlalchemy/migration_cli/ext_alembic.py
+++ b/oslo_db/sqlalchemy/migration_cli/ext_alembic.py
@@ -15,6 +15,7 @@ import os
 import alembic
 from alembic import config as alembic_config
 import alembic.migration as alembic_migration
+from alembic import script as alembic_script
 
 from oslo_db.sqlalchemy.migration_cli import ext_base
 
@@ -89,3 +90,14 @@ class AlembicExtension(ext_base.MigrationExtensionBase):
         with self.engine.begin() as connection:
             self.config.attributes['connection'] = connection
             return alembic.command.stamp(self.config, revision=revision)
+
+    def has_revision(self, rev_id):
+        if rev_id in ['base', 'head']:
+            return True
+        script = alembic_script.ScriptDirectory(
+            self.config.get_main_option('alembic_repo_path'))
+        try:
+            script.get_revision(rev_id)
+            return True
+        except alembic.util.CommandError:
+            return False
diff --git a/oslo_db/sqlalchemy/migration_cli/ext_base.py b/oslo_db/sqlalchemy/migration_cli/ext_base.py
index 205b7b2..184003d 100644
--- a/oslo_db/sqlalchemy/migration_cli/ext_base.py
+++ b/oslo_db/sqlalchemy/migration_cli/ext_base.py
@@ -70,6 +70,15 @@ class MigrationExtensionBase(object):
         """
         raise NotImplementedError()
 
+    def has_revision(self, rev_id):
+        """Checks whether the repo contains a revision
+
+        :param rev_id: Revision to check
+        :returns: Whether the revision is in the repo
+        :rtype: bool
+        """
+        raise NotImplementedError()
+
     def __cmp__(self, other):
         """Used for definition of plugin order.
 
diff --git a/oslo_db/sqlalchemy/migration_cli/ext_migrate.py b/oslo_db/sqlalchemy/migration_cli/ext_migrate.py
index eb72818..4002972 100644
--- a/oslo_db/sqlalchemy/migration_cli/ext_migrate.py
+++ b/oslo_db/sqlalchemy/migration_cli/ext_migrate.py
@@ -13,6 +13,8 @@
 import logging
 import os
 
+from migrate.versioning import version as migrate_version
+
 from oslo_db._i18n import _LE
 from oslo_db.sqlalchemy import migration
 from oslo_db.sqlalchemy.migration_cli import ext_base
@@ -65,3 +67,14 @@ class MigrateExtension(ext_base.MigrationExtensionBase):
     def version(self):
         return migration.db_version(
             self.engine, self.repository, init_version=self.init_version)
+
+    def has_revision(self, rev_id):
+        collection = migrate_version.Collection(self.repository)
+        try:
+            collection.version(rev_id)
+            return True
+        except (KeyError, ValueError):
+            # NOTE(breton): migrate raises KeyError if an int is passed but not
+            # found in the list of revisions and ValueError if non-int is
+            # passed. Both mean there is no requested revision.
+            return False
diff --git a/oslo_db/sqlalchemy/migration_cli/manager.py b/oslo_db/sqlalchemy/migration_cli/manager.py
index 9b55887..d335fa5 100644
--- a/oslo_db/sqlalchemy/migration_cli/manager.py
+++ b/oslo_db/sqlalchemy/migration_cli/manager.py
@@ -13,6 +13,8 @@
 import sqlalchemy
 from stevedore import enabled
 
+from oslo_db import exception
+
 
 MIGRATION_NAMESPACE = 'oslo.db.migration'
 
@@ -50,17 +52,40 @@ class MigrationManager(object):
 
     def upgrade(self, revision):
         """Upgrade database with all available backends."""
+        # a revision exists only in a single plugin. Until we reached it, we
+        # should upgrade to the plugins' heads.
+        # revision=None is a special case meaning latest revision.
+        rev_in_plugins = [p.has_revision(revision) for p in self._plugins]
+        if not any(rev_in_plugins) and revision is not None:
+            raise exception.DbMigrationError('Revision does not exist')
+
         results = []
-        for plugin in self._plugins:
-            results.append(plugin.upgrade(revision))
+        for plugin, has_revision in zip(self._plugins, rev_in_plugins):
+            if not has_revision or revision is None:
+                results.append(plugin.upgrade(None))
+            else:
+                results.append(plugin.upgrade(revision))
+                break
         return results
 
     def downgrade(self, revision):
         """Downgrade database with available backends."""
+        # a revision exists only in a single plugin. Until we reached it, we
+        # should upgrade to the plugins' first revision.
+        # revision=None is a special case meaning initial revision.
+        rev_in_plugins = [p.has_revision(revision) for p in self._plugins]
+        if not any(rev_in_plugins) and revision is not None:
+            raise exception.DbMigrationError('Revision does not exist')
+
         # downgrading should be performed in reversed order
         results = []
-        for plugin in reversed(self._plugins):
-            results.append(plugin.downgrade(revision))
+        for plugin, has_revision in zip(reversed(self._plugins),
+                                        reversed(rev_in_plugins)):
+            if not has_revision or revision is None:
+                results.append(plugin.downgrade(None))
+            else:
+                results.append(plugin.downgrade(revision))
+                break
         return results
 
     def version(self):