openstack
/
neutron
Code Issues Proposed changes

Add "default" behaviour to QoS policies 

This patch implements the "default" behaviour for QoS policies.
If this flag is enabled for a QoS policy in a project, all
new networks created will have this QoS policy assigned by default.

If a new QoS policy is created or updated with this flag and another
QoS policy in the same project is set as the default policy, the new
one won't be created or updated. To set another QoS policy as default,
the current one must be unset.

DocImpact: A "default" flag is introduced for QoS policies. If this flag
           is enabled in a QoS policy (attached to a project), then all
           networks created in this project would have this QoS policy
           assigned, unless an explicit policy is specified.
APIImpact

Closes-Bug: #1639220
Change-Id: If5ff2b00fa828f93aa089e275ddbd1ff542b79d4
This commit is contained in:
Rodolfo Alonso Hernandez 2017-02-01 15:15:16 +00:00 committed by Kevin Benton
parent e137e63db3
commit 9d69822e43
23 changed files with 530 additions and 116 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
neutron/common/exceptions.py
View File

@ -31,6 +31,15 @@ class QosRuleNotFound(e.NotFound):
"could not be found.")
class QoSPolicyDefaultAlreadyExists(e.Conflict):
message = _("A default QoS policy exists for project %(project_id)s.")
class QoSPolicyDefaultNotFound(e.Conflict):
message = _("Default QoS policy for project %(project_id)s could not be "
"found.")
class PortQosBindingNotFound(e.NotFound):
message = _("QoS binding for port %(port_id)s and policy %(policy_id)s "
"could not be found.")

6
neutron/core_extensions/base.py
View File

@ -20,6 +20,8 @@ import six
NETWORK = 'network'
PORT = 'port'
EVENT_CREATE = 'create'
EVENT_UPDATE = 'update'
CORE_RESOURCES = [NETWORK, PORT]
@ -29,12 +31,14 @@ CORE_RESOURCES = [NETWORK, PORT]
class CoreResourceExtension(object):
@abc.abstractmethod
def process_fields(self, context, resource_type,
def process_fields(self, context, resource_type, event_type,
requested_resource, actual_resource):
"""Process extension fields.
:param context: neutron api request context
:param resource_type: core resource type (one of CORE_RESOURCES)
:param event_type: kind of event triggering this action (update,
create)
:param requested_resource: resource dict that contains extension fields
:param actual_resource: actual resource dict known to plugin
"""

19
neutron/core_extensions/qos.py
View File

@ -63,6 +63,17 @@ class QosCoreResourceExtension(base.CoreResourceExtension):
policy.attach_port(port['id'])
port[qos_consts.QOS_POLICY_ID] = qos_policy_id
def _create_network_policy(self, context, network, network_changes):
qos_policy_id = network_changes.get(qos_consts.QOS_POLICY_ID)
if not qos_policy_id:
qos_policy_id = policy_object.QosPolicyDefault.get_object(
context, project_id=network['project_id'])
if qos_policy_id is not None:
policy = self._get_policy_obj(context, qos_policy_id)
policy.attach_network(network['id'])
network[qos_consts.QOS_POLICY_ID] = qos_policy_id
def _update_network_policy(self, context, network, network_changes):
old_policy = policy_object.QosPolicy.get_network_policy(
context.elevated(), network['id'])
@ -80,11 +91,13 @@ class QosCoreResourceExtension(base.CoreResourceExtension):
with db_api.autonested_transaction(context.session):
return getattr(self, method_name)(context=context, **kwargs)
def process_fields(self, context, resource_type,
def process_fields(self, context, resource_type, event_type,
requested_resource, actual_resource):
if (qos_consts.QOS_POLICY_ID in requested_resource and
self.plugin_loaded):
self._exec('_update_%s_policy' % resource_type, context,
self.plugin_loaded):
method_name = ('_%(event)s_%(resource)s_policy' %
{'event': event_type, 'resource': resource_type})
self._exec(method_name, context,
{resource_type: actual_resource,
"%s_changes" % resource_type: requested_resource})

2
neutron/db/migration/alembic_migrations/versions/EXPAND_HEAD
View File

@ -1 +1 @@
2b42d90729da
62c781cb6192

44
neutron/db/migration/alembic_migrations/versions/pike/expand/62c781cb6192_add_qos_policies_default_table.py Normal file
View File

@ -0,0 +1,44 @@
# Copyright 2017 Intel Corporation
#
# 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.
#
"""add is default to qos policies
Revision ID: 62c781cb6192
Revises: 2b42d90729da
Create Date: 2017-02-07 13:28:35.894357
"""
# revision identifiers, used by Alembic.
revision = '62c781cb6192'
down_revision = '2b42d90729da'
from alembic import op
import sqlalchemy as sa
def upgrade():
op.create_table(
'qos_policies_default',
sa.Column('qos_policy_id',
sa.String(length=36),
sa.ForeignKey('qos_policies.id', ondelete='CASCADE'),
nullable=False),
sa.Column('project_id',
sa.String(length=255),
nullable=False,
index=True,
primary_key=True),
)

10
neutron/db/qos/models.py
View File

@ -73,6 +73,16 @@ class QosPortPolicyBinding(model_base.BASEV2):
cascade='delete', lazy='joined'))
class QosPolicyDefault(model_base.BASEV2,
model_base.HasProjectPrimaryKeyIndex):
__tablename__ = 'qos_policies_default'
qos_policy_id = sa.Column(sa.String(36),
sa.ForeignKey('qos_policies.id',
ondelete='CASCADE'),
nullable=False)
revises_on_change = ('qos_policy',)
class QosBandwidthLimitRule(model_base.HasId, model_base.BASEV2):
__tablename__ = 'qos_bandwidth_limit_rules'
qos_policy_id = sa.Column(sa.String(36),

4
neutron/extensions/qos.py
View File

@ -32,8 +32,10 @@ from neutron.objects.qos import rule as rule_object
from neutron.plugins.common import constants
from neutron.services.qos import qos_consts
ALIAS = "qos"
QOS_PREFIX = "/qos"
COLLECTION_NAME = 'policies'
# Attribute Map
QOS_RULE_COMMON_FIELDS = {
@ -47,7 +49,7 @@ QOS_RULE_COMMON_FIELDS = {
}
RESOURCE_ATTRIBUTE_MAP = {
'policies': {
COLLECTION_NAME: {
'id': {'allow_post': False, 'allow_put': False,
'validate': {'type:uuid': None},
'is_visible': True, 'primary_key': True},

80
neutron/extensions/qos_default.py Normal file
View File

@ -0,0 +1,80 @@
# Copyright (c) 2017 Intel Corporation.
# 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 neutron_lib.api import converters
from neutron_lib.api import extensions
from neutron.extensions import qos
# The alias of the extension.
ALIAS = 'qos-default'
# The name of the extension.
NAME = 'QoS default policy'
# The description of the extension.
DESCRIPTION = 'Expose the QoS default policy per project'
# A timestamp of when the extension was introduced.
TIMESTAMP = '2017-041-06T10:00:00-00:00'
# The list of required extensions.
REQUIRED_EXTENSIONS = [qos.ALIAS]
# The list of optional extensions.
OPTIONAL_EXTENSIONS = None
# The resource attribute map for the extension.
RESOURCE_ATTRIBUTE_MAP = {
qos.COLLECTION_NAME: {
'is_default': {'allow_post': True,
'allow_put': True,
'default': False,
'convert_to': converters.convert_to_boolean,
'is_visible': True}
}
}
class Qos_default(extensions.ExtensionDescriptor):
@classmethod
def get_name(cls):
return NAME
@classmethod
def get_alias(cls):
return ALIAS
@classmethod
def get_description(cls):
return DESCRIPTION
@classmethod
def get_updated(cls):
return TIMESTAMP
def get_required_extensions(self):
return REQUIRED_EXTENSIONS or []
def get_optional_extensions(self):
return OPTIONAL_EXTENSIONS or []
def get_extended_resources(self, version):
if version == "2.0":
return RESOURCE_ATTRIBUTE_MAP
else:
return {}

93
neutron/objects/qos/policy.py
View File

@ -20,7 +20,6 @@ from oslo_versionedobjects import base as obj_base
from oslo_versionedobjects import exception
from oslo_versionedobjects import fields as obj_fields
from neutron._i18n import _
from neutron.common import constants as n_const
from neutron.common import exceptions
from neutron.db import api as db_api
@ -28,6 +27,7 @@ from neutron.db import models_v2
from neutron.db.qos import api as qos_db_api
from neutron.db.qos import models as qos_db_model
from neutron.db.rbac_db_models import QosPolicyRBAC
from neutron.objects import base as base_db
from neutron.objects import common_types
from neutron.objects.db import api as obj_db_api
from neutron.objects.qos import rule as rule_obj_impl
@ -42,7 +42,8 @@ class QosPolicy(rbac_db.NeutronRbacObject):
# Version 1.3: Added standard attributes (created_at, revision, etc)
# Version 1.4: Changed tenant_id to project_id
# Version 1.5: Direction for bandwidth limit rule added
VERSION = '1.5'
# Version 1.6: Added "is_default" field
VERSION = '1.6'
# required by RbacNeutronMetaclass
rbac_db_model = QosPolicyRBAC
@ -57,32 +58,37 @@ class QosPolicy(rbac_db.NeutronRbacObject):
'name': obj_fields.StringField(),
'shared': obj_fields.BooleanField(default=False),
'rules': obj_fields.ListOfObjectsField('QosRule', subclasses=True),
'is_default': obj_fields.BooleanField(default=False),
}
fields_no_update = ['id', 'project_id']
synthetic_fields = ['rules']
synthetic_fields = ['rules', 'is_default']
extra_filter_names = {'is_default'}
binding_models = {'network': network_binding_model,
'port': port_binding_model}
def obj_load_attr(self, attrname):
if attrname == 'project_id':
return super(QosPolicy, self).obj_load_attr(attrname)
if attrname == 'rules':
return self._reload_rules()
elif attrname == 'is_default':
return self._reload_is_default()
return super(QosPolicy, self).obj_load_attr(attrname)
if attrname != 'rules':
raise exceptions.ObjectActionError(
action='obj_load_attr',
reason=_('unable to load %s') % attrname)
if not hasattr(self, attrname):
self.reload_rules()
def reload_rules(self):
def _reload_rules(self):
rules = rule_obj_impl.get_rules(self.obj_context, self.id)
setattr(self, 'rules', rules)
self.obj_reset_changes(['rules'])
def _reload_is_default(self):
if self.get_default() == self.id:
setattr(self, 'is_default', True)
else:
setattr(self, 'is_default', False)
self.obj_reset_changes(['is_default'])
def get_rule_by_id(self, rule_id):
"""Return rule specified by rule_id.
@ -107,7 +113,8 @@ class QosPolicy(rbac_db.NeutronRbacObject):
not cls.is_accessible(context, policy_obj)):
return
policy_obj.reload_rules()
policy_obj.obj_load_attr('rules')
policy_obj.obj_load_attr('is_default')
return policy_obj
@classmethod
@ -124,7 +131,8 @@ class QosPolicy(rbac_db.NeutronRbacObject):
for obj in objs:
if not cls.is_accessible(context, obj):
continue
obj.reload_rules()
obj.obj_load_attr('rules')
obj.obj_load_attr('is_default')
result.append(obj)
return result
@ -149,7 +157,18 @@ class QosPolicy(rbac_db.NeutronRbacObject):
def create(self):
with db_api.autonested_transaction(self.obj_context.session):
super(QosPolicy, self).create()
self.reload_rules()
if self.is_default:
self.set_default()
self.obj_load_attr('rules')
def update(self):
with db_api.autonested_transaction(self.obj_context.session):
if 'is_default' in self.obj_what_changed():
if self.is_default:
self.set_default()
else:
self.unset_default()
super(QosPolicy, self).update()
def delete(self):
with db_api.autonested_transaction(self.obj_context.session):
@ -184,6 +203,28 @@ class QosPolicy(rbac_db.NeutronRbacObject):
policy_id=self.id,
port_id=port_id)
def set_default(self):
if not self.get_default():
qos_default_policy = QosPolicyDefault(self.obj_context,
qos_policy_id=self.id,
project_id=self.project_id)
qos_default_policy.create()
elif self.get_default() != self.id:
raise exceptions.QoSPolicyDefaultAlreadyExists(
project_id=self.project_id)
def unset_default(self):
if self.get_default() == self.id:
qos_default_policy = QosPolicyDefault.get_object(
self.obj_context, project_id=self.project_id)
qos_default_policy.delete()
def get_default(self):
qos_default_policy = QosPolicyDefault.get_object(
self.obj_context, project_id=self.project_id)
if qos_default_policy:
return qos_default_policy.qos_policy_id
def get_bound_networks(self):
return qos_db_api.get_network_ids_by_network_policy_binding(
self.obj_context, self.id)
@ -264,3 +305,21 @@ class QosPolicy(rbac_db.NeutronRbacObject):
if 'rules' in primitive:
primitive['rules'] = filter_ingress_bandwidth_limit_rules(
primitive['rules'])
if _target_version < (1, 6):
primitive.pop('is_default', None)
@obj_base.VersionedObjectRegistry.register
class QosPolicyDefault(base_db.NeutronDbObject):
# Version 1.0: Initial version
VERSION = '1.0'
db_model = qos_db_model.QosPolicyDefault
fields = {
'qos_policy_id': common_types.UUIDField(),
'project_id': obj_fields.StringField(),
}
primary_keys = ['project_id']

8
neutron/plugins/ml2/extensions/qos.py
View File

@ -32,13 +32,15 @@ class QosExtensionDriver(api.ExtensionDriver):
def process_create_network(self, context, data, result):
self.core_ext_handler.process_fields(
context, base_core.NETWORK, data, result)
context, base_core.NETWORK, base_core.EVENT_CREATE, data, result)
process_update_network = process_create_network
def process_update_network(self, context, data, result):
self.core_ext_handler.process_fields(
context, base_core.NETWORK, base_core.EVENT_UPDATE, data, result)
def process_create_port(self, context, data, result):
self.core_ext_handler.process_fields(
context, base_core.PORT, data, result)
context, base_core.PORT, base_core.EVENT_UPDATE, data, result)
process_update_port = process_create_port

10
neutron/services/qos/qos_plugin.py
View File

@ -38,7 +38,9 @@ class QoSPlugin(qos.QoSPluginBase):
service parameters over ports and networks.
"""
supported_extension_aliases = ['qos', 'qos-bw-limit-direction']
supported_extension_aliases = ['qos',
'qos-bw-limit-direction',
'qos-default']
__native_pagination_support = True
__native_sorting_support = True
@ -299,7 +301,7 @@ class QoSPlugin(qos.QoSPluginBase):
checker.check_bandwidth_rule_conflict(policy, rule_data)
rule = rule_cls(context, qos_policy_id=policy_id, **rule_data)
rule.create()
policy.reload_rules()
policy.obj_load_attr('rules')
self.validate_policy(context, policy)
self.driver_manager.call(qos_consts.UPDATE_POLICY_PRECOMMIT,
context, policy)
@ -338,7 +340,7 @@ class QoSPlugin(qos.QoSPluginBase):
rule = rule_cls(context, id=rule_id)
rule.update_fields(rule_data, reset_changes=True)
rule.update()
policy.reload_rules()
policy.obj_load_attr('rules')
self.validate_policy(context, policy)
self.driver_manager.call(qos_consts.UPDATE_POLICY_PRECOMMIT,
context, policy)
@ -366,7 +368,7 @@ class QoSPlugin(qos.QoSPluginBase):
policy = self._get_policy_obj(context, policy_id)
rule = policy.get_rule_by_id(rule_id)
rule.delete()
policy.reload_rules()
policy.obj_load_attr('rules')
self.driver_manager.call(qos_consts.UPDATE_POLICY_PRECOMMIT,
context, policy)

6
neutron/tests/fullstack/resources/client.py
View File

@ -132,12 +132,14 @@ class ClientFixture(fixtures.Fixture):
router=router_id, body=body)
return router_interface_info
def create_qos_policy(self, tenant_id, name, description, shared):
def create_qos_policy(self, tenant_id, name, description, shared,
is_default):
policy = self.client.create_qos_policy(
body={'policy': {'name': name,
'description': description,
'shared': shared,
'tenant_id': tenant_id}})
'tenant_id': tenant_id,
'is_default': is_default}})
def detach_and_delete_policy():
qos_policy_id = policy['policy']['id']

60
neutron/tests/fullstack/test_qos.py
View File

@ -15,6 +15,7 @@
import functools
from neutron_lib import constants
from neutronclient.common import exceptions
from oslo_utils import uuidutils
from neutron.agent.linux import tc_lib
@ -72,7 +73,7 @@ class BaseQoSRuleTestCase(object):
def _create_qos_policy(self):
return self.safe_client.create_qos_policy(
self.tenant_id, 'fs_policy', 'Fullstack testing policy',
shared='False')
shared='False', is_default='False')
def _prepare_vm_with_qos_policy(self, rule_add_functions):
qos_policy = self._create_qos_policy()
@ -282,3 +283,60 @@ class TestQoSWithL2Population(base.BaseFullStackTestCase):
rule_types = {t['type'] for t in res['rule_types']}
expected_rules = set(ovs_drv.SUPPORTED_RULES)
self.assertEqual(expected_rules, rule_types)
class TestQoSPolicyIsDefault(base.BaseFullStackTestCase):
NAME = 'fs_policy'
DESCRIPTION = 'Fullstack testing policy'
SHARED = True
def setUp(self):
host_desc = [] # No need to register agents for this test case
env_desc = environment.EnvironmentDescription(qos=True)
env = environment.Environment(env_desc, host_desc)
super(TestQoSPolicyIsDefault, self).setUp(env)
def _create_qos_policy(self, project_id, is_default):
return self.safe_client.create_qos_policy(
project_id, self.NAME, self.DESCRIPTION, shared=self.SHARED,
is_default=is_default)
def _update_qos_policy(self, qos_policy_id, is_default):
return self.client.update_qos_policy(
qos_policy_id, body={'policy': {'is_default': is_default}})
def test_create_one_default_qos_policy_per_project(self):
project_ids = [uuidutils.generate_uuid(), uuidutils.generate_uuid()]
for project_id in project_ids:
qos_policy = self._create_qos_policy(project_id, True)
self.assertTrue(qos_policy['is_default'])
self.assertEqual(project_id, qos_policy['project_id'])
qos_policy = self._create_qos_policy(project_id, False)
self.assertFalse(qos_policy['is_default'])
self.assertEqual(project_id, qos_policy['project_id'])
def test_create_two_default_qos_policies_per_project(self):
project_id = uuidutils.generate_uuid()
qos_policy = self._create_qos_policy(project_id, True)
self.assertTrue(qos_policy['is_default'])
self.assertEqual(project_id, qos_policy['project_id'])
self.assertRaises(exceptions.Conflict,
self._create_qos_policy, project_id, True)
def test_update_default_status(self):
project_ids = [uuidutils.generate_uuid(), uuidutils.generate_uuid()]
for project_id in project_ids:
qos_policy = self._create_qos_policy(project_id, True)
self.assertTrue(qos_policy['is_default'])
qos_policy = self._update_qos_policy(qos_policy['id'], False)
self.assertTrue(qos_policy['policy']['is_default'])
def test_update_default_status_conflict(self):
project_id = uuidutils.generate_uuid()
qos_policy_1 = self._create_qos_policy(project_id, True)
self.assertTrue(qos_policy_1['is_default'])
qos_policy_2 = self._create_qos_policy(project_id, False)
self.assertFalse(qos_policy_2['is_default'])
self.assertRaises(exceptions.Conflict,
self._update_qos_policy, qos_policy_2['id'], True)

9
neutron/tests/tempest/api/test_qos.py
View File

@ -82,7 +82,8 @@ class QosTestJSON(base.BaseAdminNetworkTest):
def test_policy_update(self):
policy = self.create_qos_policy(name='test-policy',
description='',
shared=False)
shared=False,
tenant_id=self.admin_client.tenant_id)
self.admin_client.update_qos_policy(policy['id'],
description='test policy desc2',
shared=True)
@ -119,7 +120,8 @@ class QosTestJSON(base.BaseAdminNetworkTest):
def test_shared_policy_update(self):
policy = self.create_qos_policy(name='test-policy',
description='',
shared=True)
shared=True,
tenant_id=self.admin_client.tenant_id)
self.admin_client.update_qos_policy(policy['id'],
description='test policy desc2')
@ -606,7 +608,8 @@ class RbacSharedQosPoliciesTest(base.BaseAdminNetworkTest):
def test_policy_sharing_with_wildcard(self):
qos_pol = self.create_qos_policy(
name=data_utils.rand_name('test-policy'),
description='test-shared-policy', shared=False)
description='test-shared-policy', shared=False,
tenant_id=self.admin_client.tenant_id)
self.assertNotIn(qos_pol, self.client2.list_qos_policies()['policies'])
# test update shared False -> True

85
neutron/tests/unit/core_extensions/test_qos.py
View File

@ -19,6 +19,7 @@ from neutron_lib import context
from neutron.common import exceptions as n_exc
from neutron.core_extensions import base as base_core
from neutron.core_extensions import qos as qos_core
from neutron.objects.qos import policy
from neutron.plugins.common import constants as plugin_constants
from neutron.services.qos import qos_consts
from neutron.tests import base
@ -41,7 +42,7 @@ class QosCoreResourceExtensionTestCase(base.BaseTestCase):
def test_process_fields_no_qos_policy_id(self):
self.core_extension.process_fields(
self.context, base_core.PORT, {}, None)
self.context, base_core.PORT, mock.ANY, {}, None)
self.assertFalse(self.policy_m.called)
def _mock_plugin_loaded(self, plugin_loaded):
@ -54,7 +55,7 @@ class QosCoreResourceExtensionTestCase(base.BaseTestCase):
def test_process_fields_no_qos_plugin_loaded(self):
with self._mock_plugin_loaded(False):
self.core_extension.process_fields(
self.context, base_core.PORT,
self.context, base_core.PORT, mock.ANY,
{qos_consts.QOS_POLICY_ID: None}, None)
self.assertFalse(self.policy_m.called)
@ -66,7 +67,7 @@ class QosCoreResourceExtensionTestCase(base.BaseTestCase):
qos_policy = mock.MagicMock()
self.policy_m.get_object = mock.Mock(return_value=qos_policy)
self.core_extension.process_fields(
self.context, base_core.PORT,
self.context, base_core.PORT, base_core.EVENT_UPDATE,
{qos_consts.QOS_POLICY_ID: qos_policy_id},
actual_port)
@ -85,7 +86,7 @@ class QosCoreResourceExtensionTestCase(base.BaseTestCase):
new_qos_policy = mock.MagicMock()
self.policy_m.get_object = mock.Mock(return_value=new_qos_policy)
self.core_extension.process_fields(
self.context, base_core.PORT,
self.context, base_core.PORT, base_core.EVENT_UPDATE,
{qos_consts.QOS_POLICY_ID: qos_policy2_id},
actual_port)
@ -105,7 +106,7 @@ class QosCoreResourceExtensionTestCase(base.BaseTestCase):
new_qos_policy = mock.MagicMock()
self.policy_m.get_object = mock.Mock(return_value=new_qos_policy)
self.core_extension.process_fields(
self.context, base_core.PORT,
self.context, base_core.PORT, base_core.EVENT_UPDATE,
{qos_consts.QOS_POLICY_ID: None},
actual_port)
@ -125,7 +126,7 @@ class QosCoreResourceExtensionTestCase(base.BaseTestCase):
self.policy_m.get_port_policy = mock.Mock(
return_value=old_qos_policy)
self.core_extension.process_fields(
context, base_core.PORT,
context, base_core.PORT, base_core.EVENT_UPDATE,
{qos_consts.QOS_POLICY_ID: None},
actual_port)
@ -157,7 +158,7 @@ class QosCoreResourceExtensionTestCase(base.BaseTestCase):
shared=False,
policy_tenant_id=self.context.tenant_id)
def test_process_resource_network_updated_no_policy(self):
def test_process_resource_update_network_updated_no_policy(self):
with self._mock_plugin_loaded(True):
network_id = mock.Mock()
qos_policy_id = mock.Mock()
@ -169,14 +170,14 @@ class QosCoreResourceExtensionTestCase(base.BaseTestCase):
new_qos_policy = mock.MagicMock()
self.policy_m.get_object = mock.Mock(return_value=new_qos_policy)
self.core_extension.process_fields(
self.context, base_core.NETWORK,
self.context, base_core.NETWORK, base_core.EVENT_UPDATE,
{qos_consts.QOS_POLICY_ID: None},
actual_network)
old_qos_policy.detach_network.assert_called_once_with(network_id)
self.assertIsNone(actual_network['qos_policy_id'])
def test_process_fields_network_new_policy(self):
def test_process_fields_update_network_new_policy(self):
with self._mock_plugin_loaded(True):
qos_policy_id = mock.Mock()
actual_network = {'id': mock.Mock(),
@ -184,13 +185,13 @@ class QosCoreResourceExtensionTestCase(base.BaseTestCase):
qos_policy = mock.MagicMock()
self.policy_m.get_object = mock.Mock(return_value=qos_policy)
self.core_extension.process_fields(
self.context, base_core.NETWORK,
self.context, base_core.NETWORK, base_core.EVENT_UPDATE,
{qos_consts.QOS_POLICY_ID: qos_policy_id}, actual_network)
qos_policy.attach_network.assert_called_once_with(
actual_network['id'])
def test_process_fields_network_updated_policy(self):
def test_process_fields_update_network_updated_policy(self):
with self._mock_plugin_loaded(True):
qos_policy_id = mock.Mock()
network_id = mock.Mock()
@ -202,7 +203,7 @@ class QosCoreResourceExtensionTestCase(base.BaseTestCase):
new_qos_policy = mock.MagicMock()
self.policy_m.get_object = mock.Mock(return_value=new_qos_policy)
self.core_extension.process_fields(
self.context, base_core.NETWORK,
self.context, base_core.NETWORK, base_core.EVENT_UPDATE,
{qos_consts.QOS_POLICY_ID: qos_policy_id}, actual_network)
old_qos_policy.detach_network.assert_called_once_with(network_id)
@ -220,12 +221,12 @@ class QosCoreResourceExtensionTestCase(base.BaseTestCase):
old_qos_policy.tenant_id = policy_tenant_id
self.policy_m.get_network_policy.return_value = old_qos_policy
self.core_extension.process_fields(
context, base_core.NETWORK,
context, base_core.NETWORK, base_core.EVENT_UPDATE,
{qos_consts.QOS_POLICY_ID: None}, actual_network)
old_qos_policy.detach_network.assert_called_once_with(network_id)
def test_process_fields_network_updated_remove_shared_policy(self):
def test_process_fields_update_network_updated_remove_shared_policy(self):
self._process_network_updated_policy(
context=self.non_admin_context,
shared=True,
@ -237,13 +238,13 @@ class QosCoreResourceExtensionTestCase(base.BaseTestCase):
shared=True,
policy_tenant_id=self.non_admin_context.tenant_id)
def test_process_fields_network_updated_admin_remove_provided_policy(self):
def test_process_fields_update_network_admin_remove_provided_policy(self):
self._process_network_updated_policy(
context=self.context,
shared=True,
policy_tenant_id=self.non_admin_context.tenant_id)
def test_process_fields_network_updated_remove_provided_policy(self):
def test_process_fields_update_network_remove_provided_policy(self):
self.policy_m.is_accessible.return_value = False
self.assertRaises(n_exc.PolicyRemoveAuthorizationError,
self._process_network_updated_policy,
@ -251,6 +252,58 @@ class QosCoreResourceExtensionTestCase(base.BaseTestCase):
shared=False,
policy_tenant_id=self.context.tenant_id)
def test_process_fields_create_network(self):
with self._mock_plugin_loaded(True):
qos_policy_id = mock.Mock()
network_id = mock.Mock()
actual_network = {'id': network_id,
qos_consts.QOS_POLICY_ID: qos_policy_id}
self.policy_m.get_network_policy = mock.Mock(
return_value=qos_policy_id)
qos_policy = mock.MagicMock()
self.policy_m.get_object = mock.Mock(return_value=qos_policy)
self.core_extension.process_fields(
self.context, base_core.NETWORK, base_core.EVENT_CREATE,
actual_network, actual_network)
qos_policy.attach_network.assert_called_once_with(network_id)
def test_process_fields_create_network_no_policy(self):
with self._mock_plugin_loaded(True):
project_id = mock.Mock()
network_id = mock.Mock()
actual_network = {'project_id': project_id,
'id': network_id,
qos_consts.QOS_POLICY_ID: None}
qos_policy_id = mock.Mock()
qos_policy = mock.MagicMock()
with mock.patch.object(policy.QosPolicyDefault, "get_object",
return_value=qos_policy_id) as mock_get_default_policy_id:
self.policy_m.get_object = mock.Mock(return_value=qos_policy)
self.core_extension.process_fields(
self.context, base_core.NETWORK, base_core.EVENT_CREATE,
actual_network, actual_network)
qos_policy.attach_network.assert_called_once_with(network_id)
mock_get_default_policy_id.assert_called_once_with(
self.context, project_id=project_id)
def test_process_fields_create_network_no_default_policy(self):
with self._mock_plugin_loaded(True):
project_id = mock.Mock()
network_id = mock.Mock()
actual_network = {'project_id': project_id,
'id': network_id,
qos_consts.QOS_POLICY_ID: None}
qos_policy = mock.MagicMock()
with mock.patch.object(policy.QosPolicyDefault, "get_object",
return_value=None) as mock_get_default_policy_id:
self.policy_m.get_object = mock.Mock(return_value=qos_policy)
self.core_extension.process_fields(
self.context, base_core.NETWORK, base_core.EVENT_CREATE,
actual_network, actual_network)
qos_policy.attach_network.assert_not_called()
mock_get_default_policy_id.assert_called_once_with(
self.context, project_id=project_id)
def test_extract_fields_plugin_not_loaded(self):
with self._mock_plugin_loaded(False):
fields = self.core_extension.extract_fields(None, None)

93
neutron/tests/unit/objects/qos/test_policy.py
View File

@ -11,6 +11,7 @@
# under the License.
import mock
from oslo_utils import uuidutils
from oslo_versionedobjects import exception
import testtools
@ -39,6 +40,8 @@ class QosPolicyObjectTestCase(test_base.BaseObjectIfaceTestCase):
def setUp(self):
super(QosPolicyObjectTestCase, self).setUp()
mock.patch.object(policy.QosPolicy, 'get_default').start()
# qos_policy_ids will be incorrect, but we don't care in this test
self.db_qos_bandwidth_rules = [
self.get_random_db_fields(rule.QosBandwidthLimitRule)
@ -98,18 +101,16 @@ class QosPolicyObjectTestCase(test_base.BaseObjectIfaceTestCase):
def test_get_object(self):
admin_context = self.context.elevated()
with mock.patch.object(
db_api, 'get_object',
return_value=self.db_objs[0]) as get_object_mock:
with mock.patch.object(self.context,
'elevated',
return_value=admin_context) as context_mock:
obj = self._test_class.get_object(self.context, id='fake_id')
self.assertTrue(self._is_test_class(obj))
self._check_equal(self.objs[0], obj)
context_mock.assert_called_once_with()
get_object_mock.assert_called_once_with(
admin_context, self._test_class.db_model, id='fake_id')
with mock.patch.object(db_api, 'get_object',
return_value=self.db_objs[0]) as get_object_mock, \
mock.patch.object(self.context, 'elevated',
return_value=admin_context) as context_mock:
obj = self._test_class.get_object(self.context, id='fake_id')
self.assertTrue(self._is_test_class(obj))
self._check_equal(self.objs[0], obj)
context_mock.assert_called_once_with()
get_object_mock.assert_called_once_with(
admin_context, self._test_class.db_model, id='fake_id')
def test_to_dict_makes_primitive_field_value(self):
# is_shared_with_tenant requires DB
@ -149,7 +150,7 @@ class QosPolicyDbObjectTestCase(test_base.BaseDbObjectTestCase,
rules.append(rule_obj)
if reload_rules:
policy_obj.reload_rules()
policy_obj.obj_load_attr('rules')
return policy_obj, rules
def test_attach_network_get_network_policy(self):
@ -303,6 +304,42 @@ class QosPolicyDbObjectTestCase(test_base.BaseDbObjectTestCase,
self.assertRaises(n_exc.NetworkQosBindingNotFound,
policy_obj.detach_network, self._network_id)
@mock.patch.object(policy.QosPolicyDefault, 'create')
def test_set_default_no_default_policy_exists(self, mock_default_create):
obj = self._create_test_policy()
with mock.patch.object(obj, 'get_default', return_value=None):
obj.set_default()
mock_default_create.assert_called_once_with()
def test_set_default_default_policy_exists(self):
obj = self._create_test_policy()
with mock.patch.object(obj, 'get_default', return_value=mock.Mock()):
self.assertRaises(n_exc.QoSPolicyDefaultAlreadyExists,
obj.set_default)
def test_set_default_is_default_policy(self):
obj = self._create_test_policy()
with mock.patch.object(obj, 'get_default', return_value=obj.id), \
mock.patch.object(obj, 'set_default'):
obj.set_default()
@mock.patch.object(policy.QosPolicyDefault, 'get_object')
@mock.patch.object(policy.QosPolicyDefault, 'delete')
def test_unset_default_default_policy_exists(self, mock_default_delete,
mock_default_get):
obj = self._create_test_policy()
with mock.patch.object(obj, 'get_default', return_value=obj.id):
mock_default_get.return_value = policy.QosPolicyDefault()
obj.unset_default()
mock_default_get.assert_called_once_with(obj.obj_context,
project_id=obj.project_id)
mock_default_delete.assert_called_once_with()
def test_unset_default_no_default_policy_exists(self):
obj = self._create_test_policy()
with mock.patch.object(obj, 'get_default', return_value=None):
obj.unset_default()
def test_synthetic_rule_fields(self):
policy_obj, rule_obj = self._create_test_policy_with_rules(
[qos_consts.RULE_TYPE_BANDWIDTH_LIMIT])
@ -363,9 +400,16 @@ class QosPolicyDbObjectTestCase(test_base.BaseDbObjectTestCase,
[qos_consts.RULE_TYPE_BANDWIDTH_LIMIT])
self.assertEqual([], policy_obj.rules)
policy_obj.reload_rules()
policy_obj._reload_rules()
self.assertEqual(rule_obj, policy_obj.rules)
def test_reload_is_default(self):
policy_obj = self._create_test_policy()
self.assertFalse(policy_obj.is_default)
policy_obj.set_default()
policy_obj._reload_is_default()
self.assertTrue(policy_obj.is_default)
def test_get_bound_tenant_ids_returns_set_of_tenant_ids(self):
obj = self._create_test_policy()
obj.attach_port(self._port['id'])
@ -475,13 +519,23 @@ class QosPolicyDbObjectTestCase(test_base.BaseDbObjectTestCase,
self.assertIn(rule_objs[1], policy_obj_v1_4.rules)
self.assertIn(rule_objs[2], policy_obj_v1_4.rules)
def test_filter_by_shared(self):
def test_v1_6_to_v1_5_drops_is_default(self):
policy_new = self._create_test_policy()
policy_v1_5 = policy_new.obj_to_primitive(target_version='1.5')
self.assertNotIn('is_default', policy_v1_5['versioned_object.data'])
@mock.patch.object(policy.QosPolicy, 'unset_default')
def test_filter_by_shared(self, *mocks):
project_id = uuidutils.generate_uuid()
policy_obj = policy.QosPolicy(
self.context, name='shared-policy', shared=True)
self.context, name='shared-policy', shared=True,
project_id=project_id, is_default=False)
policy_obj.create()
policy_obj = policy.QosPolicy(
self.context, name='private-policy', shared=False)
self.context, name='private-policy', shared=False,
project_id=project_id)
policy_obj.create()
shared_policies = policy.QosPolicy.get_objects(
@ -499,3 +553,8 @@ class QosPolicyDbObjectTestCase(test_base.BaseDbObjectTestCase,
# QoSPolicy currently cannot be loaded using constant queries number.
# It can be reworked in follow-up patch.
pass
class QosPolicyDefaultObjectTestCase(test_base.BaseObjectIfaceTestCase):
_test_class = policy.QosPolicyDefault

10
neutron/tests/unit/objects/qos/test_rule.py
View File

@ -12,6 +12,7 @@
from neutron_lib import constants
from oslo_utils import uuidutils
from oslo_versionedobjects import exception
from neutron.common import constants as n_const
@ -150,7 +151,8 @@ class QosBandwidthLimitRuleDbObjectTestCase(test_base.BaseDbObjectTestCase,
for obj in self.db_objs:
generated_qos_policy_id = obj['qos_policy_id']
policy_obj = policy.QosPolicy(self.context,
id=generated_qos_policy_id)
id=generated_qos_policy_id,
project_id=uuidutils.generate_uuid())
policy_obj.create()
@ -176,7 +178,8 @@ class QosDscpMarkingRuleDbObjectTestCase(test_base.BaseDbObjectTestCase,
for obj in self.db_objs:
generated_qos_policy_id = obj['qos_policy_id']
policy_obj = policy.QosPolicy(self.context,
id=generated_qos_policy_id)
id=generated_qos_policy_id,
project_id=uuidutils.generate_uuid())
policy_obj.create()
@ -203,5 +206,6 @@ class QosMinimumBandwidthRuleDbObjectTestCase(test_base.BaseDbObjectTestCase,
for obj in self.db_objs:
generated_qos_policy_id = obj['qos_policy_id']
policy_obj = policy.QosPolicy(self.context,
id=generated_qos_policy_id)
id=generated_qos_policy_id,
project_id=uuidutils.generate_uuid())
policy_obj.create()

8
neutron/tests/unit/objects/test_network.py
View File

@ -10,6 +10,8 @@
# License for the specific language governing permissions and limitations
# under the License.
import mock
from neutron.objects import base as obj_base
from neutron.objects import network
from neutron.objects.qos import policy
@ -90,7 +92,8 @@ class NetworkDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
testlib_api.SqlTestCase):
_test_class = network.Network
def test_qos_policy_id(self):
@mock.patch.object(policy.QosPolicy, 'unset_default')
def test_qos_policy_id(self, *mocks):
policy_obj = policy.QosPolicy(self.context)
policy_obj.create()
@ -116,7 +119,8 @@ class NetworkDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
obj = network.Network.get_object(self.context, id=obj.id)
self.assertIsNone(obj.qos_policy_id)
def test__attach_qos_policy(self):
@mock.patch.object(policy.QosPolicy, 'unset_default')
def test__attach_qos_policy(self, *mocks):
obj = self._make_object(self.obj_fields[0])
obj.create()

3
neutron/tests/unit/objects/test_objects.py
View File

@ -66,7 +66,8 @@ object_data = {
'QosDscpMarkingRule': '1.3-0313c6554b34fd10c753cb63d638256c',
'QosMinimumBandwidthRule': '1.3-314c3419f4799067cc31cc319080adff',
'QosRuleType': '1.2-e6fd08fcca152c339cbd5e9b94b1b8e7',
'QosPolicy': '1.5-50460f619c34428ec5651916e938e5a0',
'QosPolicy': '1.6-4adb0cde3102c10d8970ec9487fd7fe7',
'QosPolicyDefault': '1.0-59e5060eedb1f06dd0935a244d27d11c',
'Quota': '1.0-6bb6a0f1bd5d66a2134ffa1a61873097',
'QuotaUsage': '1.0-6fbf820368681aac7c5d664662605cf9',
'Reservation': '1.0-49929fef8e82051660342eed51b48f2a',

7
neutron/tests/unit/objects/test_ports.py
View File

@ -10,6 +10,7 @@
# License for the specific language governing permissions and limitations
# under the License.
import mock
from oslo_utils import uuidutils
import testscenarios
@ -262,7 +263,8 @@ class PortDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
obj = ports.Port.get_object(self.context, id=obj.id)
self.assertIn(sg2_id, obj.security_group_ids)
def test_qos_policy_id(self):
@mock.patch.object(policy.QosPolicy, 'unset_default')
def test_qos_policy_id(self, *mocks):
policy_obj = policy.QosPolicy(self.context)
policy_obj.create()
@ -288,7 +290,8 @@ class PortDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
obj = ports.Port.get_object(self.context, id=obj.id)
self.assertIsNone(obj.qos_policy_id)
def test__attach_qos_policy(self):
@mock.patch.object(policy.QosPolicy, 'unset_default')
def test__attach_qos_policy(self, *mocks):
obj = self._make_object(self.obj_fields[0])
obj.create()

68
neutron/tests/unit/services/qos/test_qos_plugin.py
View File

@ -9,6 +9,7 @@
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import mock
from neutron_lib import context
from neutron_lib.plugins import directory
@ -39,13 +40,16 @@ class TestQosPlugin(base.BaseQosTestCase):
mock.patch('neutron.objects.db.api.update_object').start()
mock.patch('neutron.objects.db.api.delete_object').start()
mock.patch('neutron.objects.db.api.get_object').start()
mock.patch(
'neutron.objects.qos.policy.QosPolicy.obj_load_attr').start()
_mock_qos_load_attr = mock.patch(
'neutron.objects.qos.policy.QosPolicy.obj_load_attr')
self.mock_qos_load_attr = _mock_qos_load_attr.start()
# We don't use real models as per mocks above. We also need to mock-out
# methods that work with real data types
mock.patch(
'neutron.objects.base.NeutronDbObject.modify_fields_from_db'
).start()
mock.patch.object(policy_object.QosPolicy, 'unset_default').start()
mock.patch.object(policy_object.QosPolicy, 'set_default').start()
cfg.CONF.set_override("core_plugin", DB_PLUGIN_KLASS)
cfg.CONF.set_override("service_plugins", ["qos"])
@ -67,7 +71,8 @@ class TestQosPlugin(base.BaseQosTestCase):
'project_id': uuidutils.generate_uuid(),
'name': 'test-policy',
'description': 'Test policy description',
'shared': True}}
'shared': True,
'is_default': False}}
self.rule_data = {
'bandwidth_limit_rule': {'id': uuidutils.generate_uuid(),
@ -339,13 +344,15 @@ class TestQosPlugin(base.BaseQosTestCase):
'tenant_id': project_id,
'name': 'test-policy',
'description': 'Test policy description',
'shared': True}}
'shared': True,
'is_default': False}}
policy_details = {'id': policy_id,
'project_id': project_id,
'name': 'test-policy',
'description': 'Test policy description',
'shared': True}
'shared': True,
'is_default': False}
with mock.patch('neutron.objects.qos.policy.QosPolicy') as QosMocked:
self.qos_plugin.create_policy(self.ctxt, tenant_policy)
@ -412,27 +419,23 @@ class TestQosPlugin(base.BaseQosTestCase):
def test_create_policy_rule_check_rule_min_less_than_max(self):
_policy = self._get_policy()
setattr(_policy, "rules", [self.rule])
with mock.patch('neutron.objects.qos.rule.get_rules',
return_value=[self.rule]) as mock_get_rules, \
mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
return_value=_policy) as mock_qos_get_obj:
with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
return_value=_policy) as mock_qos_get_obj:
self.qos_plugin.create_policy_minimum_bandwidth_rule(
self.ctxt, _policy.id, self.rule_data)
self._validate_driver_params('update_policy')
mock_get_rules.assert_called_once_with(self.ctxt, _policy.id)
self.mock_qos_load_attr.assert_called_once_with('rules')
mock_qos_get_obj.assert_called_once_with(self.ctxt, id=_policy.id)
def test_create_policy_rule_check_rule_max_more_than_min(self):
_policy = self._get_policy()
setattr(_policy, "rules", [self.min_rule])
with mock.patch('neutron.objects.qos.rule.get_rules',
return_value=[self.rule]) as mock_get_rules, \
mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
return_value=_policy) as mock_qos_get_obj:
with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
return_value=_policy) as mock_qos_get_obj:
self.qos_plugin.create_policy_bandwidth_limit_rule(
self.ctxt, _policy.id, self.rule_data)
self._validate_driver_params('update_policy')
mock_get_rules.assert_called_once_with(self.ctxt, _policy.id)
self.mock_qos_load_attr.assert_called_once_with('rules')
mock_qos_get_obj.assert_called_once_with(self.ctxt, id=_policy.id)
def test_create_policy_rule_check_rule_bwlimit_less_than_minbw(self):
@ -486,39 +489,32 @@ class TestQosPlugin(base.BaseQosTestCase):
def test_update_policy_rule_check_rule_min_less_than_max(self):
_policy = self._get_policy()
setattr(_policy, "rules", [self.rule])
with mock.patch('neutron.objects.qos.rule.get_rules',
return_value=[self.rule]) as mock_get_rules, \
mock.patch(
'neutron.objects.qos.policy.QosPolicy.get_object',
return_value=_policy):
with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
return_value=_policy):
self.qos_plugin.update_policy_bandwidth_limit_rule(
self.ctxt, self.rule.id, self.policy.id, self.rule_data)
mock_get_rules.assert_called_once_with(self.ctxt, _policy.id)
self.mock_qos_load_attr.assert_called_once_with('rules')
self._validate_driver_params('update_policy')
rules = [self.rule, self.min_rule]
setattr(_policy, "rules", rules)
with mock.patch('neutron.objects.qos.rule.get_rules',
return_value=rules) as mock_get_rules, \
mock.patch(
'neutron.objects.qos.policy.QosPolicy.get_object',
return_value=_policy):
self.mock_qos_load_attr.reset_mock()
with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
return_value=_policy):
self.qos_plugin.update_policy_minimum_bandwidth_rule(
self.ctxt, self.min_rule.id,
self.policy.id, self.rule_data)
mock_get_rules.assert_called_once_with(self.ctxt, _policy.id)
self.mock_qos_load_attr.assert_called_once_with('rules')
self._validate_driver_params('update_policy')
def test_update_policy_rule_check_rule_bwlimit_less_than_minbw(self):
_policy = self._get_policy()
setattr(_policy, "rules", [self.rule])
with mock.patch('neutron.objects.qos.rule.get_rules',
return_value=[self.rule]), mock.patch(
'neutron.objects.qos.policy.QosPolicy.get_object',
return_value=_policy):
with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
return_value=_policy):
self.qos_plugin.update_policy_bandwidth_limit_rule(
self.ctxt, self.rule.id, self.policy.id, self.rule_data)
self.assertTrue(getattr(rule_object, "get_rules").called)
self.mock_qos_load_attr.assert_called_once_with('rules')
self._validate_driver_params('update_policy')
self.rule_data['minimum_bandwidth_rule']['min_kbps'] = 1000
with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
@ -532,13 +528,11 @@ class TestQosPlugin(base.BaseQosTestCase):
def test_update_policy_rule_check_rule_minbw_gr_than_bwlimit(self):
_policy = self._get_policy()
setattr(_policy, "rules", [self.min_rule])
with mock.patch('neutron.objects.qos.rule.get_rules',
return_value=[self.min_rule]), mock.patch(
'neutron.objects.qos.policy.QosPolicy.get_object',
return_value=_policy):
with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
return_value=_policy):
self.qos_plugin.update_policy_minimum_bandwidth_rule(
self.ctxt, self.min_rule.id, self.policy.id, self.rule_data)
self.assertTrue(getattr(rule_object, "get_rules").called)
self.mock_qos_load_attr.assert_called_once_with('rules')
self._validate_driver_params('update_policy')
self.rule_data['bandwidth_limit_rule']['max_kbps'] = 1
with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',

6
neutron/tests/unit/services/revisions/test_revision_plugin.py
View File

@ -170,7 +170,8 @@ class TestRevisionPlugin(test_plugin.Ml2PluginV2TestCase):
with self.port() as port:
rev = port['port']['revision_number']
qos_plugin = directory.get_plugin('QOS')
qos_policy = {'policy': {'name': "policy1",
qos_policy = {'policy': {'id': uuidutils.generate_uuid(),
'name': "policy1",
'project_id': uuidutils.generate_uuid()}}
qos_obj = qos_plugin.create_policy(self.ctx, qos_policy)
data = {'port': {'qos_policy_id': qos_obj['id']}}
@ -182,7 +183,8 @@ class TestRevisionPlugin(test_plugin.Ml2PluginV2TestCase):
with self.network() as network:
rev = network['network']['revision_number']
qos_plugin = directory.get_plugin('QOS')
qos_policy = {'policy': {'name': "policy1",
qos_policy = {'policy': {'id': uuidutils.generate_uuid(),
'name': "policy1",
'project_id': uuidutils.generate_uuid()}}
qos_obj = qos_plugin.create_policy(self.ctx, qos_policy)
data = {'network': {'qos_policy_id': qos_obj['id']}}

6
releasenotes/notes/add_is_default_to_qos_policies-f7c6bbac08d474d5.yaml Normal file
View File

@ -0,0 +1,6 @@
---
prelude: >
Add 'default' behaviour to QoS policies
features:
- Neutron now supports having a default QoS policy in a project, assigned
automatically to all new networks created.