Port _is_db_connection_error check to exception filters

The last part where we are handling DB exceptions is the
_is_db_connection_error() function, which is called within a retry
loop inside of session.create_engine() to perform an initial
connection test.  SQLAlchemy currently does not run "connect"
errors through the exception handling system.  In order to have
these exceptions participate in the filtering system, add
a new function handle_connect_error(engine) which runs
engine.connect() and feeds the exception into the
handler(context) system directly, producing a compatible context
that the filters can use.   Refactor the looping mechanism within
session.create_engine() into a separate function _test_connection()
so that the logic is encapsulated and can be tested.

partially implement bp: use-events-for-error-wrapping

Change-Id: Iad1bac9d0f9202b21e4c9c170aa84494b770728d

1 files changed, 88 insertions, 0 deletions

diff --git a/tests/sqlalchemy/test_exc_filters.py b/tests/sqlalchemy/test_exc_filters.py
index b4f25b4..8fa4486 100644
--- a/tests/sqlalchemy/test_exc_filters.py
+++ b/tests/sqlalchemy/test_exc_filters.py
@@ -21,6 +21,7 @@ import sqlalchemy as sqla
 from sqlalchemy.orm import mapper
 
 from oslo.db import exception
+from oslo.db.sqlalchemy import session
 from oslo.db.sqlalchemy import test_base
 
 _TABLE_NAME = '__tmp__test__tmp__'
@@ -509,3 +510,90 @@ class TestDBDisconnected(TestsExceptionFilter):
             self.OperationalError(
                 'SQL30081N: DB2 Server connection is no longer active')
         )
+
+
+class TestDBConnectRetry(TestsExceptionFilter):
+
+    def _run_test(self, dialect_name, exception, count, retries):
+        counter = itertools.count()
+
+        engine = self.engine
+
+        # empty out the connection pool
+        engine.dispose()
+
+        connect_fn = engine.dialect.connect
+
+        def cant_connect(*arg, **kw):
+            if next(counter) < count:
+                raise exception
+            else:
+                return connect_fn(*arg, **kw)
+
+        with self._dbapi_fixture(dialect_name):
+            with mock.patch.object(engine.dialect, "connect", cant_connect):
+                return session._test_connection(engine, retries, .01)
+
+    def test_connect_no_retries(self):
+        conn = self._run_test(
+            "mysql",
+            self.OperationalError("Error: (2003) something wrong"),
+            2, 0
+        )
+        # didnt connect because nothing was tried
+        self.assertIsNone(conn)
+
+    def test_connect_inifinite_retries(self):
+        conn = self._run_test(
+            "mysql",
+            self.OperationalError("Error: (2003) something wrong"),
+            2, -1
+        )
+        # conn is good
+        self.assertEqual(conn.scalar(sqla.select([1])), 1)
+
+    def test_connect_retry_past_failure(self):
+        conn = self._run_test(
+            "mysql",
+            self.OperationalError("Error: (2003) something wrong"),
+            2, 3
+        )
+        # conn is good
+        self.assertEqual(conn.scalar(sqla.select([1])), 1)
+
+    def test_connect_retry_not_candidate_exception(self):
+        self.assertRaises(
+            sqla.exc.OperationalError,  # remember, we pass OperationalErrors
+                                        # through at the moment :)
+            self._run_test,
+            "mysql",
+            self.OperationalError("Error: (2015) I can't connect period"),
+            2, 3
+        )
+
+    def test_connect_retry_stops_infailure(self):
+        self.assertRaises(
+            exception.DBConnectionError,
+            self._run_test,
+            "mysql",
+            self.OperationalError("Error: (2003) something wrong"),
+            3, 2
+        )
+
+    def test_db2_error_positive(self):
+        conn = self._run_test(
+            "ibm_db_sa",
+            self.OperationalError("blah blah -30081 blah blah"),
+            2, -1
+        )
+        # conn is good
+        self.assertEqual(conn.scalar(sqla.select([1])), 1)
+
+    def test_db2_error_negative(self):
+        self.assertRaises(
+            sqla.exc.OperationalError,
+            self._run_test,
+            "ibm_db_sa",
+            self.OperationalError("blah blah -39981 blah blah"),
+            2, 3
+        )