diff --git a/openstack/common/db/sqlalchemy/test_migrations.py b/openstack/common/db/sqlalchemy/test_migrations.py index 7ca36aa7..f8e7504a 100644 --- a/openstack/common/db/sqlalchemy/test_migrations.py +++ b/openstack/common/db/sqlalchemy/test_migrations.py @@ -23,6 +23,7 @@ from six import moves import sqlalchemy import sqlalchemy.exc +from openstack.common.db.sqlalchemy import utils from openstack.common.gettextutils import _ from openstack.common import log as logging from openstack.common.py3kcompat import urlutils @@ -31,67 +32,20 @@ from openstack.common import test LOG = logging.getLogger(__name__) -def _get_connect_string(backend, user, passwd, database): - """Get database connection - - Try to get a connection with a very specific set of values, if we get - these then we'll run the tests, otherwise they are skipped - """ - if backend == "postgres": - backend = "postgresql+psycopg2" - elif backend == "mysql": - backend = "mysql+mysqldb" - else: - raise Exception("Unrecognized backend: '%s'" % backend) - - return ("%(backend)s://%(user)s:%(passwd)s@localhost/%(database)s" - % {'backend': backend, 'user': user, 'passwd': passwd, - '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 - - def _have_mysql(user, passwd, database): present = os.environ.get('TEST_MYSQL_PRESENT') if present is None: - return _is_backend_avail('mysql', user, passwd, database) + return utils.is_backend_avail('mysql', user, passwd, database) return present.lower() in ('', 'true') def _have_postgresql(user, passwd, database): present = os.environ.get('TEST_POSTGRESQL_PRESENT') if present is None: - return _is_backend_avail('postgres', user, passwd, database) + return utils.is_backend_avail('postgres', user, passwd, database) return present.lower() in ('', 'true') -def get_db_connection_info(conn_pieces): - database = conn_pieces.path.strip('/') - loc_pieces = conn_pieces.netloc.split('@') - host = loc_pieces[1] - - auth_pieces = loc_pieces[0].split(':') - user = auth_pieces[0] - password = "" - if len(auth_pieces) > 1: - password = auth_pieces[1].strip() - - return (user, password, database, host) - - def _set_db_lock(lock_path=None, lock_prefix=None): def decorator(f): @functools.wraps(f) @@ -166,7 +120,10 @@ class BaseMigrationTestCase(test.BaseTestCase): "Failed to run: %s\n%s" % (cmd, output)) def _reset_pg(self, conn_pieces): - (user, password, database, host) = get_db_connection_info(conn_pieces) + (user, + password, + database, + host) = utils.get_db_connection_info(conn_pieces) os.environ['PGPASSWORD'] = password os.environ['PGUSER'] = user # note(boris-42): We must create and drop database, we can't @@ -205,7 +162,7 @@ class BaseMigrationTestCase(test.BaseTestCase): # the MYSQL database, which is easier and less error-prone # than using SQLAlchemy to do this via MetaData...trust me. (user, password, database, host) = \ - get_db_connection_info(conn_pieces) + utils.get_db_connection_info(conn_pieces) sql = ("drop database if exists %(db)s; " "create database %(db)s;") % {'db': database} cmd = ("mysql -u \"%(user)s\" -p\"%(password)s\" -h %(host)s " diff --git a/openstack/common/db/sqlalchemy/utils.py b/openstack/common/db/sqlalchemy/utils.py index 2bf87eb9..30a2d984 100644 --- a/openstack/common/db/sqlalchemy/utils.py +++ b/openstack/common/db/sqlalchemy/utils.py @@ -497,3 +497,50 @@ def _change_deleted_column_type_to_id_type_sqlite(migrate_engine, table_name, where(new_table.c.deleted == deleted).\ values(deleted=default_deleted_value).\ execute() + + +def get_connect_string(backend, user, passwd, database): + """Get database connection + + Try to get a connection with a very specific set of values, if we get + these then we'll run the tests, otherwise they are skipped + """ + if backend == "postgres": + backend = "postgresql+psycopg2" + elif backend == "mysql": + backend = "mysql+mysqldb" + else: + raise Exception("Unrecognized backend: '%s'" % backend) + + return ("%(backend)s://%(user)s:%(passwd)s@localhost/%(database)s" + % {'backend': backend, 'user': user, 'passwd': passwd, + '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 + + +def get_db_connection_info(conn_pieces): + database = conn_pieces.path.strip('/') + loc_pieces = conn_pieces.netloc.split('@') + host = loc_pieces[1] + + auth_pieces = loc_pieces[0].split(':') + user = auth_pieces[0] + password = "" + if len(auth_pieces) > 1: + password = auth_pieces[1].strip() + + return (user, password, database, host) diff --git a/tests/unit/db/sqlalchemy/base.py b/tests/unit/db/sqlalchemy/base.py index 2ec1d901..5602078b 100644 --- a/tests/unit/db/sqlalchemy/base.py +++ b/tests/unit/db/sqlalchemy/base.py @@ -22,7 +22,7 @@ from oslo.config import cfg import six from openstack.common.db.sqlalchemy import session -from openstack.common.db.sqlalchemy import test_migrations as tm +from openstack.common.db.sqlalchemy import utils from tests import utils as test_utils @@ -108,10 +108,10 @@ class OpportunisticFixture(DbFixture): DBNAME = PASSWORD = USERNAME = 'openstack_citest' def _get_uri(self): - return tm._get_connect_string(backend=self.DRIVER, - user=self.USERNAME, - passwd=self.PASSWORD, - database=self.DBNAME) + return utils.get_connect_string(backend=self.DRIVER, + user=self.USERNAME, + passwd=self.PASSWORD, + database=self.DBNAME) @six.add_metaclass(abc.ABCMeta) @@ -131,7 +131,7 @@ class OpportunisticTestCase(DbTestCase): self.FIXTURE.PASSWORD, self.FIXTURE.DBNAME) - if self.FIXTURE.DRIVER and not tm._is_backend_avail(*credentials): + if self.FIXTURE.DRIVER and not utils.is_backend_avail(*credentials): msg = '%s backend is not available.' % self.FIXTURE.DRIVER return self.skip(msg) diff --git a/tests/unit/db/sqlalchemy/test_utils.py b/tests/unit/db/sqlalchemy/test_utils.py index ec57f842..fc677af7 100644 --- a/tests/unit/db/sqlalchemy/test_utils.py +++ b/tests/unit/db/sqlalchemy/test_utils.py @@ -23,15 +23,17 @@ from sqlalchemy.dialects import mysql from sqlalchemy import Boolean, Index, Integer, DateTime, String from sqlalchemy import MetaData, Table, Column, ForeignKey from sqlalchemy.engine import reflection -from sqlalchemy.exc import SAWarning +from sqlalchemy.exc import SAWarning, OperationalError from sqlalchemy.sql import select from sqlalchemy.types import UserDefinedType, NullType +from openstack.common.py3kcompat import urlutils from openstack.common.db.sqlalchemy import migration from openstack.common.db.sqlalchemy import test_migrations from openstack.common.db.sqlalchemy import utils from openstack.common.fixture import moxstubout from openstack.common import test +from tests import utils as test_utils class TestSanitizeDbUrl(test.BaseTestCase): @@ -548,3 +550,41 @@ class TestMigrationUtils(test_migrations.BaseMigrationTestCase): self.assertEqual(f_key['referred_table'], 'table0') self.assertEqual(f_key['referred_columns'], ['id']) self.assertEqual(f_key['constrained_columns'], ['bar']) + + +class TestConnectionUtils(test_utils.BaseTestCase): + + def setUp(self): + super(TestConnectionUtils, self).setUp() + + self.full_credentials = {'backend': 'mysql', + 'database': 'test', + 'user': 'dude', + 'passwd': 'pass'} + + self.connect_string = 'mysql+mysqldb://dude:pass@localhost/test' + + def test_connect_string(self): + connect_string = utils.get_connect_string(**self.full_credentials) + self.assertEqual(connect_string, self.connect_string) + + def test_is_backend_avail(self): + self.mox.StubOutWithMock(sqlalchemy.engine.base.Engine, 'connect') + fake_connection = self.mox.CreateMockAnything() + fake_connection.close() + sqlalchemy.engine.base.Engine.connect().AndReturn(fake_connection) + self.mox.ReplayAll() + + self.assertTrue(utils.is_backend_avail(**self.full_credentials)) + + def test_is_backend_unavail(self): + self.mox.StubOutWithMock(sqlalchemy.engine.base.Engine, 'connect') + sqlalchemy.engine.base.Engine.connect().AndRaise(OperationalError) + self.mox.ReplayAll() + + self.assertFalse(utils.is_backend_avail(**self.full_credentials)) + + def test_get_db_connection_info(self): + conn_pieces = urlutils.urlparse(self.connect_string) + self.assertEqual(utils.get_db_connection_info(conn_pieces), + ('dude', 'pass', 'test', 'localhost'))