Add "qos_network_policy_id" to port definition

Added "qos_network_policy_id" key to port dictionary.

Change-Id: I75713e0e3924ad4c9177e7c9b04c58882e292dc0
Closes-Bug: #1851362
changes/44/693244/4
Rodolfo Alonso Hernandez 3 years ago
parent 6c410b166a
commit f2b52b5e8f
  1. 8
      neutron/core_extensions/qos.py
  2. 20
      neutron/extensions/qos_port_network_policy.py
  3. 4
      neutron/services/qos/qos_plugin.py
  4. 38
      neutron/tests/unit/core_extensions/test_qos.py

@ -107,4 +107,10 @@ class QosCoreResourceExtension(base.CoreResourceExtension):
binding = resource['qos_policy_binding']
qos_policy_id = binding['policy_id'] if binding else None
return {qos_consts.QOS_POLICY_ID: qos_policy_id}
retval = {qos_consts.QOS_POLICY_ID: qos_policy_id}
if resource_type == base.PORT:
network_binding = resource.get('qos_network_policy_binding')
qos_net_policy_id = (network_binding['policy_id'] if
network_binding else None)
retval[qos_consts.QOS_NETWORK_POLICY_ID] = qos_net_policy_id
return retval

@ -0,0 +1,20 @@
# Copyright (c) 2020 Red Hat, Inc.
#
# 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 neutron_lib.api.definitions import qos_port_network_policy
from neutron_lib.api import extensions as api_extensions
class Qos_port_network_policy(api_extensions.APIExtensionDescriptor):
api_definition = qos_port_network_policy

@ -20,6 +20,7 @@ from neutron_lib.api.definitions import qos as qos_apidef
from neutron_lib.api.definitions import qos_bw_limit_direction
from neutron_lib.api.definitions import qos_bw_minimum_ingress
from neutron_lib.api.definitions import qos_default
from neutron_lib.api.definitions import qos_port_network_policy
from neutron_lib.api.definitions import qos_rule_type_details
from neutron_lib.api.definitions import qos_rules_alias
from neutron_lib.callbacks import events as callbacks_events
@ -61,7 +62,8 @@ class QoSPlugin(qos.QoSPluginBase):
qos_rule_type_details.ALIAS,
port_resource_request.ALIAS,
qos_bw_minimum_ingress.ALIAS,
qos_rules_alias.ALIAS]
qos_rules_alias.ALIAS,
qos_port_network_policy.ALIAS]
__native_pagination_support = True
__native_sorting_support = True

@ -13,6 +13,8 @@
# License for the specific language governing permissions and limitations
# under the License.
import itertools
import mock
from neutron_lib import context
from neutron_lib.exceptions import qos as qos_exc
@ -26,9 +28,14 @@ from neutron.objects.qos import policy
from neutron.tests import base
def _get_test_dbdata(qos_policy_id):
return {'id': None, 'qos_policy_binding': {'policy_id': qos_policy_id,
'network_id': 'fake_net_id'}}
def _get_test_dbdata(qos_policy_id, qos_network_policy_id=None):
retval = {'id': None,
'qos_policy_binding': {'policy_id': qos_policy_id,
'network_id': 'fake_net_id'}}
if qos_network_policy_id:
retval['qos_network_policy_binding'] = {
'policy_id': qos_network_policy_id}
return retval
class QosCoreResourceExtensionTestCase(base.BaseTestCase):
@ -313,18 +320,23 @@ class QosCoreResourceExtensionTestCase(base.BaseTestCase):
fields = self.core_extension.extract_fields(None, None)
self.assertEqual({}, fields)
def _test_extract_fields_for_port(self, qos_policy_id):
def _test_extract_fields_for_port(self, qos_policy_id,
qos_network_policy_id=None):
with self._mock_plugin_loaded(True):
fields = self.core_extension.extract_fields(
base_core.PORT, _get_test_dbdata(qos_policy_id))
self.assertEqual({qos_consts.QOS_POLICY_ID: qos_policy_id}, fields)
def test_extract_fields_no_port_policy(self):
self._test_extract_fields_for_port(None)
def test_extract_fields_port_policy_exists(self):
qos_policy_id = mock.Mock()
self._test_extract_fields_for_port(qos_policy_id)
base_core.PORT, _get_test_dbdata(qos_policy_id,
qos_network_policy_id))
expected = {
qos_consts.QOS_POLICY_ID: qos_policy_id,
qos_consts.QOS_NETWORK_POLICY_ID: qos_network_policy_id}
self.assertEqual(expected, fields)
def test_extract_fields_for_port(self):
port_qos_policies = [None, uuidutils.generate_uuid()]
network_qos_policies = [None, uuidutils.generate_uuid()]
for port_qos, net_qos in itertools.product(port_qos_policies,
network_qos_policies):
self._test_extract_fields_for_port(port_qos, net_qos)
def _test_extract_fields_for_network(self, qos_policy_id):
with self._mock_plugin_loaded(True):

Loading…
Cancel
Save