From 5450998914ce9ee35bd8ba89a161781ee6117e99 Mon Sep 17 00:00:00 2001 From: Deepthi Kandavara Jayarama Date: Mon, 12 Aug 2019 21:23:28 +0000 Subject: [PATCH] [QOS] Correcting cleanup order of qos api test test_qos_policy_assoc_ipv6_subnet_static_ip was failing earlier due to incorrect cleanup order Change-Id: I51d5bb698a6f671bca195ea8c750994b21a8aae6 --- .../tests/test_ipv6_qos.py | 192 ++++++++++++++++++ 1 file changed, 192 insertions(+) create mode 100644 vmware_nsx_tempest_plugin/tests/test_ipv6_qos.py diff --git a/vmware_nsx_tempest_plugin/tests/test_ipv6_qos.py b/vmware_nsx_tempest_plugin/tests/test_ipv6_qos.py new file mode 100644 index 0000000..0365b27 --- /dev/null +++ b/vmware_nsx_tempest_plugin/tests/test_ipv6_qos.py @@ -0,0 +1,192 @@ +# Copyright 2019 VMware Inc +# All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +from oslo_log import log as logging + +from tempest import config +from tempest.lib.common.utils import data_utils +from tempest.lib import decorators +from tempest.lib import exceptions +from tempest import test + +from vmware_nsx_tempest_plugin.lib import feature_manager + + +CONF = config.CONF +CONF.validation.auth_method = 'None' + +LOG = logging.getLogger(__name__) + +DSCP_MARK = 12 +DSCP_MARK_UPDATED = 16 +BW_VALUE_KBPS = 1024 +BW_VALUE_MBPS = 1 +UPDATED_BW_VALUE_KBPS = 2048 +UPDATED_BW_VALUE_MBPS = 2 +MAX_BURST_KBPS = 1024000 +MAX_BURST_MBPS = 1 + + +class TestQos(feature_manager.FeatureManager): + + @classmethod + def skip_checks(cls): + super(TestQos, cls).skip_checks() + if not (CONF.network.project_networks_reachable or + CONF.network.public_network_id): + msg = ('Either project_networks_reachable must be "true", or ' + 'public_network_id must be defined.') + raise cls.skipException(msg) + if not CONF.network.public_network_cidr: + msg = "public_network_cidr must be defined in network section." + raise cls.skipException(msg) + if not test.is_extension_enabled('qos', 'network'): + msg = "q-qos extension not enabled." + raise cls.skipException(msg) + if not (CONF.network_feature_enabled.ipv6 and + CONF.network_feature_enabled.ipv6_subnet_attributes): + raise cls.skipException('IPv6 or its attributes not supported') + + @classmethod + def setup_clients(cls): + super(TestQos, cls).setup_clients() + cls.cmgr_adm = cls.get_client_manager('admin') + + @classmethod + def resource_setup(cls): + super(TestQos, cls).resource_setup() + + def _create_ipv6_subnet(self, network, router, slaac=False): + subnet_client = self.cmgr_adm.subnets_client + subnet_name = network['name'] + 'ipv6-sub' + address_cidr = CONF.network.project_network_v6_cidr + address_prefixlen = CONF.network.project_network_v6_mask_bits + if ((address_prefixlen >= 126)): + msg = ("Subnet %s isn't large enough for the test" % address_cidr) + raise exceptions.InvalidConfiguration(msg) + allocation_pools = {'allocation_pools': [{ + 'start': str(address_cidr).split('/')[0] + '2', + 'end':str(address_cidr).split('/')[0] + '70'}]} + if slaac: + subnet = self.create_topology_subnet(subnet_name, network, + subnets_client=subnet_client, + routers_client=self.cmgr_adm.routers_client, + router_id=router['id'], + ip_version=6, ipv6_ra_mode='slaac', + ipv6_address_mode='slaac', + **allocation_pools) + else: + subnet = self.create_topology_subnet(subnet_name, network, + subnets_client=subnet_client, + ip_version=6, enable_dhcp=False, + **allocation_pools) + return subnet + + def _create_single_ipv6_rtr_topology(self, slaac=False): + """Create IPv6 subnet and attach them to a router + """ + rtr_name = data_utils.rand_name("qos-rtr") + router = self.create_topology_router( + rtr_name, routers_client=self.cmgr_adm.routers_client) + name = data_utils.rand_name("qos-network") + networks_client = self.cmgr_adm.networks_client + network = self.create_topology_network(name, + networks_client=networks_client) + self._create_ipv6_subnet(network, router, slaac) + return network, router + + def _create_qos_policy(self): + """Create a qos policy with bandwidth limit rule + and dscp rule + """ + name = data_utils.rand_name('test-qos-policy-') + policy = self.create_qos_policy(name, + description='bw_dscp_rule', + shared=False) + # add bw rule + self.create_bandwidth_limit_rule( + policy_id=policy['id'], max_kbps=BW_VALUE_KBPS, + max_burst_kbps=MAX_BURST_KBPS) + # add dscp rule + self.create_dscp_marking_rule( + policy_id=policy['id'], dscp_mark=DSCP_MARK) + return policy + + @decorators.attr(type=['nsxv3', 'positive']) + @decorators.idempotent_id('b46d699f-9bd5-4430-a816-167ffed66551') + def test_qos_policy_assoc_ipv6_subnet_static_ip(self): + """Create network with IPv6 static assignment based subnet + Associate qos policy to the network + """ + policy = self. _create_qos_policy() + network, _ = self._create_single_ipv6_rtr_topology() + self.cmgr_adm.networks_client.update_network( + network['id'], qos_policy_id=policy['id']) + updated_network = self.cmgr_adm.networks_client.show_network( + network['id']) + self.assertEqual(updated_network['network']['qos_policy_id'], + policy['id']) + + @decorators.attr(type=['nsxv3', 'positive']) + @decorators.idempotent_id('a121d38e-d977-4bd0-b7db-6f7d9e206f7a') + def test_qos_policy_assoc_ipv6_subnet_slaac_ip(self): + """Create network with IPv6 SLAAC subnet + Associate qos policy to the network + """ + policy = self. _create_qos_policy() + network, _ = self._create_single_ipv6_rtr_topology(slaac=True) + self.cmgr_adm.networks_client.update_network( + network['id'], qos_policy_id=policy['id']) + updated_network = self.cmgr_adm.networks_client.show_network( + network['id']) + self.assertEqual(updated_network['network']['qos_policy_id'], + policy['id']) + + @decorators.attr(type=['nsxv3', 'positive']) + @decorators.idempotent_id('7a6a2088-612f-4fc6-87be-d58c5d04b946') + def test_qos_policy_assoc_ipv6_port_static_ip(self): + """Create network with IPv6 static assignment based subnet + Associate qos policy to a port attached to the ipv6 + subnet + """ + network, _ = self._create_single_ipv6_rtr_topology() + policy = self. _create_qos_policy() + port_client = self.cmgr_adm.ports_client + body = self.create_topology_port(network=network, + ports_client=port_client, qos_policy_id=policy['id']) + port = body['port'] + body = self.show_topology_port(port['id'], + ports_client=port_client) + port = body['port'] + self.assertEqual(port['qos_policy_id'], + policy['id']) + + @decorators.attr(type=['nsxv3', 'positive']) + @decorators.idempotent_id('f6297a98-f487-4403-852b-61fd0fce329c') + def test_qos_policy_assoc_ipv6_port_slaac_ip(self): + """Create network with IPv6 SLAAC subnet + Associate qos policy to a port attached + to the ipv6 subnet + """ + policy = self. _create_qos_policy() + network, _ = self._create_single_ipv6_rtr_topology(slaac=True) + port_client = self.cmgr_adm.ports_client + self.create_topology_port(network=network, + ports_client=port_client) + self.cmgr_adm.networks_client.update_network( + network['id'], qos_policy_id=policy['id']) + updated_network = self.cmgr_adm.networks_client.show_network( + network['id']) + self.assertEqual(updated_network['network']['qos_policy_id'], + policy['id'])