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

How to use get_host_for_server method in tempest

Best Python code snippet using tempest_python

test_live_migration.py

Source:test_live_migration.py

...62        for live_migration in migration_list:63            if (live_migration['instance_uuid'] == server_id):64                msg += "\n%s" % live_migration65        msg += "]"66        self.assertEqual(target_host, self.get_host_for_server(server_id),67                         msg)68    def _test_live_migration(self, state='ACTIVE', volume_backed=False):69        """Tests live migration between two hosts.70        Requires CONF.compute_feature_enabled.live_migration to be True.71        :param state: The vm_state the migrated server should be in before and72                      after the live migration. Supported values are 'ACTIVE'73                      and 'PAUSED'.74        :param volume_backed: If the instance is volume backed or not. If75                              volume_backed, *block* migration is not used.76        """77        # Live migrate an instance to another host78        server_id = self.create_test_server(wait_until="ACTIVE",79                                            volume_backed=volume_backed)['id']80        source_host = self.get_host_for_server(server_id)81        destination_host = self.get_host_other_than(server_id)82        if state == 'PAUSED':83            self.admin_servers_client.pause_server(server_id)84            waiters.wait_for_server_status(self.admin_servers_client,85                                           server_id, state)86        LOG.info("Live migrate from source %s to destination %s",87                 source_host, destination_host)88        self._live_migrate(server_id, destination_host, state, volume_backed)89        if CONF.compute_feature_enabled.live_migrate_back_and_forth:90            # If live_migrate_back_and_forth is enabled it is a grenade job.91            # Therefore test should validate whether LM is compatible in both92            # ways, so live migrate VM back to the source host93            LOG.info("Live migrate back to source %s", source_host)94            self._live_migrate(server_id, source_host, state, volume_backed)95    @decorators.idempotent_id('1dce86b8-eb04-4c03-a9d8-9c1dc3ee0c7b')96    def test_live_block_migration(self):97        self._test_live_migration()98    @decorators.idempotent_id('1e107f21-61b2-4988-8f22-b196e938ab88')99    @testtools.skipUnless(CONF.compute_feature_enabled.pause,100                          'Pause is not available.')101    def test_live_block_migration_paused(self):102        self._test_live_migration(state='PAUSED')103    @decorators.skip_because(bug="1524898")104    @decorators.idempotent_id('5071cf17-3004-4257-ae61-73a84e28badd')105    @utils.services('volume')106    def test_volume_backed_live_migration(self):107        self._test_live_migration(volume_backed=True)108    @decorators.idempotent_id('e19c0cc6-6720-4ed8-be83-b6603ed5c812')109    @testtools.skipIf(not CONF.compute_feature_enabled.110                      block_migration_for_live_migration,111                      'Block Live migration not available')112    @testtools.skipIf(not CONF.compute_feature_enabled.113                      block_migrate_cinder_iscsi,114                      'Block Live migration not configured for iSCSI')115    def test_iscsi_volume(self):116        server = self.create_test_server(wait_until="ACTIVE")117        server_id = server['id']118        target_host = self.get_host_other_than(server_id)119        volume = self.create_volume()120        # Attach the volume to the server121        self.attach_volume(server, volume, device='/dev/xvdb')122        server = self.admin_servers_client.show_server(server_id)['server']123        volume_id1 = server["os-extended-volumes:volumes_attached"][0]["id"]124        self._migrate_server_to(server_id, target_host)125        waiters.wait_for_server_status(self.servers_client,126                                       server_id, 'ACTIVE')127        server = self.admin_servers_client.show_server(server_id)['server']128        volume_id2 = server["os-extended-volumes:volumes_attached"][0]["id"]129        self.assertEqual(target_host, self.get_host_for_server(server_id))130        self.assertEqual(volume_id1, volume_id2)131class LiveMigrationRemoteConsolesV26Test(LiveMigrationTest):132    min_microversion = '2.6'133    max_microversion = 'latest'134    @decorators.idempotent_id('6190af80-513e-4f0f-90f2-9714e84955d7')135    @testtools.skipUnless(CONF.compute_feature_enabled.serial_console,136                          'Serial console not supported.')137    @testtools.skipUnless(138        compute.is_scheduler_filter_enabled("DifferentHostFilter"),139        'DifferentHostFilter is not available.')140    def test_live_migration_serial_console(self):141        """Test the live-migration of an instance which has a serial console142        The serial console feature of an instance uses ports on the host.143        These ports need to be updated when they are already in use by144        another instance on the target host. This test checks if this145        update behavior is correctly done, by connecting to the serial146        consoles of the instances before and after the live migration.147        """148        server01_id = self.create_test_server(wait_until='ACTIVE')['id']149        hints = {'different_host': server01_id}150        server02_id = self.create_test_server(scheduler_hints=hints,151                                              wait_until='ACTIVE')['id']152        host01_id = self.get_host_for_server(server01_id)153        host02_id = self.get_host_for_server(server02_id)154        self.assertNotEqual(host01_id, host02_id)155        # At this step we have 2 instances on different hosts, both with156        # serial consoles, both with port 10000 (the default value).157        # https://bugs.launchpad.net/nova/+bug/1455252 describes the issue158        # when live-migrating in such a scenario.159        self._verify_console_interaction(server01_id)160        self._verify_console_interaction(server02_id)161        self._migrate_server_to(server01_id, host02_id)162        waiters.wait_for_server_status(self.servers_client,163                                       server01_id, 'ACTIVE')164        self.assertEqual(host02_id, self.get_host_for_server(server01_id))165        self._verify_console_interaction(server01_id)166        # At this point, both instances have a valid serial console167        # connection, which means the ports got updated.168    def _verify_console_interaction(self, server_id):169        body = self.servers_client.get_remote_console(server_id,170                                                      console_type='serial',171                                                      protocol='serial')172        console_url = body['remote_console']['url']173        data = "test_live_migration_serial_console"174        console_output = ''175        t = 0.0176        interval = 0.1177        ws = compute.create_websocket(console_url)178        try:...

test_servers_on_multinodes.py

Source:test_servers_on_multinodes.py

...21    @classmethod22    def resource_setup(cls):23        super(ServersOnMultiNodesTest, cls).resource_setup()24        cls.server01 = cls.create_test_server(wait_until='ACTIVE')['id']25        cls.host01 = cls.get_host_for_server(cls.server01)26    @classmethod27    def skip_checks(cls):28        super(ServersOnMultiNodesTest, cls).skip_checks()29        if CONF.compute.min_compute_nodes < 2:30            raise cls.skipException(31                "Less than 2 compute nodes, skipping multi-nodes test.")32    def _create_servers_with_group(self, policy):33        group_id = self.create_test_server_group(policy=[policy])['id']34        hints = {'group': group_id}35        reservation_id = self.create_test_server(36            scheduler_hints=hints, wait_until='ACTIVE', min_count=2,37            return_reservation_id=True)['reservation_id']38        # Get the servers using the reservation_id.39        servers = self.servers_client.list_servers(40            detail=True, reservation_id=reservation_id)['servers']41        self.assertEqual(2, len(servers))42        # Assert the servers are in the group.43        server_group = self.server_groups_client.show_server_group(44            group_id)['server_group']45        hosts = {}46        for server in servers:47            self.assertIn(server['id'], server_group['members'])48            hosts[server['id']] = self.get_host_for_server(server['id'])49        return hosts50    @decorators.idempotent_id('26a9d5df-6890-45f2-abc4-a659290cb130')51    @testtools.skipUnless(52        compute.is_scheduler_filter_enabled("SameHostFilter"),53        'SameHostFilter is not available.')54    def test_create_servers_on_same_host(self):55        hints = {'same_host': self.server01}56        server02 = self.create_test_server(scheduler_hints=hints,57                                           wait_until='ACTIVE')['id']58        host02 = self.get_host_for_server(server02)59        self.assertEqual(self.host01, host02)60    @decorators.idempotent_id('cc7ca884-6e3e-42a3-a92f-c522fcf25e8e')61    @testtools.skipUnless(62        compute.is_scheduler_filter_enabled("DifferentHostFilter"),63        'DifferentHostFilter is not available.')64    def test_create_servers_on_different_hosts(self):65        hints = {'different_host': self.server01}66        server02 = self.create_test_server(scheduler_hints=hints,67                                           wait_until='ACTIVE')['id']68        host02 = self.get_host_for_server(server02)69        self.assertNotEqual(self.host01, host02)70    @decorators.idempotent_id('7869cc84-d661-4e14-9f00-c18cdc89cf57')71    @testtools.skipUnless(72        compute.is_scheduler_filter_enabled("DifferentHostFilter"),73        'DifferentHostFilter is not available.')74    def test_create_servers_on_different_hosts_with_list_of_servers(self):75        # This scheduler-hint supports list of servers also.76        hints = {'different_host': [self.server01]}77        server02 = self.create_test_server(scheduler_hints=hints,78                                           wait_until='ACTIVE')['id']79        host02 = self.get_host_for_server(server02)80        self.assertNotEqual(self.host01, host02)81    @decorators.idempotent_id('f8bd0867-e459-45f5-ba53-59134552fe04')82    @testtools.skipUnless(83        compute.is_scheduler_filter_enabled("ServerGroupAntiAffinityFilter"),84        'ServerGroupAntiAffinityFilter is not available.')85    def test_create_server_with_scheduler_hint_group_anti_affinity(self):86        """Tests the ServerGroupAntiAffinityFilter87        Creates two servers in an anti-affinity server group and88        asserts the servers are in the group and on different hosts.89        """90        hosts = self._create_servers_with_group('anti-affinity')91        hostnames = list(hosts.values())92        self.assertNotEqual(hostnames[0], hostnames[1],93                            'Servers are on the same host: %s' % hosts)...

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.