openstack/ironic
Code Issues Proposed changes

Merge "Switch to new location for oslo.db test cases"

Browse Source
This commit is contained in:
Jenkins 2017-02-22 18:02:58 +00:00 committed by Gerrit Code Review
commit 42c138eb6a
1 changed files with 1 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
ironic/tests/unit/db/sqlalchemy/test_migrations.py
View File

@ -42,9 +42,9 @@ from alembic import script
import mock import mock
from oslo_db import exception as db_exc from oslo_db import exception as db_exc
from oslo_db.sqlalchemy import enginefacade from oslo_db.sqlalchemy import enginefacade
from oslo_db.sqlalchemy import test_base
from oslo_db.sqlalchemy import test_migrations from oslo_db.sqlalchemy import test_migrations
from oslo_db.sqlalchemy import utils as db_utils from oslo_db.sqlalchemy import utils as db_utils
from oslo_db.tests.sqlalchemy import base as test_base
from oslo_log import log as logging from oslo_log import log as logging
from oslo_utils import uuidutils from oslo_utils import uuidutils
import sqlalchemy import sqlalchemy
@ -81,21 +81,6 @@ def _get_connect_string(backend, user, passwd, database):
'database': database}) 'database': database})
def _is_backend_avail(backend, user, passwd, database):
try:
connect_uri = _get_connect_string(backend, user, passwd, database)
engine = sqlalchemy.create_engine(connect_uri)
connection = engine.connect()
except Exception:
# intentionally catch all to handle exceptions even if we don't
# have any backend code loaded.
return False
else:
connection.close()
engine.dispose()
return True
@contextlib.contextmanager @contextlib.contextmanager
def patch_with_engine(engine): def patch_with_engine(engine):
with mock.patch.object(enginefacade.get_legacy_facade(), with mock.patch.object(enginefacade.get_legacy_facade(),
@ -224,16 +209,6 @@ class MigrationCheckersMixin(object):
def test_walk_versions(self): def test_walk_versions(self):
self._walk_versions(self.engine, self.config) self._walk_versions(self.engine, self.config)
def test_connect_fail(self):
"""Test that we can trigger a database connection failure
Test that we can fail gracefully to ensure we don't break people
without specific database backend
"""
if _is_backend_avail(self.FIXTURE.DRIVER, "openstack_cifail",
self.FIXTURE.USERNAME, self.FIXTURE.DBNAME):
self.fail("Shouldn't have connected")
def _check_21b331f883ef(self, engine, data): def _check_21b331f883ef(self, engine, data):
nodes = db_utils.get_table(engine, 'nodes') nodes = db_utils.get_table(engine, 'nodes')
col_names = [column.name for column in nodes.c] col_names = [column.name for column in nodes.c]