openstack/python-openstackclient
Code Issues Proposed changes

Merge "Allow wait_for_delete to work for all clients"

Browse Source
This commit is contained in:
Jenkins 2016-02-04 00:39:47 +00:00 committed by Gerrit Code Review
commit e4c0d8583d
2 changed files with 28 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
openstackclient/common/utils.py
View File

@ -314,6 +314,8 @@ def wait_for_status(status_f,
def wait_for_delete(manager,
res_id,
status_field='status',
error_status=['error'],
exception_name=['NotFound'],
sleep_time=5,
timeout=300,
callback=None):
@ -324,6 +326,8 @@ def wait_for_delete(manager,
:param status_field: the status attribute in the returned resource object,
this is used to check for error states while the resource is being
deleted
:param error_status: a list of status strings for error
:param exception_name: a list of exception strings for deleted case
:param sleep_time: wait this long between checks (seconds)
:param timeout: check until this long (seconds)
:param callback: called per sleep cycle, useful to display progress; this
@ -340,12 +344,12 @@ def wait_for_delete(manager,
# handle a NotFound exception here without parsing the message
res = manager.get(res_id)
except Exception as ex:
if type(ex).__name__ == 'NotFound':
if type(ex).__name__ in exception_name:
return True
raise
status = getattr(res, status_field, '').lower()
if status == 'error':
if status in error_status:
return False
if callback:

22
openstackclient/tests/common/test_utils.py
View File

@ -212,6 +212,28 @@ class TestUtils(test_utils.TestCase):
self.assertFalse(utils.wait_for_delete(manager, res_id))
self.assertFalse(mock_sleep.called)
@mock.patch.object(time, 'sleep')
def test_wait_for_delete_error_with_overrides(self, mock_sleep):
# Tests that we fail if the resource is my_status=failed
resource = mock.MagicMock(my_status='FAILED')
mock_get = mock.Mock(return_value=resource)
manager = mock.MagicMock(get=mock_get)
res_id = str(uuid.uuid4())
self.assertFalse(utils.wait_for_delete(manager, res_id,
status_field='my_status',
error_status=['failed']))
self.assertFalse(mock_sleep.called)
@mock.patch.object(time, 'sleep')
def test_wait_for_delete_error_with_overrides_exception(self, mock_sleep):
# Tests that we succeed if the resource is specific exception
mock_get = mock.Mock(side_effect=Exception)
manager = mock.MagicMock(get=mock_get)
res_id = str(uuid.uuid4())
self.assertTrue(utils.wait_for_delete(manager, res_id,
exception_name=['Exception']))
self.assertFalse(mock_sleep.called)
def test_build_kwargs_dict_value_set(self):
self.assertEqual({'arg_bla': 'bla'},
utils.build_kwargs_dict('arg_bla', 'bla'))