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
diff --git a/tempest/common/waiters.py b/tempest/common/waiters.py
index 08e2a12..0e86f05 100644
--- a/tempest/common/waiters.py
+++ b/tempest/common/waiters.py
@@ -287,3 +287,24 @@
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)
diff --git a/tempest/tests/common/test_waiters.py b/tempest/tests/common/test_waiters.py
index 8c62ca0..eb1e2b6 100644
--- a/tempest/tests/common/test_waiters.py
+++ b/tempest/tests/common/test_waiters.py
@@ -106,3 +106,30 @@
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')