Merge "Remove method get_connect_string and is_backend_avail"4.32.0

2 files changed, 0 insertions, 81 deletions

diff --git a/oslo_db/sqlalchemy/utils.py b/oslo_db/sqlalchemy/utils.py
index 1a647b4..7c5f22c 100644
--- a/oslo_db/sqlalchemy/utils.py
+++ b/oslo_db/sqlalchemy/utils.py
@@ -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('@')
diff --git a/oslo_db/tests/sqlalchemy/test_utils.py b/oslo_db/tests/sqlalchemy/test_utils.py
index b7f38ed..d6fb3a3 100644
--- a/oslo_db/tests/sqlalchemy/test_utils.py
+++ b/oslo_db/tests/sqlalchemy/test_utils.py
@@ -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):