openstack/neutron
Code Issues Proposed changes

Merge "Refactor QoS plugin to avoid code duplication"

Browse Source
This commit is contained in:
Jenkins 2016-05-26 17:45:22 +00:00 committed by Gerrit Code Review
commit cd4a80461e
7 changed files with 400 additions and 214 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
doc/source/devref/neutron_api.rst
View File

@ -52,6 +52,13 @@ repositories under the neutron tent. Below you can find a list of known
incompatible changes that could or are known to trigger those breakages.
The changes are listed in reverse chronological order (newer at the top).
* change: QoS plugin refactor
- commit: I863f063a0cfbb464cedd00bddc15dd853cbb6389
- solution: implement the new abstract methods in
neutron.extensions.qos.QoSPluginBase.
- severity: Low (some out-of-tree plugins might be affected).
* change: Consume ConfigurableMiddleware from oslo_middleware.
- commit: If7360608f94625b7d0972267b763f3e7d7624fee

17
doc/source/devref/quality_of_service.rst
View File

@ -76,6 +76,23 @@ Service side design
integrated into other plugins with ease.
QoS plugin implementation guide
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
The neutron.extensions.qos.QoSPluginBase class uses method proxies for methods
relating to QoS policy rules. Each of these such methods is generic in the sense
that it is intended to handle any rule type. For example, QoSPluginBase has a
create_policy_rule method instead of both create_policy_dscp_marking_rule and
create_policy_bandwidth_limit_rule methods. The logic behind the proxies allows
a call to a plugin's create_policy_dscp_marking_rule to be handled by the
create_policy_rule method, which will receive a QosDscpMarkingRule object as an
argument in order to execute behavior specific to the DSCP marking rule type.
This approach allows new rule types to be introduced without requiring a plugin
to modify code as a result. As would be expected, any subclass of QoSPluginBase
must override the base class's abc.abstractmethod methods, even if to raise
NotImplemented.
Supported QoS rule types
~~~~~~~~~~~~~~~~~~~~~~~~

147
neutron/extensions/qos.py
View File

@ -15,6 +15,7 @@
import abc
import itertools
import re
from neutron_lib.api import converters
import six
@ -25,6 +26,7 @@ from neutron.api.v2 import base
from neutron.api.v2 import resource_helper
from neutron.common import constants as common_constants
from neutron import manager
from neutron.objects.qos import rule as rule_object
from neutron.plugins.common import constants
from neutron.services.qos import qos_consts
from neutron.services import service_base
@ -188,6 +190,93 @@ class QoSPluginBase(service_base.ServicePluginBase):
path_prefix = QOS_PREFIX
# The rule object type to use for each incoming rule-related request.
rule_objects = {'bandwidth_limit': rule_object.QosBandwidthLimitRule,
'dscp_marking': rule_object.QosDscpMarkingRule}
# Patterns used to call method proxies for all policy-rule-specific
# method calls (see __getattr__ docstring, below).
qos_rule_method_patterns = [
re.compile(
r"^((create|update|delete)_policy_(?P<rule_type>.*)_rule)$"),
re.compile(
r"^(get_policy_(?P<rule_type>.*)_(rules|rule))$"),
]
def __getattr__(self, attrib):
"""Implement method proxies for all policy-rule-specific requests. For
a given request type (such as to update a rule), a single method will
handle requests for all rule types. For example, the
update_policy_rule method will handle requests for both
update_policy_dscp_marking_rule and update_policy_bandwidth_limit_rule.
:param attrib: the requested method; in the normal case, this will be,
for example, "update_policy_dscp_marking_rule"
:type attrib: str
"""
# Find and call the proxy method that implements the requested one.
for pattern in self.qos_rule_method_patterns:
res = re.match(pattern, attrib)
if res:
rule_type = res.group('rule_type')
if rule_type in self.rule_objects:
# Remove the rule_type value (plus underscore) from attrib
# in order to get the proxy method name. So, for instance,
# from "delete_policy_dscp_marking_rule" we'll get
# "delete_policy_rule".
proxy_method = attrib.replace(rule_type + '_', '')
rule_obj = self.rule_objects[rule_type]
return self._call_proxy_method(proxy_method, rule_obj)
# If we got here, then either attrib matched no pattern or the
# rule_type embedded in attrib wasn't in self.rule_objects.
raise AttributeError(attrib)
def _call_proxy_method(self, method_name, rule_obj):
"""Call proxy method. We need to add the rule_obj, obtained from the
self.rule_objects dictionary, to the incoming args. The context is
passed to proxy method as first argument; the remaining args will
follow rule_obj.
Some of the incoming method calls have the policy rule name as one of
the keys in the kwargs. For instance, the incoming kwargs for the
create_policy_bandwidth_limit_rule take this form:
{ 'bandwidth_limit_rule': {
u'bandwidth_limit_rule':
{ 'max_burst_kbps': 0,
u'max_kbps': u'100',
'tenant_id': u'a8a31c9434ff431cb789c809777505ec'}
},
'policy_id': u'46985da5-9684-402e-b0d7-b7adac909c3a'
}
We need to generalize this structure for all rule types so will
(effectively) rename the rule-specific keyword (e.g., in the above, the
first occurrence of 'bandwidth_limit_rule') to be 'rule_data'.
:param method_name: the name of the method to call
:type method_name: str
:param rule_obj: the rule object, which is sent as an argument to the
proxy method
:type rule_obj: a class from the rule_object (qos.objects.rule) module
"""
def _make_call(method_name, rule_obj, *args, **kwargs):
context = args[0]
args_list = list(args[1:])
params = kwargs
rule_data_name = rule_obj.rule_type + "_rule"
if rule_data_name in params:
params['rule_data'] = params.pop(rule_data_name)
return getattr(self, method_name)(
context, rule_obj, *args_list, **params
)
return lambda *args, **kwargs: _make_call(
method_name, rule_obj, *args, **kwargs)
def get_plugin_description(self):
return "QoS Service Plugin for ports and networks"
@ -195,13 +284,8 @@ class QoSPluginBase(service_base.ServicePluginBase):
return constants.QOS
@abc.abstractmethod
def get_policy(self, context, policy_id, fields=None):
pass
@abc.abstractmethod
def get_policies(self, context, filters=None, fields=None,
sorts=None, limit=None, marker=None,
page_reverse=False):
def get_rule_types(self, context, filters=None, fields=None, sorts=None,
limit=None, marker=None, page_reverse=False):
pass
@abc.abstractmethod
@ -217,59 +301,34 @@ class QoSPluginBase(service_base.ServicePluginBase):
pass
@abc.abstractmethod
def get_policy_bandwidth_limit_rule(self, context, rule_id,
policy_id, fields=None):
def get_policy(self, context, policy_id, fields=None):
pass
@abc.abstractmethod
def get_policy_bandwidth_limit_rules(self, context, policy_id,
filters=None, fields=None,
sorts=None, limit=None,
marker=None, page_reverse=False):
def get_policies(self, context, filters=None, fields=None, sorts=None,
limit=None, marker=None, page_reverse=False):
pass
@abc.abstractmethod
def create_policy_bandwidth_limit_rule(self, context, policy_id,
bandwidth_limit_rule):
def create_policy_rule(self, context, policy_id, rule_data, rule_obj):
pass
@abc.abstractmethod
def update_policy_bandwidth_limit_rule(self, context, rule_id, policy_id,
bandwidth_limit_rule):
def update_policy_rule(self, context, rule_id, policy_id, rule_data,
rule_obj):
pass
@abc.abstractmethod
def delete_policy_bandwidth_limit_rule(self, context, rule_id, policy_id):
def delete_policy_rule(self, context, rule_id, policy_id):
pass
@abc.abstractmethod
def get_policy_dscp_marking_rule(self, context, rule_id,
policy_id, fields=None):
def get_policy_rule(self, context, rule_id, policy_id, rule_obj,
fields=None):
pass
@abc.abstractmethod
def get_policy_dscp_marking_rules(self, context, policy_id,
filters=None, fields=None,
sorts=None, limit=None,
marker=None, page_reverse=False):
pass
@abc.abstractmethod
def create_policy_dscp_marking_rule(self, context, policy_id,
dscp_marking_rule):
pass
@abc.abstractmethod
def update_policy_dscp_marking_rule(self, context, rule_id, policy_id,
dscp_marking_rule):
pass
@abc.abstractmethod
def delete_policy_dscp_marking_rule(self, context, rule_id, policy_id):
pass
@abc.abstractmethod
def get_rule_types(self, context, filters=None, fields=None,
sorts=None, limit=None,
marker=None, page_reverse=False):
def get_policy_rules(self, context, policy_id, rule_obj,
filters=None, fields=None, sorts=None, limit=None,
marker=None, page_reverse=False):
pass

18
neutron/objects/base.py
View File

@ -341,6 +341,24 @@ class NeutronDbObject(NeutronObject):
keys[key] = getattr(self, key)
return self.modify_fields_to_db(keys)
def update_nonidentifying_fields(self, obj_data, reset_changes=False):
"""Updates non-identifying fields of an object.
:param obj_data: the full set of object data
:type obj_data: dict
:param reset_changes: indicates whether the object's current set of
changed fields should be cleared
:type reset_changes: boolean
:returns: None
"""
if reset_changes:
self.obj_reset_changes()
for k, v in obj_data.items():
if k not in self.primary_keys:
setattr(self, k, v)
def update(self):
updates = self._get_changed_persistent_fields()
updates = self._validate_changed_fields(updates)

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

@ -18,7 +18,6 @@ from neutron.db import api as db_api
from neutron.db import db_base_plugin_common
from neutron.extensions import qos
from neutron.objects.qos import policy as policy_object
from neutron.objects.qos import rule as rule_object
from neutron.objects.qos import rule_type as rule_type_object
from neutron.services.qos.notification_drivers import manager as driver_mgr
from neutron.services.qos import qos_consts
@ -27,9 +26,8 @@ from neutron.services.qos import qos_consts
class QoSPlugin(qos.QoSPluginBase):
"""Implementation of the Neutron QoS Service Plugin.
This class implements a Quality of Service plugin that
provides quality of service parameters over ports and
networks.
This class implements a Quality of Service plugin that provides quality of
service parameters over ports and networks.
"""
supported_extension_aliases = ['qos']
@ -41,29 +39,67 @@ class QoSPlugin(qos.QoSPluginBase):
@db_base_plugin_common.convert_result_to_dict
def create_policy(self, context, policy):
policy = policy_object.QosPolicy(context, **policy['policy'])
policy.create()
self.notification_driver_manager.create_policy(context, policy)
return policy
"""Create a QoS policy.
:param context: neutron api request context
:type context: neutron.context.Context
:param policy: policy data to be applied
:type policy: dict
:returns: a QosPolicy object
"""
policy_obj = policy_object.QosPolicy(context, **policy['policy'])
policy_obj.create()
self.notification_driver_manager.create_policy(context, policy_obj)
return policy_obj
@db_base_plugin_common.convert_result_to_dict
def update_policy(self, context, policy_id, policy):
obj = policy_object.QosPolicy(context, id=policy_id)
obj.obj_reset_changes()
for k, v in policy['policy'].items():
if k != 'id':
setattr(obj, k, v)
obj.update()
self.notification_driver_manager.update_policy(context, obj)
return obj
"""Update a QoS policy.
:param context: neutron api request context
:type context: neutron.context.Context
:param policy_id: the id of the QosPolicy to update
:param policy_id: str uuid
:param policy: new policy data to be applied
:type policy: dict
:returns: a QosPolicy object
"""
policy_data = policy['policy']
policy_obj = policy_object.QosPolicy(context, id=policy_id)
policy_obj.update_nonidentifying_fields(policy_data,
reset_changes=True)
policy_obj.update()
self.notification_driver_manager.update_policy(context, policy_obj)
return policy_obj
def delete_policy(self, context, policy_id):
"""Delete a QoS policy.
:param context: neutron api request context
:type context: neutron.context.Context
:param policy_id: the id of the QosPolicy to delete
:type policy_id: str uuid
:returns: None
"""
policy = policy_object.QosPolicy(context)
policy.id = policy_id
self.notification_driver_manager.delete_policy(context, policy)
policy.delete()
def _get_policy_obj(self, context, policy_id):
"""Fetch a QoS policy.
:param context: neutron api request context
:type context: neutron.context.Context
:param policy_id: the id of the QosPolicy to fetch
:type policy_id: str uuid
:returns: a QosPolicy object
:raises: n_exc.QosPolicyNotFound
"""
obj = policy_object.QosPolicy.get_object(context, id=policy_id)
if obj is None:
raise n_exc.QosPolicyNotFound(policy_id=policy_id)
@ -72,169 +108,169 @@ class QoSPlugin(qos.QoSPluginBase):
@db_base_plugin_common.filter_fields
@db_base_plugin_common.convert_result_to_dict
def get_policy(self, context, policy_id, fields=None):
"""Get a QoS policy.
:param context: neutron api request context
:type context: neutron.context.Context
:param policy_id: the id of the QosPolicy to update
:type policy_id: str uuid
:returns: a QosPolicy object
"""
return self._get_policy_obj(context, policy_id)
@db_base_plugin_common.filter_fields
@db_base_plugin_common.convert_result_to_dict
def get_policies(self, context, filters=None, fields=None,
sorts=None, limit=None, marker=None,
page_reverse=False):
def get_policies(self, context, filters=None, fields=None, sorts=None,
limit=None, marker=None, page_reverse=False):
"""Get QoS policies.
:param context: neutron api request context
:type context: neutron.context.Context
:param filters: search criteria
:type filters: dict
:returns: QosPolicy objects meeting the search criteria
"""
return policy_object.QosPolicy.get_objects(context, **filters)
#TODO(mangelajo): need to add a proxy catch-all for rules, so
# we capture the API function call, and just pass
# the rule type as a parameter removing lots of
# future code duplication when we have more rules.
@db_base_plugin_common.convert_result_to_dict
def create_policy_bandwidth_limit_rule(self, context, policy_id,
bandwidth_limit_rule):
# make sure we will have a policy object to push resource update
with db_api.autonested_transaction(context.session):
# first, validate that we have access to the policy
policy = self._get_policy_obj(context, policy_id)
rule = rule_object.QosBandwidthLimitRule(
context, qos_policy_id=policy_id,
**bandwidth_limit_rule['bandwidth_limit_rule'])
rule.create()
policy.reload_rules()
self.notification_driver_manager.update_policy(context, policy)
return rule
@db_base_plugin_common.convert_result_to_dict
def update_policy_bandwidth_limit_rule(self, context, rule_id, policy_id,
bandwidth_limit_rule):
# make sure we will have a policy object to push resource update
with db_api.autonested_transaction(context.session):
# first, validate that we have access to the policy
policy = self._get_policy_obj(context, policy_id)
# check if the rule belong to the policy
policy.get_rule_by_id(rule_id)
rule = rule_object.QosBandwidthLimitRule(
context, id=rule_id)
rule.obj_reset_changes()
for k, v in bandwidth_limit_rule['bandwidth_limit_rule'].items():
if k != 'id':
setattr(rule, k, v)
rule.update()
policy.reload_rules()
self.notification_driver_manager.update_policy(context, policy)
return rule
def delete_policy_bandwidth_limit_rule(self, context, rule_id, policy_id):
# make sure we will have a policy object to push resource update
with db_api.autonested_transaction(context.session):
# first, validate that we have access to the policy
policy = self._get_policy_obj(context, policy_id)
rule = policy.get_rule_by_id(rule_id)
rule.delete()
policy.reload_rules()
self.notification_driver_manager.update_policy(context, policy)
@db_base_plugin_common.filter_fields
@db_base_plugin_common.convert_result_to_dict
def get_policy_bandwidth_limit_rule(self, context, rule_id,
policy_id, fields=None):
# make sure we have access to the policy when fetching the rule
with db_api.autonested_transaction(context.session):
# first, validate that we have access to the policy
self._get_policy_obj(context, policy_id)
rule = rule_object.QosBandwidthLimitRule.get_object(
context, id=rule_id)
if not rule:
raise n_exc.QosRuleNotFound(policy_id=policy_id, rule_id=rule_id)
return rule
@db_base_plugin_common.filter_fields
@db_base_plugin_common.convert_result_to_dict
def get_policy_bandwidth_limit_rules(self, context, policy_id,
filters=None, fields=None,
sorts=None, limit=None,
marker=None, page_reverse=False):
# make sure we have access to the policy when fetching rules
with db_api.autonested_transaction(context.session):
# first, validate that we have access to the policy
self._get_policy_obj(context, policy_id)
filters = filters or dict()
filters[qos_consts.QOS_POLICY_ID] = policy_id
return rule_object.QosBandwidthLimitRule.get_objects(context,
**filters)
@db_base_plugin_common.convert_result_to_dict
def create_policy_dscp_marking_rule(self, context, policy_id,
dscp_marking_rule):
with db_api.autonested_transaction(context.session):
# first, validate that we have access to the policy
policy = self._get_policy_obj(context, policy_id)
rule = rule_object.QosDscpMarkingRule(
context, qos_policy_id=policy_id,
**dscp_marking_rule['dscp_marking_rule'])
rule.create()
policy.reload_rules()
self.notification_driver_manager.update_policy(context, policy)
return rule
@db_base_plugin_common.convert_result_to_dict
def update_policy_dscp_marking_rule(self, context, rule_id, policy_id,
dscp_marking_rule):
with db_api.autonested_transaction(context.session):
# first, validate that we have access to the policy
policy = self._get_policy_obj(context, policy_id)
# check if the rule belong to the policy
policy.get_rule_by_id(rule_id)
rule = rule_object.QosDscpMarkingRule(
context, id=rule_id)
rule.obj_reset_changes()
for k, v in dscp_marking_rule['dscp_marking_rule'].items():
if k != 'id':
setattr(rule, k, v)
rule.update()
policy.reload_rules()
self.notification_driver_manager.update_policy(context, policy)
return rule
def delete_policy_dscp_marking_rule(self, context, rule_id, policy_id):
# make sure we will have a policy object to push resource update
with db_api.autonested_transaction(context.session):
# first, validate that we have access to the policy
policy = self._get_policy_obj(context, policy_id)
rule = policy.get_rule_by_id(rule_id)
rule.delete()
policy.reload_rules()
self.notification_driver_manager.update_policy(context, policy)
@db_base_plugin_common.filter_fields
@db_base_plugin_common.convert_result_to_dict
def get_policy_dscp_marking_rule(self, context, rule_id,
policy_id, fields=None):
# make sure we have access to the policy when fetching the rule
with db_api.autonested_transaction(context.session):
# first, validate that we have access to the policy
self._get_policy_obj(context, policy_id)
rule = rule_object.QosDscpMarkingRule.get_object(
context, id=rule_id)
if not rule:
raise n_exc.QosRuleNotFound(policy_id=policy_id, rule_id=rule_id)
return rule
@db_base_plugin_common.filter_fields
@db_base_plugin_common.convert_result_to_dict
def get_policy_dscp_marking_rules(self, context, policy_id,
filters=None, fields=None,
sorts=None, limit=None,
marker=None, page_reverse=False):
# make sure we have access to the policy when fetching rules
with db_api.autonested_transaction(context.session):
# first, validate that we have access to the policy
self._get_policy_obj(context, policy_id)
filters = filters or dict()
filters[qos_consts.QOS_POLICY_ID] = policy_id
return rule_object.QosDscpMarkingRule.get_objects(context,
**filters)
# TODO(QoS): enforce rule types when accessing rule objects
@db_base_plugin_common.filter_fields
@db_base_plugin_common.convert_result_to_dict
def get_rule_types(self, context, filters=None, fields=None,
sorts=None, limit=None,
marker=None, page_reverse=False):
if not filters:
filters = {}
return rule_type_object.QosRuleType.get_objects(**filters)
@db_base_plugin_common.convert_result_to_dict
def create_policy_rule(self, context, rule_obj, policy_id, rule_data):
"""Create a QoS policy rule.
:param context: neutron api request context
:type context: neutron.context.Context
:param rule_obj: the rule object
:type rule_obj: a class from the rule_object (qos.objects.rule) module
:param policy_id: the id of the QosPolicy for which to create the rule
:type policy_id: str uuid
:param rule_data: the rule data to be applied
:type rule_data: dict
:returns: a QoS policy rule object
"""
rule_type = rule_obj.rule_type
rule_data = rule_data[rule_type + '_rule']
with db_api.autonested_transaction(context.session):
# Ensure that we have access to the policy.
policy = self._get_policy_obj(context, policy_id)
rule = rule_obj(context, qos_policy_id=policy_id, **rule_data)
rule.create()
policy.reload_rules()
self.notification_driver_manager.update_policy(context, policy)
return rule
@db_base_plugin_common.convert_result_to_dict
def update_policy_rule(self, context, rule_obj, rule_id, policy_id,
rule_data):
"""Update a QoS policy rule.
:param context: neutron api request context
:type context: neutron.context.Context
:param rule_obj: the rule object
:type rule_obj: a class from the rule_object (qos.objects.rule) module
:param rule_id: the id of the QoS policy rule to update
:type rule_id: str uuid
:param policy_id: the id of the rule's policy
:type policy_id: str uuid
:param rule_data: the new rule data to update
:type rule_data: dict
:returns: a QoS policy rule object
"""
rule_type = rule_obj.rule_type
rule_data = rule_data[rule_type + '_rule']
with db_api.autonested_transaction(context.session):
# Ensure we have access to the policy.
policy = self._get_policy_obj(context, policy_id)
# Ensure the rule belongs to the policy.
policy.get_rule_by_id(rule_id)
rule = rule_obj(context, id=rule_id)
rule.update_nonidentifying_fields(rule_data, reset_changes=True)
rule.update()
policy.reload_rules()
self.notification_driver_manager.update_policy(context, policy)
return rule
def delete_policy_rule(self, context, rule_obj, rule_id, policy_id):
"""Delete a QoS policy rule.
:param context: neutron api request context
:type context: neutron.context.Context
:param rule_obj: the rule object
:type rule_obj: a class from the rule_object (qos.objects.rule) module
:param rule_id: the id of the QosPolicy Rule to delete
:type rule_id: str uuid
:param policy_id: the id of the rule's policy
:type policy_id: str uuid
:returns: None
"""
with db_api.autonested_transaction(context.session):
# Ensure we have access to the policy.
policy = self._get_policy_obj(context, policy_id)
rule = policy.get_rule_by_id(rule_id)
rule.delete()
policy.reload_rules()
self.notification_driver_manager.update_policy(context, policy)
@db_base_plugin_common.filter_fields
@db_base_plugin_common.convert_result_to_dict
def get_policy_rule(self, context, rule_obj, rule_id, policy_id,
fields=None):
"""Get a QoS policy rule.
:param context: neutron api request context
:type context: neutron.context.Context
:param rule_obj: the rule object
:type rule_obj: a class from the rule_object (qos.objects.rule) module
:param rule_id: the id of the QoS policy rule to get
:type rule_id: str uuid
:param policy_id: the id of the rule's policy
:type policy_id: str uuid
:returns: a QoS policy rule object
:raises: n_exc.QosRuleNotFound
"""
with db_api.autonested_transaction(context.session):
# Ensure we have access to the policy.
self._get_policy_obj(context, policy_id)
rule = rule_obj.get_object(context, id=rule_id)
if not rule:
raise n_exc.QosRuleNotFound(policy_id=policy_id, rule_id=rule_id)
return rule
# TODO(QoS): enforce rule types when accessing rule objects
@db_base_plugin_common.filter_fields
@db_base_plugin_common.convert_result_to_dict
def get_policy_rules(self, context, rule_obj, policy_id, filters=None,
fields=None, sorts=None, limit=None, marker=None,
page_reverse=False):
"""Get QoS policy rules.
:param context: neutron api request context
:type context: neutron.context.Context
:param rule_obj: the rule object
:type rule_obj: a class from the rule_object (qos.objects.rule) module
:param policy_id: the id of the QosPolicy for which to get rules
:type policy_id: str uuid
:returns: QoS policy rule objects meeting the search criteria
"""
with db_api.autonested_transaction(context.session):
# Ensure we have access to the policy.
self._get_policy_obj(context, policy_id)
filters = filters or dict()
filters[qos_consts.QOS_POLICY_ID] = policy_id
return rule_obj.get_objects(context, **filters)

21
neutron/tests/unit/objects/test_base.py
View File

@ -477,6 +477,27 @@ class BaseObjectIfaceTestCase(_BaseObjectTestCase, test_base.BaseTestCase):
obj = self._test_class(self.context, **self.obj_fields[0])
self.assertRaises(base.NeutronDbObjectDuplicateEntry, obj.create)
def test_update_nonidentifying_fields(self):
if not self._test_class.primary_keys:
self.skipTest(
'Test class %r has no primary keys' % self._test_class)
with mock.patch.object(obj_base.VersionedObject, 'obj_reset_changes'):
expected = self._test_class(self.context, **self.obj_fields[0])
for key, val in self.obj_fields[1].items():
if key not in expected.primary_keys:
setattr(expected, key, val)
observed = self._test_class(self.context, **self.obj_fields[0])
observed.update_nonidentifying_fields(self.obj_fields[1],
reset_changes=True)
self.assertEqual(expected, observed)
self.assertTrue(observed.obj_reset_changes.called)
with mock.patch.object(obj_base.VersionedObject, 'obj_reset_changes'):
obj = self._test_class(self.context, **self.obj_fields[0])
obj.update_nonidentifying_fields(self.obj_fields[1])
self.assertFalse(obj.obj_reset_changes.called)
@mock.patch.object(obj_db_api, 'update_object')
def test_update_no_changes(self, update_mock):
with mock.patch.object(base.NeutronDbObject,

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

@ -21,6 +21,7 @@ from neutron.objects import base as base_object
from neutron.objects.qos import policy as policy_object
from neutron.objects.qos import rule as rule_object
from neutron.plugins.common import constants
from neutron.services.qos import qos_consts
from neutron.tests.unit.services.qos import base
@ -157,6 +158,17 @@ class TestQosPlugin(base.BaseQosTestCase):
self.qos_plugin.delete_policy_bandwidth_limit_rule,
self.ctxt, self.rule.id, _policy.id)
def test_get_policy_bandwidth_limit_rule(self):
with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
return_value=self.policy):
with mock.patch('neutron.objects.qos.rule.'
'QosBandwidthLimitRule.'
'get_object') as get_object_mock:
self.qos_plugin.get_policy_bandwidth_limit_rule(
self.ctxt, self.rule.id, self.policy.id)
get_object_mock.assert_called_once_with(self.ctxt,
id=self.rule.id)
def test_get_policy_bandwidth_limit_rules_for_policy(self):
with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
return_value=self.policy):
@ -300,3 +312,19 @@ class TestQosPlugin(base.BaseQosTestCase):
n_exc.QosPolicyNotFound,
self.qos_plugin.delete_policy_bandwidth_limit_rule,
self.ctxt, self.rule.id, self.policy.id)
def test_verify_bad_method_call(self):
self.assertRaises(AttributeError, getattr, self.qos_plugin,
'create_policy_bandwidth_limit_rules')
def test_get_rule_types(self):
core_plugin = manager.NeutronManager.get_plugin()
rule_types_mock = mock.PropertyMock(
return_value=qos_consts.VALID_RULE_TYPES)
filters = {'type': 'type_id'}
with mock.patch.object(core_plugin, 'supported_qos_rule_types',
new_callable=rule_types_mock,
create=True):
types = self.qos_plugin.get_rule_types(self.ctxt, filters=filters)
self.assertEqual(sorted(qos_consts.VALID_RULE_TYPES),
sorted(type_['type'] for type_ in types))