# 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
    def test_manager_update_revision_none(self):
        self.migration_manager.upgrade(None)
        self.ext.obj.upgrade.assert_called_once_with(None)

    def test_downgrade_normal_revision(self):
        self.migration_manager.downgrade('111abcd')
        self.ext.obj.downgrade.assert_called_once_with('111abcd')

    def test_version(self):
        self.migration_manager.version()
        self.ext.obj.version.assert_called_once_with()

    def test_version_return_value(self):
        version = self.migration_manager.version()
        self.assertEqual(0, version)

    def test_revision_message_autogenerate(self):
        self.migration_manager.revision('test', True)
        self.ext.obj.revision.assert_called_once_with('test', True)

    def test_revision_only_message(self):
        self.migration_manager.revision('test', False)
        self.ext.obj.revision.assert_called_once_with('test', False)

    def test_stamp(self):
        self.migration_manager.stamp('stamp')
        self.ext.obj.stamp.assert_called_once_with('stamp')

    def test_wrong_config(self):
        err = self.assertRaises(ValueError,

# 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

    def test_manager_update_revision_none(self):
        self.migration_manager.upgrade(None)
        self.ext.obj.upgrade.assert_called_once_with(None)

    def test_downgrade_normal_revision(self):
        self.migration_manager.downgrade('111abcd')
        self.ext.obj.downgrade.assert_called_once_with('111abcd')

    def test_version(self):
        self.migration_manager.version()
        self.ext.obj.version.assert_called_once_with()

    def test_version_return_value(self):
        version = self.migration_manager.version()
        self.assertEqual(0, version)

    def test_revision_message_autogenerate(self):
        self.migration_manager.revision('test', True)
        self.ext.obj.revision.assert_called_once_with('test', True)

    def test_revision_only_message(self):
        self.migration_manager.revision('test', False)
        self.ext.obj.revision.assert_called_once_with('test', False)

    def test_stamp(self):
        self.migration_manager.stamp('stamp')
        self.ext.obj.stamp.assert_called_once_with('stamp')

    def test_wrong_config(self):
        err = self.assertRaises(ValueError,

                                manager.MigrationManager,