KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use wait_for_volume_attachment_remove_from_server method in tempest

Best Python code snippet using tempest_python

waiters.py

Source:waiters.py

...301            raise lib_exc.TimeoutException(message)302        attachments = client.show_volume(volume_id)['volume']['attachments']303    LOG.info('Attachment %s removed from volume %s after waiting for %f '304             'seconds', attachment_id, volume_id, time.time() - start)305def wait_for_volume_attachment_remove_from_server(306        client, server_id, volume_id):307    """Waits for a volume to be removed from a given server.308    This waiter checks the compute API if the volume attachment is removed.309    """310    start = int(time.time())311    try:312        volumes = client.list_volume_attachments(313            server_id)['volumeAttachments']314    except lib_exc.NotFound:315        # Ignore 404s on detach in case the server is deleted or the volume316        # is already detached.317        return318    while any(volume for volume in volumes if volume['volumeId'] == volume_id):319        time.sleep(client.build_interval)...

test_waiters.py

Source:test_waiters.py

...396        waiters.wait_for_volume_attachment_remove(client, uuids.volume_id,397                                                  uuids.attachment_id)398        # Assert that show volume is only called once before we return399        show_volume.assert_called_once_with(uuids.volume_id)400    def test_wait_for_volume_attachment_remove_from_server(self):401        volume_attached = {402            "volumeAttachments": [{"volumeId": uuids.volume_id}]}403        volume_not_attached = {"volumeAttachments": []}404        mock_list_volume_attachments = mock.Mock(405            side_effect=[volume_attached, volume_not_attached])406        mock_client = mock.Mock(407            spec=servers_client.ServersClient,408            build_interval=1,409            build_timeout=1,410            list_volume_attachments=mock_list_volume_attachments)411        self.patch(412            'time.time',413            side_effect=[0., 0.5, mock_client.build_timeout + 1.])414        self.patch('time.sleep')415        waiters.wait_for_volume_attachment_remove_from_server(416            mock_client, uuids.server_id, uuids.volume_id)417        # Assert that list_volume_attachments is called until the attachment is418        # removed.419        mock_list_volume_attachments.assert_has_calls([420            mock.call(uuids.server_id),421            mock.call(uuids.server_id)])422    def test_wait_for_volume_attachment_remove_from_server_timeout(self):423        volume_attached = {424            "volumeAttachments": [{"volumeId": uuids.volume_id}]}425        mock_list_volume_attachments = mock.Mock(426            side_effect=[volume_attached, volume_attached])427        mock_get_console_output = mock.Mock(428            return_value={'output': 'output'})429        mock_client = mock.Mock(430            spec=servers_client.ServersClient,431            build_interval=1,432            build_timeout=1,433            list_volume_attachments=mock_list_volume_attachments,434            get_console_output=mock_get_console_output)435        self.patch(436            'time.time',437            side_effect=[0., 0.5, mock_client.build_timeout + 1.])438        self.patch('time.sleep')439        self.assertRaises(440            lib_exc.TimeoutException,441            waiters.wait_for_volume_attachment_remove_from_server,442            mock_client, uuids.server_id, uuids.volume_id)443        # Assert that list_volume_attachments is called until the attachment is444        # removed.445        mock_list_volume_attachments.assert_has_calls([446            mock.call(uuids.server_id),447            mock.call(uuids.server_id)])448        # Assert that we fetch console output449        mock_get_console_output.assert_called_once_with(uuids.server_id)450    def test_wait_for_volume_attachment_remove_from_server_not_found(self):451        mock_list_volume_attachments = mock.Mock(452            side_effect=lib_exc.NotFound)453        mock_client = mock.Mock(454            spec=servers_client.ServersClient,455            list_volume_attachments=mock_list_volume_attachments)456        # Assert that nothing is raised when lib_exc_NotFound is raised457        # by the client call to list_volume_attachments458        waiters.wait_for_volume_attachment_remove_from_server(459            mock_client, mock.sentinel.server_id, mock.sentinel.volume_id)460        # Assert that list_volume_attachments was actually called461        mock_list_volume_attachments.assert_called_once_with(462            mock.sentinel.server_id)463    @mock.patch('os.system')464    def test_wait_for_ping_host_alive(self, mock_ping):465        mock_ping.return_value = 0466        # Assert that nothing is raised as the host is alive467        waiters.wait_for_ping('127.0.0.1', 10, 1)468    @mock.patch('os.system')469    def test_wait_for_ping_host_eventually_alive(self, mock_ping):470        mock_ping.side_effect = [1, 1, 0]471        # Assert that nothing is raised when the host is eventually alive472        waiters.wait_for_ping('127.0.0.1', 10, 1)...

test_volume.py

Source:test_volume.py

...99        self.servers_client.detach_volume(100            server['id'], attachment['volumeId'])101        waiters.wait_for_volume_resource_status(102            self.volumes_client, attachment['volumeId'], 'available')103        waiters.wait_for_volume_attachment_remove_from_server(...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.