openstack
/
tempest
Code Issues Proposed changes

Add wait_for_interface_detach method 

The method waits for an interface to be detached from a server.
It will be used in the subsequent patch that tests tagged device
attach and detach (nova API 2.49).

Change-Id: If23bd5947dea345b30a77bc83c5b3dbfa5a4267b
This commit is contained in:
Artom Lifshitz 2018-03-22 12:20:54 -04:00 committed by ghanshyam
parent df0d6d7c5d
commit 3306d428d2
2 changed files with 48 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
tempest/common/waiters.py
View File

@ -287,3 +287,24 @@ def wait_for_interface_status(client, server_id, port_id, status):
raise lib_exc.TimeoutException(message)
return body
def wait_for_interface_detach(client, server_id, port_id):
"""Waits for an interface to be detached from a server."""
body = client.list_interfaces(server_id)['interfaceAttachments']
ports = [iface['port_id'] for iface in body]
start = int(time.time())
while port_id in ports:
time.sleep(client.build_interval)
body = client.list_interfaces(server_id)['interfaceAttachments']
ports = [iface['port_id'] for iface in body]
if port_id not in ports:
return body
timed_out = int(time.time()) - start >= client.build_timeout
if timed_out:
message = ('Interface %s failed to detach from server %s within '
'the required time (%s s)' % (port_id, server_id,
client.build_timeout))
raise lib_exc.TimeoutException(message)

27
tempest/tests/common/test_waiters.py
View File

@ -106,3 +106,30 @@ class TestInterfaceWaiters(base.TestCase):
self.assertRaises(lib_exc.TimeoutException,
waiters.wait_for_interface_status,
self.client, 'server_id', 'port_id', 'ACTIVE')
def _one_interface(self):
return {'interfaceAttachments': [{'port_id': 'port_one'}]}
def _two_interfaces(self):
return {'interfaceAttachments': [{'port_id': 'port_one'},
{'port_id': 'port_two'}]}
def test_wait_for_interface_detach(self):
self.client.list_interfaces.side_effect = [self._two_interfaces(),
self._one_interface()]
with mock.patch.object(time, 'sleep') as sleep_mock:
start_time = int(time.time())
waiters.wait_for_interface_detach(self.client, 'server_id',
'port_two')
end_time = int(time.time())
self.assertLess(end_time, (start_time + self.client.build_timeout))
sleep_mock.assert_called_once_with(self.client.build_interval)
def test_wait_for_interface_detach_timeout(self):
time_mock = self.patch('time.time')
time_mock.side_effect = utils.generate_timeout_series(1)
self.client.list_interfaces.return_value = self._one_interface()
self.assertRaises(lib_exc.TimeoutException,
waiters.wait_for_interface_detach,
self.client, 'server_id', 'port_one')