x/deb-oslo.db
Code Issues Proposed changes

Add tests for float interval values in wrap_db_retry()

Browse Source 
Added unit test for wrap_db_retry that used a float for retry_interval
Added mocking of time.sleep() and checking the calls on retry_interval
related tests

Closes Bug: #1549241

Change-Id: Ic7eeb5654aca263aee6e84ab591e061e2ccc6c86
This commit is contained in:
Julian Sy
2016-02-23 02:39:48 -05:00
parent fdb5ad8c2d
commit e9cbc6e958
1 changed files with 27 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
oslo_db/tests/test_api.py
View File

@@ -123,7 +123,8 @@ class DBReconnectTestCase(DBAPITestCase):
self.dbapi.api_raise_default)
self.assertEqual(4, self.test_db_api.error_counter, 'Unexpected retry')
def test_retry_one(self):
@mock.patch('oslo_db.api.time.sleep', return_value=None)
def test_retry_one(self, p_time_sleep):
self.dbapi = api.DBAPI('sqlalchemy',
{'sqlalchemy': __name__},
use_db_reconnect=True,
@@ -135,12 +136,13 @@ class DBReconnectTestCase(DBAPITestCase):
self.assertTrue(func(), 'Single retry did not succeed.')
except Exception:
self.fail('Single retry raised an un-wrapped error.')
p_time_sleep.assert_called_with(1)
self.assertEqual(
0, self.test_db_api.error_counter,
'Counter not decremented, retry logic probably failed.')
def test_retry_two(self):
@mock.patch('oslo_db.api.time.sleep', return_value=None)
def test_retry_two(self, p_time_sleep):
self.dbapi = api.DBAPI('sqlalchemy',
{'sqlalchemy': __name__},
use_db_reconnect=True,
@@ -153,12 +155,31 @@ class DBReconnectTestCase(DBAPITestCase):
self.assertTrue(func(), 'Multiple retry did not succeed.')
except Exception:
self.fail('Multiple retry raised an un-wrapped error.')
p_time_sleep.assert_called_with(1)
self.assertEqual(
0, self.test_db_api.error_counter,
'Counter not decremented, retry logic probably failed.')
def test_retry_until_failure(self):
@mock.patch('oslo_db.api.time.sleep', return_value=None)
def test_retry_float_interval(self, p_time_sleep):
self.dbapi = api.DBAPI('sqlalchemy',
{'sqlalchemy': __name__},
use_db_reconnect=True,
retry_interval=0.5)
try:
func = self.dbapi.api_raise_enable_retry
self.test_db_api.error_counter = 1
self.assertTrue(func(), 'Single retry did not succeed.')
except Exception:
self.fail('Single retry raised an un-wrapped error.')
p_time_sleep.assert_called_with(0.5)
self.assertEqual(
0, self.test_db_api.error_counter,
'Counter not decremented, retry logic probably failed.')
@mock.patch('oslo_db.api.time.sleep', return_value=None)
def test_retry_until_failure(self, p_time_sleep):
self.dbapi = api.DBAPI('sqlalchemy',
{'sqlalchemy': __name__},
use_db_reconnect=True,
@@ -171,7 +192,7 @@ class DBReconnectTestCase(DBAPITestCase):
self.assertRaises(
exception.DBError, func,
'Retry of permanent failure did not throw DBError exception.')
p_time_sleep.assert_called_with(1)
self.assertNotEqual(
0, self.test_db_api.error_counter,
'Retry did not stop after sql_max_retries iterations.')