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

How to use wait_for_server_volume_swap method in tempest

Best Python code snippet using tempest_python

test_volume_swap.py

Source:test_volume_swap.py

...22    def skip_checks(cls):23        super(TestVolumeSwapBase, cls).skip_checks()24        if not CONF.compute_feature_enabled.swap_volume:25            raise cls.skipException("Swapping volumes is not supported.")26    def wait_for_server_volume_swap(self, server_id, old_volume_id,27                                    new_volume_id):28        """Waits for a server to swap the old volume to a new one."""29        volume_attachments = self.servers_client.list_volume_attachments(30            server_id)['volumeAttachments']31        attached_volume_ids = [attachment['volumeId']32                               for attachment in volume_attachments]33        start = int(time.time())34        while (old_volume_id in attached_volume_ids) \35                or (new_volume_id not in attached_volume_ids):36            time.sleep(self.servers_client.build_interval)37            volume_attachments = self.servers_client.list_volume_attachments(38                server_id)['volumeAttachments']39            attached_volume_ids = [attachment['volumeId']40                                   for attachment in volume_attachments]41            if int(time.time()) - start >= self.servers_client.build_timeout:42                old_vol_bdm_status = 'in BDM' \43                    if old_volume_id in attached_volume_ids else 'not in BDM'44                new_vol_bdm_status = 'in BDM' \45                    if new_volume_id in attached_volume_ids else 'not in BDM'46                message = ('Failed to swap old volume %(old_volume_id)s '47                           '(current %(old_vol_bdm_status)s) to new volume '48                           '%(new_volume_id)s (current %(new_vol_bdm_status)s)'49                           ' on server %(server_id)s within the required time '50                           '(%(timeout)s s)' %51                           {'old_volume_id': old_volume_id,52                            'old_vol_bdm_status': old_vol_bdm_status,53                            'new_volume_id': new_volume_id,54                            'new_vol_bdm_status': new_vol_bdm_status,55                            'server_id': server_id,56                            'timeout': self.servers_client.build_timeout})57                raise lib_exc.TimeoutException(message)58class TestVolumeSwap(TestVolumeSwapBase):59    """The test suite for swapping of volume with admin user.60    The following is the scenario outline:61    1. Create a volume "volume1" with non-admin.62    2. Create a volume "volume2" with non-admin.63    3. Boot an instance "instance1" with non-admin.64    4. Attach "volume1" to "instance1" with non-admin.65    5. Swap volume from "volume1" to "volume2" as admin.66    6. Check the swap volume is successful and "volume2"67       is attached to "instance1" and "volume1" is in available state.68    7. Swap volume from "volume2" to "volume1" as admin.69    8. Check the swap volume is successful and "volume1"70       is attached to "instance1" and "volume2" is in available state.71    """72    @decorators.idempotent_id('1769f00d-a693-4d67-a631-6a3496773813')73    @utils.services('volume')74    def test_volume_swap(self):75        # Create two volumes.76        # NOTE(gmann): Volumes are created before server creation so that77        # volumes cleanup can happen successfully irrespective of which volume78        # is attached to server.79        volume1 = self.create_volume()80        volume2 = self.create_volume()81        # Boot server82        server = self.create_test_server(wait_until='ACTIVE')83        # Attach "volume1" to server84        self.attach_volume(server, volume1)85        # Swap volume from "volume1" to "volume2"86        self.admin_servers_client.update_attached_volume(87            server['id'], volume1['id'], volumeId=volume2['id'])88        waiters.wait_for_volume_resource_status(self.volumes_client,89                                                volume1['id'], 'available')90        waiters.wait_for_volume_resource_status(self.volumes_client,91                                                volume2['id'], 'in-use')92        self.wait_for_server_volume_swap(server['id'], volume1['id'],93                                         volume2['id'])94        # Verify "volume2" is attached to the server95        vol_attachments = self.servers_client.list_volume_attachments(96            server['id'])['volumeAttachments']97        self.assertEqual(1, len(vol_attachments))98        self.assertIn(volume2['id'], vol_attachments[0]['volumeId'])99        # Swap volume from "volume2" to "volume1"100        self.admin_servers_client.update_attached_volume(101            server['id'], volume2['id'], volumeId=volume1['id'])102        waiters.wait_for_volume_resource_status(self.volumes_client,103                                                volume2['id'], 'available')104        waiters.wait_for_volume_resource_status(self.volumes_client,105                                                volume1['id'], 'in-use')106        self.wait_for_server_volume_swap(server['id'], volume2['id'],107                                         volume1['id'])108        # Verify "volume1" is attached to the server109        vol_attachments = self.servers_client.list_volume_attachments(110            server['id'])['volumeAttachments']111        self.assertEqual(1, len(vol_attachments))112        self.assertIn(volume1['id'], vol_attachments[0]['volumeId'])113class TestMultiAttachVolumeSwap(TestVolumeSwapBase):114    min_microversion = '2.60'115    max_microversion = 'latest'116    @classmethod117    def skip_checks(cls):118        super(TestMultiAttachVolumeSwap, cls).skip_checks()119        if not CONF.compute_feature_enabled.volume_multiattach:120            raise cls.skipException('Volume multi-attach is not available.')121    @decorators.idempotent_id('e8f8f9d1-d7b7-4cd2-8213-ab85ef697b6e')122    @utils.services('volume')123    def test_volume_swap_with_multiattach(self):124        # Create two volumes.125        # NOTE(gmann): Volumes are created before server creation so that126        # volumes cleanup can happen successfully irrespective of which volume127        # is attached to server.128        volume1 = self.create_volume(multiattach=True)129        volume2 = self.create_volume(multiattach=True)130        # Boot server1131        server1 = self.create_test_server(wait_until='ACTIVE')132        # Attach volume1 to server1133        self.attach_volume(server1, volume1)134        # Boot server2135        server2 = self.create_test_server(wait_until='ACTIVE')136        # Attach volume1 to server2137        self.attach_volume(server2, volume1)138        # Swap volume1 to volume2 on server1, volume1 should remain attached139        # to server 2140        self.admin_servers_client.update_attached_volume(141            server1['id'], volume1['id'], volumeId=volume2['id'])142        # volume1 will return to in-use after the swap143        waiters.wait_for_volume_resource_status(self.volumes_client,144                                                volume1['id'], 'in-use')145        waiters.wait_for_volume_resource_status(self.volumes_client,146                                                volume2['id'], 'in-use')147        self.wait_for_server_volume_swap(server1['id'], volume1['id'],148                                         volume2['id'])149        # Verify volume2 is attached to server1150        vol_attachments = self.servers_client.list_volume_attachments(151            server1['id'])['volumeAttachments']152        self.assertEqual(1, len(vol_attachments))153        self.assertIn(volume2['id'], vol_attachments[0]['volumeId'])154        # Verify volume1 is still attached to server2155        vol_attachments = self.servers_client.list_volume_attachments(156            server2['id'])['volumeAttachments']157        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.