How to use assert_allocations method in tempest

Best Python code snippet using tempest_python

test_network_qos_placement.py

Source:test_network_qos_placement.py

...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(720        CONF.compute_feature_enabled.cold_migration,721        'Cold migration is not available.')722    @testtools.skipUnless(723        CONF.compute.min_compute_nodes > 1,724        'Less than 2 compute nodes, skipping multinode tests.')725    @utils.services('compute', 'network')726    def test_server_migrate(self):727        min_kbps = 1000728        min_kpps = 100729        policy = self._create_qos_policy_with_bw_and_pps_rules(730            min_kbps, min_kpps)731        port = self._create_port_with_qos_policy(policy)732        server = self.create_server(733            networks=[{'port': port['id']}],734            wait_until='ACTIVE'735        )736        self.assert_allocations(server, port, min_kbps, min_kpps)737        self.os_adm.servers_client.migrate_server(server_id=server['id'])738        waiters.wait_for_server_status(739            client=self.servers_client, server_id=server['id'],740            status='VERIFY_RESIZE', ready_wait=False, raise_on_error=False)741        self.assert_allocations(server, port, min_kbps, min_kpps)742        self.os_adm.servers_client.confirm_resize_server(743            server_id=server['id'])744        waiters.wait_for_server_status(745            client=self.servers_client, server_id=server['id'],746            status='ACTIVE', ready_wait=False, raise_on_error=True)747        self.assert_allocations(server, port, min_kbps, min_kpps)748    @decorators.idempotent_id('fdb260e3-caa5-482d-ac7c-8c22adf3d750')749    @utils.services('compute', 'network')750    def test_qos_policy_update_on_bound_port(self):751        min_kbps = 1000752        min_kpps = 100753        policy = self._create_qos_policy_with_bw_and_pps_rules(754            min_kbps, min_kpps)755        min_kbps2 = 2000756        min_kpps2 = 50757        policy2 = self._create_qos_policy_with_bw_and_pps_rules(758            min_kbps2, min_kpps2)759        port = self._create_port_with_qos_policy(policy)760        server = self.create_server(761            networks=[{'port': port['id']}],762            wait_until='ACTIVE'763        )764        self.assert_allocations(server, port, min_kbps, min_kpps)765        self.ports_client.update_port(766            port['id'],767            qos_policy_id=policy2['id'])768        self.assert_allocations(server, port, min_kbps2, min_kpps2)769    @decorators.idempotent_id('e6a20125-a02e-49f5-bcf6-894305ee3715')770    @utils.services('compute', 'network')771    def test_qos_policy_update_on_bound_port_from_null_policy(self):772        min_kbps = 1000773        min_kpps = 100774        policy = self._create_qos_policy_with_bw_and_pps_rules(775            min_kbps, min_kpps)776        port = self._create_port_with_qos_policy(policy=None)777        server = self.create_server(778            networks=[{'port': port['id']}],779            wait_until='ACTIVE'780        )781        self.assert_allocations(server, port, 0, 0)782        self.ports_client.update_port(783            port['id'],784            qos_policy_id=policy['id'])785        # NOTE(gibi): This is unintuitive but it is the expected behavior.786        # If there was no policy attached to the port when the server was787        # created then neutron still allows adding a policy to the port later788        # as this operation was support before placement enforcement was added789        # for the qos minimum bandwidth rule. However neutron cannot create790        # the placement resource allocation for this port.791        self.assert_allocations(server, port, 0, 0)792    @decorators.idempotent_id('f5864761-966c-4e49-b430-ac0044b7d658')793    @utils.services('compute', 'network')794    def test_qos_policy_update_on_bound_port_additional_rule(self):795        min_kbps = 1000796        policy = self._create_qos_policy_with_bw_and_pps_rules(797            min_kbps, 0)798        min_kbps2 = 2000799        min_kpps2 = 50800        policy2 = self._create_qos_policy_with_bw_and_pps_rules(801            min_kbps2, min_kpps2)802        port = self._create_port_with_qos_policy(policy=policy)803        server = self.create_server(804            networks=[{'port': port['id']}],805            wait_until='ACTIVE'806        )807        self.assert_allocations(server, port, min_kbps, 0)808        self.ports_client.update_port(809            port['id'],810            qos_policy_id=policy2['id'])811        # FIXME(gibi): Agree in the spec: do we ignore the pps request or we812        # reject the update? It seems current implementation goes with813        # ignoring the additional pps rule.814        self.assert_allocations(server, port, min_kbps2, 0)815    @decorators.idempotent_id('fbbb9c81-ed21-48c3-bdba-ce2361e93aad')816    @utils.services('compute', 'network')817    def test_qos_policy_update_on_bound_port_to_null_policy(self):818        min_kbps = 1000819        min_kpps = 100820        policy = self._create_qos_policy_with_bw_and_pps_rules(821            min_kbps, min_kpps)822        port = self._create_port_with_qos_policy(policy=policy)823        server = self.create_server(824            networks=[{'port': port['id']}],825            wait_until='ACTIVE'826        )827        self.assert_allocations(server, port, min_kbps, min_kpps)828        self.ports_client.update_port(829            port['id'],830            qos_policy_id=None)831        self.assert_allocations(server, port, 0, 0)832    @decorators.idempotent_id('0393d038-03ad-4844-a0e4-83010f69dabb')833    @utils.services('compute', 'network')834    def test_interface_attach_detach(self):835        min_kbps = 1000836        min_kpps = 100837        policy = self._create_qos_policy_with_bw_and_pps_rules(838            min_kbps, min_kpps)839        port = self._create_port_with_qos_policy(policy=None)840        port2 = self._create_port_with_qos_policy(policy=policy)841        server = self.create_server(842            networks=[{'port': port['id']}],843            wait_until='ACTIVE'844        )845        self.assert_allocations(server, port, 0, 0)846        self.interface_client.create_interface(847            server_id=server['id'],848            port_id=port2['id'])849        waiters.wait_for_interface_status(850            self.interface_client, server['id'], port2['id'], 'ACTIVE')851        self.assert_allocations(server, port2, min_kbps, min_kpps)852        req_id = self.interface_client.delete_interface(853            server_id=server['id'],854            port_id=port2['id']).response['x-openstack-request-id']855        waiters.wait_for_interface_detach(856            self.servers_client, server['id'], port2['id'], req_id)857        self.assert_allocations(server, port2, 0, 0)858    @decorators.idempotent_id('36ffdb85-6cc2-4cc9-a426-cad5bac8626b')859    @testtools.skipUnless(860        CONF.compute.min_compute_nodes > 1,861        'Less than 2 compute nodes, skipping multinode tests.')862    @testtools.skipUnless(863        CONF.compute_feature_enabled.live_migration,864        'Live migration not available')865    @utils.services('compute', 'network')866    def test_server_live_migrate(self):867        min_kbps = 1000868        min_kpps = 100869        policy = self._create_qos_policy_with_bw_and_pps_rules(870            min_kbps, min_kpps)871        port = self._create_port_with_qos_policy(policy=policy)872        server = self.create_server(873            networks=[{'port': port['id']}],874            wait_until='ACTIVE'875        )876        self.assert_allocations(server, port, min_kbps, min_kpps)877        server_details = self.os_adm.servers_client.show_server(server['id'])878        source_host = server_details['server']['OS-EXT-SRV-ATTR:host']879        self.os_adm.servers_client.live_migrate_server(880            server['id'], block_migration=True, host=None)881        waiters.wait_for_server_status(882            self.servers_client, server['id'], 'ACTIVE')883        server_details = self.os_adm.servers_client.show_server(server['id'])884        new_host = server_details['server']['OS-EXT-SRV-ATTR:host']885        self.assertNotEqual(source_host, new_host, "Live migration failed")...

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.