How to use _assert_allocation_is_as_expected method in tempest

Best Python code snippet using tempest_python
test_network_qos_placement.py
Source:test_network_qos_placement.py
...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):...
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.