openstack
/
oslo.db
Code Issues Proposed changes

Move helper DB functions to db.sqlalchemy.utils 

Database utilities are contained in
openstack/common/db/sqlalchemy/test_migrations.py which is not
appropriate to the module appointment. All the utility functions moved
to openstack/common/db/sqlalchemy/utils.py.

Change-Id: Ia5e1efd3f8e2e1f9553ca24ca7bb19bea911cdbc
This commit is contained in:
Ilya Pekelny 2013-12-12 17:32:39 +02:00
parent 985ee188c9
commit 977eb52cca
4 changed files with 102 additions and 58 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

59
openstack/common/db/sqlalchemy/test_migrations.py
View File

@ -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 "

47
openstack/common/db/sqlalchemy/utils.py
View File

@ -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)

12
tests/unit/db/sqlalchemy/base.py
View File

@ -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)

42
tests/unit/db/sqlalchemy/test_utils.py
View File

@ -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'))