openstack
/
oslo.db
Code Issues Proposed changes

Merge "Remove method get_connect_string and is_backend_avail"

This commit is contained in:
Zuul 2017-12-09 18:14:54 +00:00 committed by Gerrit Code Review
parent 15152f8e63 73e72f6cc2
commit 85cf42e841
2 changed files with 0 additions and 81 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

48
oslo_db/sqlalchemy/utils.py
View File

@ -712,54 +712,6 @@ def _change_deleted_column_type_to_id_type_sqlite(engine, table_name,
execute()
def get_connect_string(backend, database, user=None, passwd=None,
host='localhost'):
"""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
DEPRECATED: this function is deprecated and will be removed from oslo_db
in a few releases. Please use the provisioning system for dealing
with URLs and database provisioning.
"""
args = {'backend': backend,
'user': user,
'passwd': passwd,
'host': host,
'database': database}
if backend == 'sqlite':
template = '%(backend)s:///%(database)s'
else:
template = "%(backend)s://%(user)s:%(passwd)s@%(host)s/%(database)s"
return template % args
def is_backend_avail(backend, database, user=None, passwd=None):
"""Return True if the given backend is available.
DEPRECATED: this function is deprecated and will be removed from oslo_db
in a few releases. Please use the provisioning system to access
databases based on backend availability.
"""
from oslo_db.sqlalchemy import provision
connect_uri = get_connect_string(backend=backend,
database=database,
user=user,
passwd=passwd)
try:
eng = provision.Backend._ensure_backend_available(connect_uri)
eng.dispose()
except exception.BackendNotAvailable:
return False
else:
return True
def get_db_connection_info(conn_pieces):
database = conn_pieces.path.strip('/')
loc_pieces = conn_pieces.netloc.split('@')

33
oslo_db/tests/sqlalchemy/test_utils.py
View File

@ -998,34 +998,6 @@ class TestConnectionUtils(test_utils.BaseTestCase):
patch_onconnect.start()
self.addCleanup(patch_onconnect.stop)
def test_connect_string(self):
connect_string = utils.get_connect_string(**self.full_credentials)
self.assertEqual(self.connect_string, connect_string)
def test_connect_string_sqlite(self):
sqlite_credentials = {'backend': 'sqlite', 'database': 'test.db'}
connect_string = utils.get_connect_string(**sqlite_credentials)
self.assertEqual('sqlite:///test.db', 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):
log = self.useFixture(fixtures.FakeLogger())
err = OperationalError("Can't connect to database", None, None)
error_msg = "The postgresql backend is unavailable: %s\n" % err
self.mox.StubOutWithMock(sqlalchemy.engine.base.Engine, 'connect')
sqlalchemy.engine.base.Engine.connect().AndRaise(err)
self.mox.ReplayAll()
self.assertFalse(utils.is_backend_avail(**self.full_credentials))
self.assertEqual(error_msg, log.output)
def test_ensure_backend_available(self):
self.mox.StubOutWithMock(sqlalchemy.engine.base.Engine, 'connect')
fake_connection = self.mox.CreateMockAnything()
@ -1079,11 +1051,6 @@ class TestConnectionUtils(test_utils.BaseTestCase):
self.assertEqual(('dude', 'pass', 'test', 'localhost'),
utils.get_db_connection_info(conn_pieces))
def test_connect_string_host(self):
self.full_credentials['host'] = 'myhost'
connect_string = utils.get_connect_string(**self.full_credentials)
self.assertEqual('postgresql://dude:pass@myhost/test', connect_string)
class MyModelSoftDeletedProjectId(declarative_base(), models.ModelBase,
models.SoftDeleteMixin):