tests for port-resource-request
Change-Id: Ib72b01cb25ccdaa00b2b364dca7f8e485aaaf46b
Depends-On: https://review.openstack.org/590363
Partial-Bug: #1578989
See-Also: https://review.openstack.org/502306 (nova spec)
See-Also: https://review.openstack.org/508149 (neutron spec)
diff --git a/neutron_tempest_plugin/api/admin/test_ports.py b/neutron_tempest_plugin/api/admin/test_ports.py
index cbcd933..642e910 100644
--- a/neutron_tempest_plugin/api/admin/test_ports.py
+++ b/neutron_tempest_plugin/api/admin/test_ports.py
@@ -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 @@
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)
diff --git a/neutron_tempest_plugin/api/base.py b/neutron_tempest_plugin/api/base.py
index 3101af8..03ad085 100644
--- a/neutron_tempest_plugin/api/base.py
+++ b/neutron_tempest_plugin/api/base.py
@@ -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 @@
CONF = config.CONF
+LOG = log.getLogger(__name__)
+
class BaseNetworkTest(test.BaseTestCase):
@@ -674,6 +679,16 @@
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
if 'routes' in router:
@@ -974,6 +989,32 @@
"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):