# Path: oslo_db/api.py
# LOG = logging.getLogger(__name__)
# def safe_for_db_retry(f):
# def retry_on_deadlock(f):
# def retry_on_request(f):
#     def __init__(self, retry_interval=1, max_retries=20,
#                  inc_retry_interval=True,
#                  max_retry_interval=10, retry_on_disconnect=False,
#                  retry_on_deadlock=False,
#                  exception_checker=lambda exc: False,
#                  jitter=False):
#     def __call__(self, f):
#         def wrapper(*args, **kwargs):
#     def _is_exception_expected(self, exc):
#     def _get_inc_interval(self, n, jitter):
#     def __init__(self, backend_name, backend_mapping=None, lazy=False,
#                  **kwargs):
#     def _load_backend(self):
#     def __getattr__(self, key):
#     def from_config(cls, conf, backend_mapping=None, lazy=False):
# class wrap_db_retry(object):
# class DBAPI(object):
# 
# 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/tests/base.py
# class BaseTestCase(base.BaseTestCase):
#     def setUp(self):


from unittest import mock
from oslo_config import cfg
from oslo_utils import importutils
from oslo_db import api
from oslo_db import exception
from oslo_db.tests import base as test_base
    def test_dbapi_lazy_loading(self):
        dbapi = api.DBAPI('oslo_db.tests.test_api', lazy=True)

        self.assertIsNone(dbapi._backend)
        dbapi.api_class_call1(1, 'abc')
        self.assertIsNotNone(dbapi._backend)

    def test_dbapi_from_config(self):
        conf = cfg.ConfigOpts()

        dbapi = api.DBAPI.from_config(conf,
                                      backend_mapping={'sqlalchemy': __name__})
        self.assertIsNotNone(dbapi._backend)


class DBReconnectTestCase(DBAPITestCase):

    def setUp(self):
        super().setUp()

        self.test_db_api = DBAPI()
        patcher = mock.patch(__name__ + '.get_backend',
                             return_value=self.test_db_api)
        patcher.start()
        self.addCleanup(patcher.stop)

    def test_raise_connection_error(self):
        self.dbapi = api.DBAPI('sqlalchemy', {'sqlalchemy': __name__})

        self.test_db_api.error_counter = 5

# Path: oslo_db/api.py
# LOG = logging.getLogger(__name__)
# def safe_for_db_retry(f):
# def retry_on_deadlock(f):
# def retry_on_request(f):
#     def __init__(self, retry_interval=1, max_retries=20,
#                  inc_retry_interval=True,
#                  max_retry_interval=10, retry_on_disconnect=False,
#                  retry_on_deadlock=False,
#                  exception_checker=lambda exc: False,
#                  jitter=False):
#     def __call__(self, f):
#         def wrapper(*args, **kwargs):
#     def _is_exception_expected(self, exc):
#     def _get_inc_interval(self, n, jitter):
#     def __init__(self, backend_name, backend_mapping=None, lazy=False,
#                  **kwargs):
#     def _load_backend(self):
#     def __getattr__(self, key):
#     def from_config(cls, conf, backend_mapping=None, lazy=False):
# class wrap_db_retry(object):
# class DBAPI(object):
# 
# 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/tests/base.py
# class BaseTestCase(base.BaseTestCase):
#     def setUp(self):


# Path: oslo_db/tests/test_api.py
from unittest import mock
from oslo_config import cfg
from oslo_utils import importutils
from oslo_db import api
from oslo_db import exception
from oslo_db.tests import base as test_base

    def test_dbapi_lazy_loading(self):
        dbapi = api.DBAPI('oslo_db.tests.test_api', lazy=True)

        self.assertIsNone(dbapi._backend)
        dbapi.api_class_call1(1, 'abc')
        self.assertIsNotNone(dbapi._backend)

    def test_dbapi_from_config(self):
        conf = cfg.ConfigOpts()

        dbapi = api.DBAPI.from_config(conf,
                                      backend_mapping={'sqlalchemy': __name__})
        self.assertIsNotNone(dbapi._backend)


class DBReconnectTestCase(DBAPITestCase):

    def setUp(self):
        super().setUp()

        self.test_db_api = DBAPI()
        patcher = mock.patch(__name__ + '.get_backend',
                             return_value=self.test_db_api)
        patcher.start()
        self.addCleanup(patcher.stop)

    def test_raise_connection_error(self):
        self.dbapi = api.DBAPI('sqlalchemy', {'sqlalchemy': __name__})

        self.test_db_api.error_counter = 5

        self.assertRaises(exception.DBConnectionError, self.dbapi._api_raise)