How to use _test_live_migration method in tempest

Best Python code snippet using tempest_python

test_live_migration.py

Source:test_live_migration.py

...72        if body['status'] == 'in-use':73            self.servers_client.detach_volume(server_id, volume_id)74            self.volumes_client.wait_for_volume_status(volume_id, 'available')75        self.volumes_client.delete_volume(volume_id)76    def _test_live_migration(self, state='ACTIVE', volume_backed=False):77        """Tests live migration between two hosts.78        Requires CONF.compute_feature_enabled.live_migration to be True.79        :param state: The vm_state the migrated server should be in before and80                      after the live migration. Supported values are 'ACTIVE'81                      and 'PAUSED'.82        :param volume_backed: If the instance is volume backed or not. If83                              volume_backed, *block* migration is not used.84        """85        # Live migrate an instance to another host86        server_id = self._create_server(volume_backed=volume_backed)87        actual_host = self._get_host_for_server(server_id)88        target_host = self._get_host_other_than(actual_host)89        if state == 'PAUSED':90            self.admin_servers_client.pause_server(server_id)91            waiters.wait_for_server_status(self.admin_servers_client,92                                           server_id, state)93        self._migrate_server_to(server_id, target_host, volume_backed)94        waiters.wait_for_server_status(self.servers_client, server_id, state)95        migration_list = (self.admin_migration_client.list_migrations()96                          ['migrations'])97        msg = ("Live Migration failed. Migrations list for Instance "98               "%s: [" % server_id)99        for live_migration in migration_list:100            if (live_migration['instance_uuid'] == server_id):101                msg += "\n%s" % live_migration102        msg += "]"103        self.assertEqual(target_host, self._get_host_for_server(server_id),104                         msg)105    @test.idempotent_id('1dce86b8-eb04-4c03-a9d8-9c1dc3ee0c7b')106    def test_live_block_migration(self):107        self._test_live_migration()108    @test.idempotent_id('1e107f21-61b2-4988-8f22-b196e938ab88')109    @testtools.skipUnless(CONF.compute_feature_enabled.pause,110                          'Pause is not available.')111    @testtools.skipUnless(CONF.compute_feature_enabled112                              .live_migrate_paused_instances,113                          'Live migration of paused instances is not '114                          'available.')115    def test_live_block_migration_paused(self):116        self._test_live_migration(state='PAUSED')117    @test.idempotent_id('5071cf17-3004-4257-ae61-73a84e28badd')118    @test.services('volume')119    def test_volume_backed_live_migration(self):120        self._test_live_migration(volume_backed=True)121    @test.idempotent_id('e19c0cc6-6720-4ed8-be83-b6603ed5c812')122    @testtools.skipIf(not CONF.compute_feature_enabled.123                      block_migration_for_live_migration,124                      'Block Live migration not available')125    @testtools.skipIf(not CONF.compute_feature_enabled.126                      block_migrate_cinder_iscsi,127                      'Block Live migration not configured for iSCSI')128    def test_iscsi_volume(self):129        server_id = self._create_server()130        actual_host = self._get_host_for_server(server_id)131        target_host = self._get_host_other_than(actual_host)132        volume = self.volumes_client.create_volume(133            display_name='test')['volume']134        self.volumes_client.wait_for_volume_status(volume['id'],...

test_livemigrationops.py

Source:test_livemigrationops.py

...28        self._livemigrops._livemigrutils = mock.MagicMock()29    @mock.patch('nova.virt.hyperv.vmops.VMOps.copy_vm_console_logs')30    @mock.patch('nova.virt.configdrive.required_by')31    @mock.patch('nova.virt.hyperv.pathutils.PathUtils.copy_configdrive')32    def _test_live_migration(self, mock_copy_configdrive, mock_required_by,33                             mock_copy_logs, side_effect, configdrive=False):34        mock_instance = fake_instance.fake_instance_obj(self.context)35        mock_post = mock.MagicMock()36        mock_recover = mock.MagicMock()37        fake_dest = mock.sentinel.DESTINATION38        self._livemigrops._livemigrutils.live_migrate_vm.side_effect = [39            side_effect]40        if side_effect is vmutils.HyperVException:41            self.assertRaises(vmutils.HyperVException,42                              self._livemigrops.live_migration,43                              self.context, mock_instance, fake_dest,44                              mock_post, mock_recover, False, None)45            mock_recover.assert_called_once_with(self.context, mock_instance,46                                                 fake_dest, False)47        else:48            if configdrive:49                mock_required_by.return_value = True50                self.flags(config_drive_cdrom=True, group='hyperv')51            self._livemigrops.live_migration(context=self.context,52                                             instance_ref=mock_instance,53                                             dest=fake_dest,54                                             post_method=mock_post,55                                             recover_method=mock_recover)56            mock_copy_logs.assert_called_once_with(mock_instance.name,57                                                   fake_dest)58            if configdrive:59                mock_copy_configdrive.assert_called_once_with(60                    mock_instance.name, fake_dest)61            mock_live_migr = self._livemigrops._livemigrutils.live_migrate_vm62            mock_live_migr.assert_called_once_with(mock_instance.name,63                                                   fake_dest)64            mock_post.assert_called_once_with(self.context, mock_instance,65                                              fake_dest, False)66    def test_live_migration(self):67        self._test_live_migration(side_effect=None)68    def test_live_migration_exception(self):69        self._test_live_migration(side_effect=vmutils.HyperVException)70    def test_live_migration_wrong_os_version(self):71        self._livemigrops._livemigrutils = None72        self.assertRaises(NotImplementedError,73                          self._livemigrops.live_migration, self.context,74                          instance_ref=mock.DEFAULT,75                          dest=mock.sentinel.DESTINATION,76                          post_method=mock.DEFAULT,77                          recover_method=mock.DEFAULT)78    def test_live_migration_with_configdrive(self):79        self._test_live_migration(side_effect=None, configdrive=True)80    @mock.patch('nova.virt.hyperv.volumeops.VolumeOps'81                '.ebs_root_in_block_devices')82    @mock.patch('nova.virt.hyperv.imagecache.ImageCache.get_cached_image')83    @mock.patch('nova.virt.hyperv.volumeops.VolumeOps'84                '.initialize_volumes_connection')85    def test_pre_live_migration(self, mock_initialize_connection,86                                mock_get_cached_image,87                                mock_ebs_root_in_block_devices):88        mock_instance = fake_instance.fake_instance_obj(self.context)89        mock_instance.image_ref = "fake_image_ref"90        mock_ebs_root_in_block_devices.return_value = None91        CONF.set_override('use_cow_images', True)92        self._livemigrops.pre_live_migration(93            self.context, mock_instance,...

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.