# 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'])