5450998914
test_qos_policy_assoc_ipv6_subnet_static_ip was failing earlier due to incorrect cleanup order Change-Id: I51d5bb698a6f671bca195ea8c750994b21a8aae6
193 lines
7.9 KiB
Python
193 lines
7.9 KiB
Python
# 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'])
|