Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use list_volume_attachments method in tempest

Best Python code snippet using tempest_python

test_volume_swap.py

Source:test_volume_swap.py

...38            raise cls.skipException("Swapping volumes is not supported.")39    def _wait_for_server_volume_swap(self, server_id, old_volume_id,40                                     new_volume_id):41        """Waits for a server to swap the old volume to a new one."""42        volume_attachments = self.servers_client.list_volume_attachments(43            server_id)['volumeAttachments']44        attached_volume_ids = [attachment['volumeId']45                               for attachment in volume_attachments]46        start = int(time.time())47        while (old_volume_id in attached_volume_ids) \48                or (new_volume_id not in attached_volume_ids):49            time.sleep(self.servers_client.build_interval)50            volume_attachments = self.servers_client.list_volume_attachments(51                server_id)['volumeAttachments']52            attached_volume_ids = [attachment['volumeId']53                                   for attachment in volume_attachments]54            if int(time.time()) - start >= self.servers_client.build_timeout:55                old_vol_bdm_status = 'in BDM' \56                    if old_volume_id in attached_volume_ids else 'not in BDM'57                new_vol_bdm_status = 'in BDM' \58                    if new_volume_id in attached_volume_ids else 'not in BDM'59                message = ('Failed to swap old volume %(old_volume_id)s '60                           '(current %(old_vol_bdm_status)s) to new volume '61                           '%(new_volume_id)s (current %(new_vol_bdm_status)s)'62                           ' on server %(server_id)s within the required time '63                           '(%(timeout)s s)' %64                           {'old_volume_id': old_volume_id,65                            'old_vol_bdm_status': old_vol_bdm_status,66                            'new_volume_id': new_volume_id,67                            'new_vol_bdm_status': new_vol_bdm_status,68                            'server_id': server_id,69                            'timeout': self.servers_client.build_timeout})70                raise lib_exc.TimeoutException(message)71    @decorators.idempotent_id('1769f00d-a693-4d67-a631-6a3496773813')72    @test.services('volume')73    def test_volume_swap(self):74        # Create two volumes.75        # NOTE(gmann): Volumes are created before server creation so that76        # volumes cleanup can happen successfully irrespective of which volume77        # is attached to server.78        volume1 = self.create_volume()79        volume2 = self.create_volume()80        # Boot server81        server = self.create_test_server(wait_until='ACTIVE')82        # Attach "volume1" to server83        self.attach_volume(server, volume1)84        # Swap volume from "volume1" to "volume2"85        self.admin_servers_client.update_attached_volume(86            server['id'], volume1['id'], volumeId=volume2['id'])87        waiters.wait_for_volume_resource_status(self.volumes_client,88                                                volume1['id'], 'available')89        waiters.wait_for_volume_resource_status(self.volumes_client,90                                                volume2['id'], 'in-use')91        self._wait_for_server_volume_swap(server['id'], volume1['id'],92                                          volume2['id'])93        # Verify "volume2" is attached to the server94        vol_attachments = self.servers_client.list_volume_attachments(95            server['id'])['volumeAttachments']96        self.assertEqual(1, len(vol_attachments))97        self.assertIn(volume2['id'], vol_attachments[0]['volumeId'])98        # Swap volume from "volume2" to "volume1"99        self.admin_servers_client.update_attached_volume(100            server['id'], volume2['id'], volumeId=volume1['id'])101        waiters.wait_for_volume_resource_status(self.volumes_client,102                                                volume2['id'], 'available')103        waiters.wait_for_volume_resource_status(self.volumes_client,104                                                volume1['id'], 'in-use')105        self._wait_for_server_volume_swap(server['id'], volume2['id'],106                                          volume1['id'])107        # Verify "volume1" is attached to the server108        vol_attachments = self.servers_client.list_volume_attachments(109            server['id'])['volumeAttachments']110        self.assertEqual(1, len(vol_attachments))...

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.