openstack/neutron-tempest-plugin
Code Issues Proposed changes

Merge "tests for port-resource-request"

Browse Source
This commit is contained in:
Zuul 2019-01-10 09:30:30 +00:00 committed by Gerrit Code Review
commit 201903d8b8
4 changed files with 176 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.zuul.yaml
View File

@ -46,6 +46,7 @@
- network-ip-availability
- network_availability_zone
- pagination
- port-resource-request
- port-mac-address-regenerate
- port-security
- port-security-groups-filtering
@ -127,6 +128,7 @@
agent_availability_zone: nova
image_is_advanced: true
available_type_drivers: flat,geneve,vlan,gre,local,vxlan
provider_net_base_segm_id: 1
irrelevant-files: &tempest-irrelevant-files
- ^(test-|)requirements.txt$
- ^releasenotes/.*$

129
neutron_tempest_plugin/api/admin/test_ports.py
View File

@ -14,11 +14,17 @@
# under the License.
import netaddr
import six
from neutron_lib import constants as const
from tempest.common import utils
from tempest.lib.common.utils import data_utils
from tempest.lib import decorators
from neutron_tempest_plugin.api import base
from neutron_tempest_plugin import config
CONF = config.CONF
class PortTestCasesAdmin(base.BaseAdminNetworkTest):
@ -58,3 +64,126 @@ class PortTestCasesAdmin(base.BaseAdminNetworkTest):
new_mac = body['port']['mac_address']
self.assertNotEqual(current_mac, new_mac)
self.assertTrue(netaddr.valid_mac(new_mac))
class PortTestCasesResourceRequest(base.BaseAdminNetworkTest):
required_extensions = ['port-resource-request',
'qos',
'qos-bw-minimum-ingress']
EGRESS_KBPS = 1000
INGRESS_KBPS = 2000
@classmethod
def skip_checks(cls):
super(PortTestCasesResourceRequest, cls).skip_checks()
if not config.CONF.neutron_plugin_options.provider_vlans:
msg = "Skipped as provider VLANs are not available in config"
raise cls.skipException(msg)
@classmethod
def resource_setup(cls):
super(PortTestCasesResourceRequest, cls).resource_setup()
cls.vnic_type = 'normal'
# Note(lajoskatona): to avoid creating provider network use vxlan
# as provider network type:
cls.network = cls.create_network(provider_network_type='vxlan')
cls.physnet_name = CONF.neutron_plugin_options.provider_vlans[0]
base_segm = CONF.neutron_plugin_options.provider_net_base_segm_id
cls.prov_network = cls.create_provider_network(
physnet_name=cls.physnet_name, start_segmentation_id=base_segm)
def _create_qos_policy_and_port(self, network, vnic_type,
network_policy=False):
qos_policy = self.create_qos_policy(
name=data_utils.rand_name('test_policy'), shared=True)
self.create_qos_minimum_bandwidth_rule(qos_policy['id'],
self.EGRESS_KBPS,
const.EGRESS_DIRECTION)
self.create_qos_minimum_bandwidth_rule(qos_policy['id'],
self.INGRESS_KBPS,
const.INGRESS_DIRECTION)
port_policy_id = qos_policy['id'] if not network_policy else None
port_kwargs = {
'qos_policy_id': port_policy_id,
'binding:vnic_type': vnic_type
}
if network_policy:
self.admin_client.update_network(network['id'],
qos_policy_id=qos_policy['id'])
port_id = self.create_port(network, **port_kwargs)['id']
return self.admin_client.show_port(port_id)['port']
def _assert_resource_request(self, port, vnic_type):
self.assertIn('resource_request', port)
vnic_trait = 'CUSTOM_VNIC_TYPE_%s' % vnic_type.upper()
physnet_trait = 'CUSTOM_PHYSNET_%s' % self.physnet_name.upper()
six.assertCountEqual(self, [physnet_trait, vnic_trait],
port['resource_request']['required'])
self.assertEqual(
{'NET_BW_EGR_KILOBIT_PER_SEC': self.EGRESS_KBPS,
'NET_BW_IGR_KILOBIT_PER_SEC': self.INGRESS_KBPS},
port['resource_request']['resources']
)
@decorators.idempotent_id('ebb86dc4-716c-4558-8516-6dfc4a67601f')
def test_port_resource_request(self):
port = self._create_qos_policy_and_port(
network=self.prov_network, vnic_type=self.vnic_type)
port_id = port['id']
self._assert_resource_request(port, self.vnic_type)
# Note(lajoskatona): port-resource-request is an admin only feature,
# so test if non-admin user can't see the new field.
port = self.client.show_port(port_id)['port']
self.assertNotIn('resource_request', port)
self.update_port(port, **{'qos_policy_id': None})
port = self.admin_client.show_port(port_id)['port']
self.assertIsNone(port['resource_request'])
@decorators.idempotent_id('10b3308b-d8a2-459b-9b89-a146863c357f')
def test_port_resource_request_no_provider_net(self):
port = self._create_qos_policy_and_port(
network=self.network, vnic_type=self.vnic_type)
self.assertIn('resource_request', port)
self.assertIsNone(port['resource_request'])
@decorators.idempotent_id('0eeb6ffa-9a7a-40b5-83dd-dbdcd67e2e64')
def test_port_resource_request_empty(self):
qos_policy = self.create_qos_policy(
name=data_utils.rand_name('test_policy'), shared=True)
# Note(lajoskatona): Add a non-minimum-bandwidth-rule to the policy
# to make sure that the resource request is not filled with it.
self.create_qos_bandwidth_limit_rule(qos_policy['id'],
self.EGRESS_KBPS, 800,
const.EGRESS_DIRECTION)
port_kwargs = {
'qos_policy_id': qos_policy['id'],
'binding:vnic_type': self.vnic_type
}
port_id = self.create_port(self.prov_network, **port_kwargs)['id']
port = self.admin_client.show_port(port_id)['port']
self.assertIn('resource_request', port)
self.assertIsNone(port['resource_request'])
@decorators.idempotent_id('b6c34ae4-44c8-47f0-86de-7ef9866fa000')
def test_port_resource_request_inherited_policy(self):
port = self._create_qos_policy_and_port(
network=self.prov_network, vnic_type=self.vnic_type,
network_policy=True)
self._assert_resource_request(port, self.vnic_type)

41
neutron_tempest_plugin/api/base.py
View File

@ -14,10 +14,13 @@
# under the License.
import functools
import itertools
import math
import time
import netaddr
from neutron_lib import constants as const
from oslo_log import log
from tempest.common import utils as tutils
from tempest.lib.common.utils import data_utils
from tempest.lib import exceptions as lib_exc
@ -31,6 +34,8 @@ from neutron_tempest_plugin import exceptions
CONF = config.CONF
LOG = log.getLogger(__name__)
class BaseNetworkTest(test.BaseTestCase):
@ -673,6 +678,16 @@ class BaseNetworkTest(test.BaseTestCase):
cls.qos_rules.append(qos_rule)
return qos_rule
@classmethod
def create_qos_minimum_bandwidth_rule(cls, policy_id, min_kbps,
direction=const.EGRESS_DIRECTION):
"""Wrapper utility that creates and returns a QoS min bw rule."""
body = cls.admin_client.create_minimum_bandwidth_rule(
policy_id, direction, min_kbps)
qos_rule = body['minimum_bandwidth_rule']
cls.qos_rules.append(qos_rule)
return qos_rule
@classmethod
def delete_router(cls, router, client=None):
client = client or cls.client
@ -974,6 +989,32 @@ class BaseAdminNetworkTest(BaseNetworkTest):
"net(%s) has no usable IP address in allocation pools" % net_id)
raise exceptions.InvalidConfiguration(message)
@classmethod
def create_provider_network(cls, physnet_name, start_segmentation_id,
max_attempts=30):
segmentation_id = start_segmentation_id
for attempts in itertools.count():
try:
prov_network = cls.create_network(
name=data_utils.rand_name('test_net'),
shared=True,
provider_network_type='vlan',
provider_physical_network=physnet_name,
provider_segmentation_id=segmentation_id)
break
except lib_exc.Conflict:
if attempts > max_attempts:
LOG.exception("Failed to create provider network after "
"%d attempts", attempts)
raise lib_exc.TimeoutException
segmentation_id += 1
if segmentation_id > 4095:
raise lib_exc.TempestException(
"No free segmentation id was found for provider "
"network creation!")
time.sleep(CONF.network.build_interval)
return prov_network
def require_qos_rule_type(rule_type):
def decorator(f):

4
neutron_tempest_plugin/config.py
View File

@ -21,6 +21,10 @@ NeutronPluginOptions = [
cfg.ListOpt('provider_vlans',
default=[],
help='List of provider networks available in the deployment.'),
cfg.IntOpt('provider_net_base_segm_id',
default=3000,
help='Base segmentation ID to create provider networks. '
'This value will be increased in case of conflict.'),
cfg.BoolOpt('specify_floating_ip_address_available',
default=True,
help='Allow passing an IP Address of the floating ip when '