# Path: oslo_db/exception.py
# class DBError(CausedByException):
# class DBDuplicateEntry(DBError):
# class DBConstraintError(DBError):
# class DBReferenceError(DBError):
# class DBNonExistentConstraint(DBError):
# class DBNonExistentTable(DBError):
# class DBNonExistentDatabase(DBError):
# class DBDeadlock(DBError):
# class DBInvalidUnicodeParameter(Exception):
# class DBMigrationError(DBError):
# class DBConnectionError(DBError):
# class DBDataError(DBError):
# class DBNotSupportedError(DBError):
# class InvalidSortKey(Exception):
# class ColumnError(Exception):
# class BackendNotAvailable(Exception):
# class RetryRequest(Exception):
# class NoEngineContextEstablished(AttributeError):
# class ContextNotRequestedError(AttributeError):
# class CantStartEngineError(Exception):
#     def __init__(self, inner_exception=None, cause=None):
#     def __init__(self, columns=None, inner_exception=None, value=None):
#     def __init__(self, table, check_name, inner_exception=None):
#     def __init__(self, table, constraint, key, key_table,
#                  inner_exception=None):
#     def __init__(self, table, constraint, inner_exception=None):
#     def __init__(self, table, inner_exception=None):
#     def __init__(self, database, inner_exception=None):
#     def __init__(self, inner_exception=None):
#     def __init__(self):
#     def __init__(self, message):
#     def __init__(self, key=None):
#     def __init__(self, inner_exc):
# 
# Path: oslo_db/sqlalchemy/migration_cli/ext_alembic.py
# class AlembicExtension(ext_base.MigrationExtensionBase):
#     def enabled(self):
#     def __init__(self, engine, migration_config):
#     def upgrade(self, version):
#     def downgrade(self, version):
#     def version(self):
#     def revision(self, message='', autogenerate=False):
#     def stamp(self, revision):
#     def has_revision(self, rev_id):
# 
# Path: oslo_db/sqlalchemy/migration_cli/ext_migrate.py
# LOG = logging.getLogger(__name__)
# class MigrateExtension(ext_base.MigrationExtensionBase):
#     def __init__(self, engine, migration_config):
#     def enabled(self):
#     def upgrade(self, version):
#     def downgrade(self, version):
#     def version(self):
#     def has_revision(self, rev_id):
# 
# Path: oslo_db/sqlalchemy/migration_cli/manager.py
# MIGRATION_NAMESPACE = 'oslo.db.migration'
# def check_plugin_enabled(ext):
#     def __init__(self, migration_config, engine=None):
#     def _plugins(self):
#     def upgrade(self, revision):
#     def downgrade(self, revision):
#     def version(self):
#     def revision(self, message, autogenerate):
#     def stamp(self, revision):
# class MigrationManager(object):
# 
# Path: oslo_db/tests/base.py
# class BaseTestCase(base.BaseTestCase):
#     def setUp(self):


from unittest import mock
from oslo_db import exception
from oslo_db.sqlalchemy.migration_cli import ext_alembic
from oslo_db.sqlalchemy.migration_cli import ext_migrate
from oslo_db.sqlalchemy.migration_cli import manager
from oslo_db.tests import base as test_base
import alembic
import sqlalchemy
        self.first_ext.obj.has_revision.return_value = False
        self.first_ext.obj.downgrade.side_effect = AssertionError(
            'this method should not have been called')

        results = self.migration_manager.downgrade(100)
        self.assertEqual([100], results)

    def test_upgrade_checks_rev_existence(self):
        self.first_ext.obj.has_revision.return_value = False
        self.second_ext.obj.has_revision.return_value = False

        # upgrade to a specific non-existent revision should fail
        self.assertRaises(exception.DBMigrationError,
                          self.migration_manager.upgrade, 100)

        # upgrade to the "head" should succeed
        self.assertEqual([100, 200], self.migration_manager.upgrade(None))

        # let's assume the second ext has the revision, upgrade should succeed
        self.second_ext.obj.has_revision.return_value = True
        self.assertEqual([100, 200], self.migration_manager.upgrade(200))

        # upgrade to the "head" should still succeed
        self.assertEqual([100, 200], self.migration_manager.upgrade(None))

    def test_downgrade_checks_rev_existence(self):
        self.first_ext.obj.has_revision.return_value = False
        self.second_ext.obj.has_revision.return_value = False

        # upgrade to a specific non-existent revision should fail

# Path: oslo_db/exception.py
# class DBError(CausedByException):
# class DBDuplicateEntry(DBError):
# class DBConstraintError(DBError):
# class DBReferenceError(DBError):
# class DBNonExistentConstraint(DBError):
# class DBNonExistentTable(DBError):
# class DBNonExistentDatabase(DBError):
# class DBDeadlock(DBError):
# class DBInvalidUnicodeParameter(Exception):
# class DBMigrationError(DBError):
# class DBConnectionError(DBError):
# class DBDataError(DBError):
# class DBNotSupportedError(DBError):
# class InvalidSortKey(Exception):
# class ColumnError(Exception):
# class BackendNotAvailable(Exception):
# class RetryRequest(Exception):
# class NoEngineContextEstablished(AttributeError):
# class ContextNotRequestedError(AttributeError):
# class CantStartEngineError(Exception):
#     def __init__(self, inner_exception=None, cause=None):
#     def __init__(self, columns=None, inner_exception=None, value=None):
#     def __init__(self, table, check_name, inner_exception=None):
#     def __init__(self, table, constraint, key, key_table,
#                  inner_exception=None):
#     def __init__(self, table, constraint, inner_exception=None):
#     def __init__(self, table, inner_exception=None):
#     def __init__(self, database, inner_exception=None):
#     def __init__(self, inner_exception=None):
#     def __init__(self):
#     def __init__(self, message):
#     def __init__(self, key=None):
#     def __init__(self, inner_exc):
# 
# Path: oslo_db/sqlalchemy/migration_cli/ext_alembic.py
# class AlembicExtension(ext_base.MigrationExtensionBase):
#     def enabled(self):
#     def __init__(self, engine, migration_config):
#     def upgrade(self, version):
#     def downgrade(self, version):
#     def version(self):
#     def revision(self, message='', autogenerate=False):
#     def stamp(self, revision):
#     def has_revision(self, rev_id):
# 
# Path: oslo_db/sqlalchemy/migration_cli/ext_migrate.py
# LOG = logging.getLogger(__name__)
# class MigrateExtension(ext_base.MigrationExtensionBase):
#     def __init__(self, engine, migration_config):
#     def enabled(self):
#     def upgrade(self, version):
#     def downgrade(self, version):
#     def version(self):
#     def has_revision(self, rev_id):
# 
# Path: oslo_db/sqlalchemy/migration_cli/manager.py
# MIGRATION_NAMESPACE = 'oslo.db.migration'
# def check_plugin_enabled(ext):
#     def __init__(self, migration_config, engine=None):
#     def _plugins(self):
#     def upgrade(self, revision):
#     def downgrade(self, revision):
#     def version(self):
#     def revision(self, message, autogenerate):
#     def stamp(self, revision):
# class MigrationManager(object):
# 
# Path: oslo_db/tests/base.py
# class BaseTestCase(base.BaseTestCase):
#     def setUp(self):


# Path: oslo_db/tests/sqlalchemy/test_migrate_cli.py
from unittest import mock
from oslo_db import exception
from oslo_db.sqlalchemy.migration_cli import ext_alembic
from oslo_db.sqlalchemy.migration_cli import ext_migrate
from oslo_db.sqlalchemy.migration_cli import manager
from oslo_db.tests import base as test_base
import alembic
import sqlalchemy

        self.first_ext.obj.has_revision.return_value = False
        self.first_ext.obj.downgrade.side_effect = AssertionError(
            'this method should not have been called')

        results = self.migration_manager.downgrade(100)
        self.assertEqual([100], results)

    def test_upgrade_checks_rev_existence(self):
        self.first_ext.obj.has_revision.return_value = False
        self.second_ext.obj.has_revision.return_value = False

        # upgrade to a specific non-existent revision should fail
        self.assertRaises(exception.DBMigrationError,
                          self.migration_manager.upgrade, 100)

        # upgrade to the "head" should succeed
        self.assertEqual([100, 200], self.migration_manager.upgrade(None))

        # let's assume the second ext has the revision, upgrade should succeed
        self.second_ext.obj.has_revision.return_value = True
        self.assertEqual([100, 200], self.migration_manager.upgrade(200))

        # upgrade to the "head" should still succeed
        self.assertEqual([100, 200], self.migration_manager.upgrade(None))

    def test_downgrade_checks_rev_existence(self):
        self.first_ext.obj.has_revision.return_value = False
        self.second_ext.obj.has_revision.return_value = False

        # upgrade to a specific non-existent revision should fail

        self.assertRaises(exception.DBMigrationError,