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

How to use _migrate_server_to method in tempest

Best Python code snippet using tempest_python

test_live_block_migration.py

Source:test_live_block_migration.py

...41        _resp, body = self.admin_servers_client.get_server(server_id)42        return body43    def _get_host_for_server(self, server_id):44        return self._get_server_details(server_id)[self._host_key]45    def _migrate_server_to(self, server_id, dest_host):46        _resp, body = self.admin_servers_client.live_migrate_server(47            server_id, dest_host,48            self.config.compute_feature_enabled.49            block_migration_for_live_migration)50        return body51    def _get_host_other_than(self, host):52        for target_host in self._get_compute_hostnames():53            if host != target_host:54                return target_host55    def _get_non_existing_host_name(self):56        random_name = ''.join(57            random.choice(string.ascii_uppercase) for x in range(20))58        self.assertNotIn(random_name, self._get_compute_hostnames())59        return random_name60    def _get_server_status(self, server_id):61        return self._get_server_details(server_id)['status']62    def _get_an_active_server(self):63        for server_id in self.created_server_ids:64            if 'ACTIVE' == self._get_server_status(server_id):65                return server_id66        else:67            _, server = self.create_test_server(wait_until="ACTIVE")68            server_id = server['id']69            self.password = server['adminPass']70            self.password = 'password'71            self.created_server_ids.append(server_id)72            return server_id73    def _volume_clean_up(self, server_id, volume_id):74        resp, body = self.volumes_client.get_volume(volume_id)75        if body['status'] == 'in-use':76            self.servers_client.detach_volume(server_id, volume_id)77            self.volumes_client.wait_for_volume_status(volume_id, 'available')78        self.volumes_client.delete_volume(volume_id)79    @testtools.skipIf(not CONF.compute_feature_enabled.live_migration,80                      'Live migration not available')81    @attr(type='gate')82    def test_live_block_migration(self):83        # Live block migrate an instance to another host84        if len(self._get_compute_hostnames()) < 2:85            raise self.skipTest(86                "Less than 2 compute nodes, skipping migration test.")87        server_id = self._get_an_active_server()88        actual_host = self._get_host_for_server(server_id)89        target_host = self._get_host_other_than(actual_host)90        self._migrate_server_to(server_id, target_host)91        self.servers_client.wait_for_server_status(server_id, 'ACTIVE')92        self.assertEqual(target_host, self._get_host_for_server(server_id))93    @testtools.skipIf(not CONF.compute_feature_enabled.live_migration,94                      'Live migration not available')95    @attr(type='gate')96    def test_invalid_host_for_migration(self):97        # Migrating to an invalid host should not change the status98        server_id = self._get_an_active_server()99        target_host = self._get_non_existing_host_name()100        self.assertRaises(exceptions.BadRequest, self._migrate_server_to,101                          server_id, target_host)102        self.assertEqual('ACTIVE', self._get_server_status(server_id))103    @testtools.skipIf(not CONF.compute_feature_enabled.live_migration or not104                      CONF.compute_feature_enabled.105                      block_migration_for_live_migration,106                      'Block Live migration not available')107    @testtools.skipIf(not CONF.compute_feature_enabled.108                      block_migrate_cinder_iscsi,109                      'Block Live migration not configured for iSCSI')110    @attr(type='gate')111    def test_iscsi_volume(self):112        # Live block migrate an instance to another host113        if len(self._get_compute_hostnames()) < 2:114            raise self.skipTest(115                "Less than 2 compute nodes, skipping migration test.")116        server_id = self._get_an_active_server()117        actual_host = self._get_host_for_server(server_id)118        target_host = self._get_host_other_than(actual_host)119        resp, volume = self.volumes_client.create_volume(1,120                                                         display_name='test')121        self.volumes_client.wait_for_volume_status(volume['id'],122                                                   'available')123        self.addCleanup(self._volume_clean_up, server_id, volume['id'])124        # Attach the volume to the server125        self.servers_client.attach_volume(server_id, volume['id'],126                                          device='/dev/xvdb')127        self.volumes_client.wait_for_volume_status(volume['id'], 'in-use')128        self._migrate_server_to(server_id, target_host)129        self.servers_client.wait_for_server_status(server_id, 'ACTIVE')130        self.assertEqual(target_host, self._get_host_for_server(server_id))131    @classmethod132    def tearDownClass(cls):133        for server_id in cls.created_server_ids:134            cls.servers_client.delete_server(server_id)135        super(LiveBlockMigrationTestJSON, cls).tearDownClass()136class LiveBlockMigrationTestXML(LiveBlockMigrationTestJSON):137    _host_key = (138        '{http://docs.openstack.org/compute/ext/extended_status/api/v1.1}host')...

test_live_migration_negative.py

Source:test_live_migration_negative.py

...24    def skip_checks(cls):25        super(LiveMigrationNegativeTest, cls).skip_checks()26        if not CONF.compute_feature_enabled.live_migration:27            raise cls.skipException("Live migration is not enabled")28    def _migrate_server_to(self, server_id, dest_host):29        bmflm = CONF.compute_feature_enabled.block_migration_for_live_migration30        kwargs = dict(host=dest_host, block_migration=bmflm)31        if self.is_requested_microversion_compatible('2.24'):32            kwargs['disk_over_commit'] = False33        self.admin_servers_client.live_migrate_server(server_id, **kwargs)34    @decorators.attr(type=['negative'])35    @decorators.idempotent_id('7fb7856e-ae92-44c9-861a-af62d7830bcb')36    def test_invalid_host_for_migration(self):37        # Migrating to an invalid host should not change the status38        target_host = data_utils.rand_name('host')39        server = self.create_test_server(wait_until="ACTIVE")40        self.assertRaises(lib_exc.BadRequest, self._migrate_server_to,41                          server['id'], target_host)42        waiters.wait_for_server_status(self.servers_client, server['id'],...

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.