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

How to use _boot_vm_with_min_bw method in tempest

Best Python code snippet using tempest_python

test_network_qos_placement.py

Source:test_network_qos_placement.py

...157                                  self.PLACEMENT_MAX_INT))158        if len(alloc_candidates['provider_summaries']) != 0:159            self.fail('For %s:%s there should be no available candidate!' %160                      (self.INGRESS_RESOURCE_CLASS, self.PLACEMENT_MAX_INT))161    def _boot_vm_with_min_bw(self, qos_policy_id, status='ACTIVE'):162        wait_until = (None if status == 'ERROR' else status)163        port = self.create_port(164            self.prov_network['id'], qos_policy_id=qos_policy_id)165        server = self.create_server(networks=[{'port': port['id']}],166                                    wait_until=wait_until)167        waiters.wait_for_server_status(168            client=self.servers_client, server_id=server['id'],169            status=status, ready_wait=False, raise_on_error=False)170        return server, port171    def _assert_allocation_is_as_expected(172        self, consumer, port_ids, min_kbps=SMALLEST_POSSIBLE_BW,173        expected_rc=NetworkQoSPlacementTestBase.INGRESS_RESOURCE_CLASS,174    ):175        allocations = self.placement_client.list_allocations(176            consumer)['allocations']177        self.assertGreater(len(allocations), 0)178        bw_resource_in_alloc = False179        allocation_rp = None180        for rp, resources in allocations.items():181            if expected_rc in resources['resources']:182                self.assertEqual(183                    min_kbps,184                    resources['resources'][expected_rc])185                bw_resource_in_alloc = True186                allocation_rp = rp187        if min_kbps:188            self.assertTrue(189                bw_resource_in_alloc,190                f"expected {min_kbps} bandwidth allocation from {expected_rc} "191                f"but instance has allocation {allocations} instead."192            )193            # Check binding_profile of the port is not empty and equals with194            # the rp uuid195            for port_id in port_ids:196                port = self.os_admin.ports_client.show_port(port_id)197                port_binding_alloc = port['port']['binding:profile'][198                    'allocation']199                # NOTE(gibi): the format of the allocation key depends on the200                # existence of port-resource-request-groups API extension.201                # TODO(gibi): drop the else branch once tempest does not need202                # to support Xena release any more.203                if utils.is_extension_enabled(204                        'port-resource-request-groups', 'network'):205                    self.assertEqual(206                        {allocation_rp},207                        set(port_binding_alloc.values()))208                else:209                    self.assertEqual(allocation_rp, port_binding_alloc)210    @decorators.idempotent_id('78625d92-212c-400e-8695-dd51706858b8')211    @utils.services('compute', 'network')212    def test_qos_min_bw_allocation_basic(self):213        """"Basic scenario with QoS min bw allocation in placement.214        Steps:215        * Create prerequisites:216        ** VLAN type provider network with subnet.217        ** valid QoS policy with minimum bandwidth rule with min_kbps=1218        (This is a simplification to skip the checks in placement for219        detecting the resource provider tree and inventories, as if220        bandwidth resource is available 1 kbs will be available).221        ** invalid QoS policy with minimum bandwidth rule with222        min_kbs=max integer from placement (this is a simplification again223        to avoid detection of RP tress and inventories, as placement will224        reject such big allocation).225        * Create port with valid QoS policy, and boot VM with that, it should226        pass.227        * Create port with invalid QoS policy, and try to boot VM with that,228        it should fail.229        """230        self._create_network_and_qos_policies(self._create_qos_basic_policies)231        server1, valid_port = self._boot_vm_with_min_bw(232            qos_policy_id=self.qos_policy_valid['id'])233        self._assert_allocation_is_as_expected(server1['id'],234                                               [valid_port['id']])235        server2, not_valid_port = self._boot_vm_with_min_bw(236            self.qos_policy_not_valid['id'], status='ERROR')237        allocations = self.placement_client.list_allocations(server2['id'])238        self.assertEqual(0, len(allocations['allocations']))239        server2 = self.servers_client.show_server(server2['id'])240        self.assertIn('fault', server2['server'])241        self.assertIn('No valid host', server2['server']['fault']['message'])242        # Check that binding_profile of the port is empty243        port = self.os_admin.ports_client.show_port(not_valid_port['id'])244        self.assertEqual(0, len(port['port']['binding:profile']))245    @decorators.idempotent_id('8a98150c-a506-49a5-96c6-73a5e7b04ada')246    @testtools.skipUnless(CONF.compute_feature_enabled.cold_migration,247                          'Cold migration is not available.')248    @testtools.skipUnless(CONF.compute.min_compute_nodes > 1,249                          'Less than 2 compute nodes, skipping multinode '250                          'tests.')251    @utils.services('compute', 'network')252    def test_migrate_with_qos_min_bw_allocation(self):253        """Scenario to migrate VM with QoS min bw allocation in placement254        Boot a VM like in test_qos_min_bw_allocation_basic, do the same255        checks, and256        * migrate the server257        * confirm the resize, if the VM state is VERIFY_RESIZE258        * If the VM goes to ACTIVE state check that allocations are as259        expected.260        """261        self._create_network_and_qos_policies(self._create_qos_basic_policies)262        server, valid_port = self._boot_vm_with_min_bw(263            qos_policy_id=self.qos_policy_valid['id'])264        self._assert_allocation_is_as_expected(server['id'],265                                               [valid_port['id']])266        self.os_adm.servers_client.migrate_server(server_id=server['id'])267        waiters.wait_for_server_status(268            client=self.servers_client, server_id=server['id'],269            status='VERIFY_RESIZE', ready_wait=False, raise_on_error=False)270        # TODO(lajoskatona): Check that the allocations are ok for the271        #  migration?272        self._assert_allocation_is_as_expected(server['id'],273                                               [valid_port['id']])274        self.os_adm.servers_client.confirm_resize_server(275            server_id=server['id'])276        waiters.wait_for_server_status(277            client=self.servers_client, server_id=server['id'],278            status='ACTIVE', ready_wait=False, raise_on_error=True)279        self._assert_allocation_is_as_expected(server['id'],280                                               [valid_port['id']])281    @decorators.idempotent_id('c29e7fd3-035d-4993-880f-70819847683f')282    @testtools.skipUnless(CONF.compute_feature_enabled.resize,283                          'Resize not available.')284    @utils.services('compute', 'network')285    def test_resize_with_qos_min_bw_allocation(self):286        """Scenario to resize VM with QoS min bw allocation in placement.287        Boot a VM like in test_qos_min_bw_allocation_basic, do the same288        checks, and289        * resize the server with new flavor290        * confirm the resize, if the VM state is VERIFY_RESIZE291        * If the VM goes to ACTIVE state check that allocations are as292        expected.293        """294        self._create_network_and_qos_policies(self._create_qos_basic_policies)295        server, valid_port = self._boot_vm_with_min_bw(296            qos_policy_id=self.qos_policy_valid['id'])297        self._assert_allocation_is_as_expected(server['id'],298                                               [valid_port['id']])299        new_flavor = self._create_flavor_to_resize_to()300        self.servers_client.resize_server(301            server_id=server['id'], flavor_ref=new_flavor['id'])302        waiters.wait_for_server_status(303            client=self.servers_client, server_id=server['id'],304            status='VERIFY_RESIZE', ready_wait=False, raise_on_error=False)305        # TODO(lajoskatona): Check that the allocations are ok for the306        #  migration?307        self._assert_allocation_is_as_expected(server['id'],308                                               [valid_port['id']])309        self.servers_client.confirm_resize_server(server_id=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.