# 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.py
# def db_sync(engine, abs_path, version=None, init_version=0, sanity_check=True):
# def _db_schema_sanity_check(engine):
# def db_version(engine, abs_path, init_version):
# def db_version_control(engine, abs_path, version=None):
# def _find_migrate_repo(abs_path):
#         EXCLUDED_TABLES = ['migrate_version', 'alembic_version']
# 
# Path: oslo_db/tests/sqlalchemy/base.py
# class Context(object):
# class DbTestCase(db_fixtures.OpportunisticDBTestMixin, test_base.BaseTestCase):
# class MySQLOpportunisticTestCase(DbTestCase):
# class PostgreSQLOpportunisticTestCase(DbTestCase):
# class _DbTestCase(
#     db_fixtures.OpportunisticDBTestMixin, test_base.BaseTestCase,
# ):
# class _MySQLOpportunisticTestCase(_DbTestCase):
# class _PostgreSQLOpportunisticTestCase(_DbTestCase):
#     def setUp(self):
#     def setUp(self):
#     FIXTURE = db_fixtures.MySQLOpportunisticFixture
#     FIXTURE = db_fixtures.PostgresqlOpportunisticFixture
#     FIXTURE = db_fixtures.MySQLOpportunisticFixture
#     FIXTURE = db_fixtures.PostgresqlOpportunisticFixture
# 
# Path: oslo_db/tests/utils.py
# def nested(*contexts):


import os
import tempfile
import sqlalchemy
from unittest import mock
from migrate import exceptions as migrate_exception
from migrate.versioning import api as versioning_api
from oslo_db import exception as db_exception
from oslo_db.sqlalchemy import migration
from oslo_db.tests.sqlalchemy import base as test_base
from oslo_db.tests import utils as test_utils
        # VersionNotFoundError which will be wrapped in DBMigrationError.
        upgrade.side_effect = migrate_exception.VersionNotFoundError
        self.assertRaises(db_exception.DBMigrationError,
                          migration.db_sync, self.engine, self.path,
                          self.test_version + 1)

    @mock.patch.object(versioning_api, 'upgrade')
    def test_db_sync_known_error_raised(self, upgrade):
        upgrade.side_effect = migrate_exception.KnownError
        self.assertRaises(db_exception.DBMigrationError,
                          migration.db_sync, self.engine, self.path,
                          self.test_version + 1)

    def test_db_sync_upgrade(self):
        init_ver = 55
        with test_utils.nested(
            mock.patch.object(migration, '_find_migrate_repo'),
            mock.patch.object(versioning_api, 'upgrade')
        ) as (mock_find_repo, mock_upgrade):

            mock_find_repo.return_value = self.return_value
            self.mock_api_db_version.return_value = self.test_version - 1

            migration.db_sync(self.engine, self.path, self.test_version,
                              init_ver)

            mock_upgrade.assert_called_once_with(
                self.engine, self.return_value, self.test_version)

    def test_db_sync_downgrade(self):

# 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.py
# def db_sync(engine, abs_path, version=None, init_version=0, sanity_check=True):
# def _db_schema_sanity_check(engine):
# def db_version(engine, abs_path, init_version):
# def db_version_control(engine, abs_path, version=None):
# def _find_migrate_repo(abs_path):
#         EXCLUDED_TABLES = ['migrate_version', 'alembic_version']
# 
# Path: oslo_db/tests/sqlalchemy/base.py
# class Context(object):
# class DbTestCase(db_fixtures.OpportunisticDBTestMixin, test_base.BaseTestCase):
# class MySQLOpportunisticTestCase(DbTestCase):
# class PostgreSQLOpportunisticTestCase(DbTestCase):
# class _DbTestCase(
#     db_fixtures.OpportunisticDBTestMixin, test_base.BaseTestCase,
# ):
# class _MySQLOpportunisticTestCase(_DbTestCase):
# class _PostgreSQLOpportunisticTestCase(_DbTestCase):
#     def setUp(self):
#     def setUp(self):
#     FIXTURE = db_fixtures.MySQLOpportunisticFixture
#     FIXTURE = db_fixtures.PostgresqlOpportunisticFixture
#     FIXTURE = db_fixtures.MySQLOpportunisticFixture
#     FIXTURE = db_fixtures.PostgresqlOpportunisticFixture
# 
# Path: oslo_db/tests/utils.py
# def nested(*contexts):


# Path: oslo_db/tests/sqlalchemy/test_migration_common.py
import os
import tempfile
import sqlalchemy
from unittest import mock
from migrate import exceptions as migrate_exception
from migrate.versioning import api as versioning_api
from oslo_db import exception as db_exception
from oslo_db.sqlalchemy import migration
from oslo_db.tests.sqlalchemy import base as test_base
from oslo_db.tests import utils as test_utils

        # VersionNotFoundError which will be wrapped in DBMigrationError.
        upgrade.side_effect = migrate_exception.VersionNotFoundError
        self.assertRaises(db_exception.DBMigrationError,
                          migration.db_sync, self.engine, self.path,
                          self.test_version + 1)

    @mock.patch.object(versioning_api, 'upgrade')
    def test_db_sync_known_error_raised(self, upgrade):
        upgrade.side_effect = migrate_exception.KnownError
        self.assertRaises(db_exception.DBMigrationError,
                          migration.db_sync, self.engine, self.path,
                          self.test_version + 1)

    def test_db_sync_upgrade(self):
        init_ver = 55
        with test_utils.nested(
            mock.patch.object(migration, '_find_migrate_repo'),
            mock.patch.object(versioning_api, 'upgrade')
        ) as (mock_find_repo, mock_upgrade):

            mock_find_repo.return_value = self.return_value
            self.mock_api_db_version.return_value = self.test_version - 1

            migration.db_sync(self.engine, self.path, self.test_version,
                              init_ver)

            mock_upgrade.assert_called_once_with(
                self.engine, self.return_value, self.test_version)

    def test_db_sync_downgrade(self):

        with test_utils.nested(