KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use setup_network_subnet_with_router method in tempest

Best Python code snippet using tempest_python

test_network_qos_placement.py

Source:test_network_qos_placement.py

...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):...

test_security_groups_basic_ops.py

Source:test_security_groups_basic_ops.py

...278            server, public_network_id,279            client=tenant.manager.floating_ips_client)280        self.floating_ips.setdefault(server['id'], floating_ip)281    def _create_tenant_network(self, tenant, port_security_enabled=True):282        network, subnet, router = self.setup_network_subnet_with_router(283            networks_client=tenant.manager.networks_client,284            routers_client=tenant.manager.routers_client,285            subnets_client=tenant.manager.subnets_client,286            port_security_enabled=port_security_enabled)287        tenant.set_network(network, subnet, router)288    def _deploy_tenant(self, tenant_or_id):289        """creates:290            network291            subnet292            router (if public not defined)293            access security group294            access-point server295        """296        if not isinstance(tenant_or_id, self.TenantProperties):...

test_network_advanced_server_ops.py

Source:test_network_advanced_server_ops.py

...54        security_groups = []55        if utils.is_extension_enabled('security-group', 'network'):56            security_group = self.create_security_group()57            security_groups = [{'name': security_group['name']}]58        network, _, _ = self.setup_network_subnet_with_router()59        server = self.create_server(60            networks=[{'uuid': network['id']}],61            key_name=keypair['name'],62            security_groups=security_groups)63        return server64    def _setup_network(self, server, keypair):65        public_network_id = CONF.network.public_network_id66        floating_ip = self.create_floating_ip(server, public_network_id)67        # Verify that we can indeed connect to the server before we mess with68        # it's state69        self._wait_server_status_and_check_network_connectivity(70            server, keypair, floating_ip)71        return floating_ip72    def _check_network_connectivity(self, server, keypair, floating_ip,...

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.