KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference
How to use _create_flavor_to_resize_to method in tempest

Best Python code snippet using tempest_python
test_network_qos_placement.py
Source:test_network_qos_placement.py
...60        cls.qos_client = cls.os_admin.qos_client61        cls.qos_min_bw_client = cls.os_admin.qos_min_bw_client62        cls.flavors_client = cls.os_adm.flavors_client63        cls.servers_client = cls.os_primary.servers_client64    def _create_flavor_to_resize_to(self):65        old_flavor = self.flavors_client.show_flavor(66            CONF.compute.flavor_ref)['flavor']67        new_flavor = self.flavors_client.create_flavor(**{68            'ram': old_flavor['ram'],69            'vcpus': old_flavor['vcpus'],70            'name': old_flavor['name'] + 'extra-%s' % data_utils.rand_int_id(),71            'disk': old_flavor['disk'] + 172        })['flavor']73        self.addCleanup(test_utils.call_and_ignore_notfound_exc,74                        self.flavors_client.delete_flavor, new_flavor['id'])75        return new_flavor76class MinBwAllocationPlacementTest(NetworkQoSPlacementTestBase):77    required_extensions = ['port-resource-request',78                           'qos',79                           'qos-bw-minimum-ingress']80    SMALLEST_POSSIBLE_BW = 181    BANDWIDTH_1 = 100082    BANDWIDTH_2 = 200083    @classmethod84    def skip_checks(cls):85        super(MinBwAllocationPlacementTest, cls).skip_checks()86        if not CONF.network_feature_enabled.qos_placement_physnet:87            msg = "Skipped as no physnet is available in config for " \88                  "placement based QoS allocation."89            raise cls.skipException(msg)90    def setUp(self):91        super(MinBwAllocationPlacementTest, self).setUp()92        self._check_if_allocation_is_possible()93    def _create_policy_and_min_bw_rule(94        self, name_prefix, min_kbps, direction="ingress"95    ):96        policy = self.qos_client.create_qos_policy(97            name=data_utils.rand_name(name_prefix),98            shared=True)['policy']99        self.addCleanup(test_utils.call_and_ignore_notfound_exc,100                        self.qos_client.delete_qos_policy, policy['id'])101        rule = self.qos_min_bw_client.create_minimum_bandwidth_rule(102            policy['id'],103            **{104                'min_kbps': min_kbps,105                'direction': direction,106            })['minimum_bandwidth_rule']107        self.addCleanup(108            test_utils.call_and_ignore_notfound_exc,109            self.qos_min_bw_client.delete_minimum_bandwidth_rule, policy['id'],110            rule['id'])111        return policy112    def _create_qos_basic_policies(self):113        self.qos_policy_valid = self._create_policy_and_min_bw_rule(114            name_prefix='test_policy_valid',115            min_kbps=self.SMALLEST_POSSIBLE_BW)116        self.qos_policy_not_valid = self._create_policy_and_min_bw_rule(117            name_prefix='test_policy_not_valid',118            min_kbps=self.PLACEMENT_MAX_INT)119    def _create_qos_policies_from_life(self):120        # For tempest-slow the max bandwidth configured is 1000000,121        # https://opendev.org/openstack/tempest/src/branch/master/122        # .zuul.yaml#L416-L420123        self.qos_policy_1 = self._create_policy_and_min_bw_rule(124            name_prefix='test_policy_1',125            min_kbps=self.BANDWIDTH_1126        )127        self.qos_policy_2 = self._create_policy_and_min_bw_rule(128            name_prefix='test_policy_2',129            min_kbps=self.BANDWIDTH_2130        )131    def _create_network_and_qos_policies(self, policy_method):132        physnet_name = CONF.network_feature_enabled.qos_placement_physnet133        base_segm = \134            CONF.network_feature_enabled.provider_net_base_segmentation_id135        self.prov_network, _, _ = self.setup_network_subnet_with_router(136            networks_client=self.networks_client,137            routers_client=self.routers_client,138            subnets_client=self.subnets_client,139            **{140                'shared': True,141                'provider:network_type': 'vlan',142                'provider:physical_network': physnet_name,143                'provider:segmentation_id': base_segm144            })145        policy_method()146    def _check_if_allocation_is_possible(self):147        alloc_candidates = self.placement_client.list_allocation_candidates(148            resources1='%s:%s' % (self.INGRESS_RESOURCE_CLASS,149                                  self.SMALLEST_POSSIBLE_BW))150        if len(alloc_candidates['provider_summaries']) == 0:151            self.fail('No allocation candidates are available for %s:%s' %152                      (self.INGRESS_RESOURCE_CLASS, self.SMALLEST_POSSIBLE_BW))153        # Just to be sure check with impossible high (placement max_int),154        # allocation155        alloc_candidates = self.placement_client.list_allocation_candidates(156            resources1='%s:%s' % (self.INGRESS_RESOURCE_CLASS,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'])310        waiters.wait_for_server_status(311            client=self.servers_client, server_id=server['id'],312            status='ACTIVE', ready_wait=False, raise_on_error=True)313        self._assert_allocation_is_as_expected(server['id'],314                                               [valid_port['id']])315    @decorators.idempotent_id('79fdaa1c-df62-4738-a0f0-1cff9dc415f6')316    @utils.services('compute', 'network')317    def test_qos_min_bw_allocation_update_policy(self):318        """Test the update of QoS policy on bound port319        Related RFE in neutron: #1882804320        The scenario is the following:321        * Have a port with QoS policy and minimum bandwidth rule.322        * Boot a VM with the port.323        * Update the port with a new policy with different minimum bandwidth324        values.325        * The allocation on placement side should be according to the new326        rules.327        """328        if not utils.is_network_feature_enabled('update_port_qos'):329            raise self.skipException("update_port_qos feature is not enabled")330        self._create_network_and_qos_policies(331            self._create_qos_policies_from_life)332        port = self.create_port(333            self.prov_network['id'], qos_policy_id=self.qos_policy_1['id'])334        server1 = self.create_server(335            networks=[{'port': port['id']}])336        self._assert_allocation_is_as_expected(server1['id'], [port['id']],337                                               self.BANDWIDTH_1)338        self.ports_client.update_port(339            port['id'],340            **{'qos_policy_id': self.qos_policy_2['id']})341        self._assert_allocation_is_as_expected(server1['id'], [port['id']],342                                               self.BANDWIDTH_2)343        # I changed my mind344        self.ports_client.update_port(345            port['id'],346            **{'qos_policy_id': self.qos_policy_1['id']})347        self._assert_allocation_is_as_expected(server1['id'], [port['id']],348                                               self.BANDWIDTH_1)349        # bad request....350        self.qos_policy_not_valid = self._create_policy_and_min_bw_rule(351            name_prefix='test_policy_not_valid',352            min_kbps=self.PLACEMENT_MAX_INT)353        port_orig = self.ports_client.show_port(port['id'])['port']354        self.assertRaises(355            lib_exc.Conflict,356            self.ports_client.update_port,357            port['id'], **{'qos_policy_id': self.qos_policy_not_valid['id']})358        self._assert_allocation_is_as_expected(server1['id'], [port['id']],359                                               self.BANDWIDTH_1)360        port_upd = self.ports_client.show_port(port['id'])['port']361        self.assertEqual(port_orig['qos_policy_id'],362                         port_upd['qos_policy_id'])363        self.assertEqual(self.qos_policy_1['id'], port_upd['qos_policy_id'])364    @decorators.idempotent_id('9cfc3bb8-f433-4c91-87b6-747cadc8958a')365    @utils.services('compute', 'network')366    def test_qos_min_bw_allocation_update_policy_from_zero(self):367        """Test port without QoS policy to have QoS policy368        This scenario checks if updating a port without QoS policy to369        have QoS policy with minimum_bandwidth rule succeeds only on370        controlplane, but placement allocation remains 0.371        """372        if not utils.is_network_feature_enabled('update_port_qos'):373            raise self.skipException("update_port_qos feature is not enabled")374        self._create_network_and_qos_policies(375            self._create_qos_policies_from_life)376        port = self.create_port(self.prov_network['id'])377        server1 = self.create_server(378            networks=[{'port': port['id']}])379        self._assert_allocation_is_as_expected(server1['id'], [port['id']], 0)380        self.ports_client.update_port(381            port['id'], **{'qos_policy_id': self.qos_policy_2['id']})382        self._assert_allocation_is_as_expected(server1['id'], [port['id']], 0)383    @decorators.idempotent_id('a9725a70-1d28-4e3b-ae0e-450abc235962')384    @utils.services('compute', 'network')385    def test_qos_min_bw_allocation_update_policy_to_zero(self):386        """Test port with QoS policy to remove QoS policy387        In this scenario port with QoS minimum_bandwidth rule update to388        remove QoS policy results in 0 placement allocation.389        """390        if not utils.is_network_feature_enabled('update_port_qos'):391            raise self.skipException("update_port_qos feature is not enabled")392        self._create_network_and_qos_policies(393            self._create_qos_policies_from_life)394        port = self.create_port(395            self.prov_network['id'], qos_policy_id=self.qos_policy_1['id'])396        server1 = self.create_server(397            networks=[{'port': port['id']}])398        self._assert_allocation_is_as_expected(server1['id'], [port['id']],399                                               self.BANDWIDTH_1)400        self.ports_client.update_port(401            port['id'],402            **{'qos_policy_id': None})403        self._assert_allocation_is_as_expected(server1['id'], [port['id']], 0)404    @decorators.idempotent_id('756ced7f-6f1a-43e7-a851-2fcfc16f3dd7')405    @utils.services('compute', 'network')406    def test_qos_min_bw_allocation_update_with_multiple_ports(self):407        if not utils.is_network_feature_enabled('update_port_qos'):408            raise self.skipException("update_port_qos feature is not enabled")409        self._create_network_and_qos_policies(410            self._create_qos_policies_from_life)411        port1 = self.create_port(412            self.prov_network['id'], qos_policy_id=self.qos_policy_1['id'])413        port2 = self.create_port(414            self.prov_network['id'], qos_policy_id=self.qos_policy_2['id'])415        server1 = self.create_server(416            networks=[{'port': port1['id']}, {'port': port2['id']}])417        self._assert_allocation_is_as_expected(418            server1['id'], [port1['id'], port2['id']],419            self.BANDWIDTH_1 + self.BANDWIDTH_2)420        self.ports_client.update_port(421            port1['id'],422            **{'qos_policy_id': self.qos_policy_2['id']})423        self._assert_allocation_is_as_expected(424            server1['id'], [port1['id'], port2['id']],425            2 * self.BANDWIDTH_2)426    @decorators.idempotent_id('0805779e-e03c-44fb-900f-ce97a790653b')427    @utils.services('compute', 'network')428    def test_empty_update(self):429        if not utils.is_network_feature_enabled('update_port_qos'):430            raise self.skipException("update_port_qos feature is not enabled")431        self._create_network_and_qos_policies(432            self._create_qos_policies_from_life)433        port = self.create_port(434            self.prov_network['id'], qos_policy_id=self.qos_policy_1['id'])435        server1 = self.create_server(436            networks=[{'port': port['id']}])437        self._assert_allocation_is_as_expected(server1['id'], [port['id']],438                                               self.BANDWIDTH_1)439        self.ports_client.update_port(440            port['id'],441            **{'description': 'foo'})442        self._assert_allocation_is_as_expected(server1['id'], [port['id']],443                                               self.BANDWIDTH_1)444    @decorators.idempotent_id('372b2728-cfed-469a-b5f6-b75779e1ccbe')445    @utils.services('compute', 'network')446    def test_qos_min_bw_allocation_update_policy_direction_change(self):447        """Test QoS min bw direction change on a bound port448        Related RFE in neutron: #1882804449        The scenario is the following:450        * Have a port with QoS policy and minimum bandwidth rule with ingress451        direction452        * Boot a VM with the port.453        * Update the port with a new policy to egress direction in454        minimum bandwidth rule.455        * The allocation on placement side should be according to the new456        rules.457        """458        if not utils.is_network_feature_enabled('update_port_qos'):459            raise self.skipException("update_port_qos feature is not enabled")460        def create_policies():461            self.qos_policy_ingress = self._create_policy_and_min_bw_rule(462                name_prefix='test_policy_ingress',463                min_kbps=self.BANDWIDTH_1,464                direction=self.INGRESS_DIRECTION,465            )466            self.qos_policy_egress = self._create_policy_and_min_bw_rule(467                name_prefix='test_policy_egress',468                min_kbps=self.BANDWIDTH_1,469                direction=self.EGRESS_DIRECTION,470            )471        self._create_network_and_qos_policies(create_policies)472        port = self.create_port(473            self.prov_network['id'],474            qos_policy_id=self.qos_policy_ingress['id'])475        server1 = self.create_server(476            networks=[{'port': port['id']}])477        self._assert_allocation_is_as_expected(478            server1['id'], [port['id']], self.BANDWIDTH_1,479            expected_rc=self.INGRESS_RESOURCE_CLASS)480        self.ports_client.update_port(481            port['id'],482            qos_policy_id=self.qos_policy_egress['id'])483        self._assert_allocation_is_as_expected(484            server1['id'], [port['id']], self.BANDWIDTH_1,485            expected_rc=self.EGRESS_RESOURCE_CLASS)486        self._assert_allocation_is_as_expected(487            server1['id'], [port['id']], 0,488            expected_rc=self.INGRESS_RESOURCE_CLASS)489class QoSBandwidthAndPacketRateTests(NetworkQoSPlacementTestBase):490    PPS_RESOURCE_CLASS = "NET_PACKET_RATE_KILOPACKET_PER_SEC"491    @classmethod492    def skip_checks(cls):493        super().skip_checks()494        if not CONF.network_feature_enabled.qos_min_bw_and_pps:495            msg = (496                "Skipped as no resource inventories are configured for QoS "497                "minimum bandwidth and packet rate testing.")498            raise cls.skipException(msg)499    @classmethod500    def setup_clients(cls):501        super().setup_clients()502        cls.qos_min_pps_client = cls.os_admin.qos_min_pps_client503    def setUp(self):504        super().setUp()505        self.network = self._create_network()506    def _create_qos_policy_with_bw_and_pps_rules(self, min_kbps, min_kpps):507        policy = self.qos_client.create_qos_policy(508            name=data_utils.rand_name(),509            shared=True510        )['policy']511        self.addCleanup(512            test_utils.call_and_ignore_notfound_exc,513            self.qos_client.delete_qos_policy,514            policy['id']515        )516        if min_kbps > 0:517            bw_rule = self.qos_min_bw_client.create_minimum_bandwidth_rule(518                policy['id'],519                min_kbps=min_kbps,520                direction=self.INGRESS_DIRECTION521            )['minimum_bandwidth_rule']522            self.addCleanup(523                test_utils.call_and_ignore_notfound_exc,524                self.qos_min_bw_client.delete_minimum_bandwidth_rule,525                policy['id'],526                bw_rule['id']527            )528        if min_kpps > 0:529            pps_rule = self.qos_min_pps_client.create_minimum_packet_rate_rule(530                policy['id'],531                min_kpps=min_kpps,532                direction=self.ANY_DIRECTION533            )['minimum_packet_rate_rule']534            self.addCleanup(535                test_utils.call_and_ignore_notfound_exc,536                self.qos_min_pps_client.delete_minimum_packet_rate_rule,537                policy['id'],538                pps_rule['id']539            )540        return policy541    def _create_network(self):542        physnet_name = CONF.network_feature_enabled.qos_placement_physnet543        base_segm = (544            CONF.network_feature_enabled.provider_net_base_segmentation_id)545        # setup_network_subnet_with_router will add the necessary cleanup calls546        network, _, _ = self.setup_network_subnet_with_router(547            networks_client=self.networks_client,548            routers_client=self.routers_client,549            subnets_client=self.subnets_client,550            shared=True,551            **{552                'provider:network_type': 'vlan',553                'provider:physical_network': physnet_name,554                # +1 to be different from the segmentation_id used in555                # MinBwAllocationPlacementTest556                'provider:segmentation_id': int(base_segm) + 1,557            }558        )559        return network560    def _create_port_with_qos_policy(self, policy):561        port = self.ports_client.create_port(562            name=data_utils.rand_name(self.__class__.__name__),563            network_id=self.network['id'],564            qos_policy_id=policy['id'] if policy else None,565        )['port']566        self.addCleanup(567            test_utils.call_and_ignore_notfound_exc,568            self.ports_client.delete_port, port['id']569        )570        return port571    def assert_allocations(572            self, server, port, expected_min_kbps, expected_min_kpps573    ):574        allocations = self.placement_client.list_allocations(575            server['id'])['allocations']576        # one allocation for the flavor related resources on the compute RP577        expected_allocation = 1578        # one allocation due to bw rule579        if expected_min_kbps > 0:580            expected_allocation += 1581        # one allocation due to pps rule582        if expected_min_kpps > 0:583            expected_allocation += 1584        self.assertEqual(expected_allocation, len(allocations), allocations)585        expected_rp_uuids_in_binding_allocation = set()586        if expected_min_kbps > 0:587            bw_rp_allocs = {588                rp: alloc['resources'][self.INGRESS_RESOURCE_CLASS]589                for rp, alloc in allocations.items()590                if self.INGRESS_RESOURCE_CLASS in alloc['resources']591            }592            self.assertEqual(1, len(bw_rp_allocs))593            bw_rp, bw_alloc = list(bw_rp_allocs.items())[0]594            self.assertEqual(expected_min_kbps, bw_alloc)595            expected_rp_uuids_in_binding_allocation.add(bw_rp)596        if expected_min_kpps > 0:597            pps_rp_allocs = {598                rp: alloc['resources'][self.PPS_RESOURCE_CLASS]599                for rp, alloc in allocations.items()600                if self.PPS_RESOURCE_CLASS in alloc['resources']601            }602            self.assertEqual(1, len(pps_rp_allocs))603            pps_rp, pps_alloc = list(pps_rp_allocs.items())[0]604            self.assertEqual(expected_min_kpps, pps_alloc)605            expected_rp_uuids_in_binding_allocation.add(pps_rp)606        # Let's check port.binding:profile.allocation points to the two607        # provider resource allocated from608        port = self.os_admin.ports_client.show_port(port['id'])609        port_binding_alloc = port[610            'port']['binding:profile'].get('allocation', {})611        self.assertEqual(612            expected_rp_uuids_in_binding_allocation,613            set(port_binding_alloc.values())614        )615    def assert_no_allocation(self, server, port):616        # check that there are no allocations617        allocations = self.placement_client.list_allocations(618            server['id'])['allocations']619        self.assertEqual(0, len(allocations))620        # check that binding_profile of the port is empty621        port = self.os_admin.ports_client.show_port(port['id'])622        self.assertEqual(0, len(port['port']['binding:profile']))623    @decorators.idempotent_id('93d1a88d-235e-4b7b-b44d-2a17dcf4e213')624    @utils.services('compute', 'network')625    def test_server_create_delete(self):626        min_kbps = 1000627        min_kpps = 100628        policy = self._create_qos_policy_with_bw_and_pps_rules(629            min_kbps, min_kpps)630        port = self._create_port_with_qos_policy(policy)631        server = self.create_server(632            networks=[{'port': port['id']}],633            wait_until='ACTIVE'634        )635        self.assert_allocations(server, port, min_kbps, min_kpps)636        self.servers_client.delete_server(server['id'])637        waiters.wait_for_server_termination(self.servers_client, server['id'])638        self.assert_no_allocation(server, port)639    def _test_create_server_negative(self, min_kbps=1000, min_kpps=100):640        policy = self._create_qos_policy_with_bw_and_pps_rules(641            min_kbps, min_kpps)642        port = self._create_port_with_qos_policy(policy)643        server = self.create_server(644            networks=[{'port': port['id']}],645            wait_until=None)646        waiters.wait_for_server_status(647            client=self.servers_client, server_id=server['id'],648            status='ERROR', ready_wait=False, raise_on_error=False)649        # check that the creation failed with No valid host650        server = self.servers_client.show_server(server['id'])['server']651        self.assertIn('fault', server)652        self.assertIn('No valid host', server['fault']['message'])653        self.assert_no_allocation(server, port)654    @decorators.idempotent_id('915dd2ce-4890-40c8-9db6-f3e04080c6c1')655    @utils.services('compute', 'network')656    def test_server_create_no_valid_host_due_to_bandwidth(self):657        self._test_create_server_negative(min_kbps=self.PLACEMENT_MAX_INT)658    @decorators.idempotent_id('2d4a755e-10b9-4ac0-bef2-3f89de1f150b')659    @utils.services('compute', 'network')660    def test_server_create_no_valid_host_due_to_packet_rate(self):661        self._test_create_server_negative(min_kpps=self.PLACEMENT_MAX_INT)662    @decorators.idempotent_id('69d93e4f-0dfc-4d17-8d84-cc5c3c842cd5')663    @testtools.skipUnless(664        CONF.compute_feature_enabled.resize, 'Resize not available.')665    @utils.services('compute', 'network')666    def test_server_resize(self):667        min_kbps = 1000668        min_kpps = 100669        policy = self._create_qos_policy_with_bw_and_pps_rules(670            min_kbps, min_kpps)671        port = self._create_port_with_qos_policy(policy)672        server = self.create_server(673            networks=[{'port': port['id']}],674            wait_until='ACTIVE'675        )676        self.assert_allocations(server, port, min_kbps, min_kpps)677        new_flavor = self._create_flavor_to_resize_to()678        self.servers_client.resize_server(679            server_id=server['id'], flavor_ref=new_flavor['id']680        )681        waiters.wait_for_server_status(682            client=self.servers_client, server_id=server['id'],683            status='VERIFY_RESIZE', ready_wait=False, raise_on_error=False)684        self.assert_allocations(server, port, min_kbps, min_kpps)685        self.servers_client.confirm_resize_server(server_id=server['id'])686        waiters.wait_for_server_status(687            client=self.servers_client, server_id=server['id'],688            status='ACTIVE', ready_wait=False, raise_on_error=True)689        self.assert_allocations(server, port, min_kbps, min_kpps)690    @decorators.idempotent_id('d01d4aee-ca06-4e4e-add7-8a47fe0daf96')691    @testtools.skipUnless(692        CONF.compute_feature_enabled.resize, 'Resize not available.')693    @utils.services('compute', 'network')694    def test_server_resize_revert(self):695        min_kbps = 1000696        min_kpps = 100697        policy = self._create_qos_policy_with_bw_and_pps_rules(698            min_kbps, min_kpps)699        port = self._create_port_with_qos_policy(policy)700        server = self.create_server(701            networks=[{'port': port['id']}],702            wait_until='ACTIVE'703        )704        self.assert_allocations(server, port, min_kbps, min_kpps)705        new_flavor = self._create_flavor_to_resize_to()706        self.servers_client.resize_server(707            server_id=server['id'], flavor_ref=new_flavor['id']708        )709        waiters.wait_for_server_status(710            client=self.servers_client, server_id=server['id'],711            status='VERIFY_RESIZE', ready_wait=False, raise_on_error=False)712        self.assert_allocations(server, port, min_kbps, min_kpps)713        self.servers_client.revert_resize_server(server_id=server['id'])714        waiters.wait_for_server_status(715            client=self.servers_client, server_id=server['id'],716            status='ACTIVE', ready_wait=False, raise_on_error=True)717        self.assert_allocations(server, port, min_kbps, min_kpps)718    @decorators.idempotent_id('bdd0b31c-c8b0-4b7b-b80a-545a46b32abe')719    @testtools.skipUnless(...
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.