From: Ihar Hrachyshka Date: Thu, 30 Jul 2015 11:51:24 +0000 (+0200) Subject: qos: forbid creating rules when there is no access to policy X-Git-Url: https://review.fuel-infra.org/gitweb?a=commitdiff_plain;h=11924b11532979090abdba9960dd2dad81debcfe;p=openstack-build%2Fneutron-build.git qos: forbid creating rules when there is no access to policy Change-Id: If06de416dfe0eb7115fd4be9feb461fae8e8358d Partially-Implements: blueprint quantum-qos-api --- diff --git a/neutron/common/exceptions.py b/neutron/common/exceptions.py index b0c434050..b4d3f5a4b 100644 --- a/neutron/common/exceptions.py +++ b/neutron/common/exceptions.py @@ -89,6 +89,10 @@ class PortNotFound(NotFound): message = _("Port %(port_id)s could not be found") +class QosPolicyNotFound(NotFound): + message = _("QoS policy %(policy_id)s could not be found") + + class PortNotFoundOnNetwork(NotFound): message = _("Port %(port_id)s could not be found " "on network %(net_id)s") diff --git a/neutron/services/qos/qos_plugin.py b/neutron/services/qos/qos_plugin.py index d5434f5bf..23135bf82 100644 --- a/neutron/services/qos/qos_plugin.py +++ b/neutron/services/qos/qos_plugin.py @@ -15,6 +15,7 @@ from oslo_log import log as logging +from neutron.common import exceptions as n_exc from neutron.db import db_base_plugin_common from neutron.extensions import qos from neutron.objects.qos import policy as policy_object @@ -61,11 +62,10 @@ class QoSPlugin(qos.QoSPluginBase): policy.delete() def _get_policy_obj(self, context, policy_id): - return policy_object.QosPolicy.get_by_id(context, policy_id) - - def _update_policy_on_driver(self, context, policy_id): - policy = self._get_policy_obj(context, policy_id) - self.notification_driver_manager.update_policy(policy) + obj = policy_object.QosPolicy.get_by_id(context, policy_id) + if obj is None: + raise n_exc.QosPolicyNotFound(policy_id=policy_id) + return obj @db_base_plugin_common.filter_fields def get_policy(self, context, policy_id, fields=None): @@ -89,31 +89,39 @@ class QoSPlugin(qos.QoSPluginBase): # in the future we need an inter-rule validation # mechanism to verify all created rules will # play well together. + # 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() - self._update_policy_on_driver(context, policy_id) + self.notification_driver_manager.update_policy(policy) return rule.to_dict() def update_policy_bandwidth_limit_rule(self, context, rule_id, policy_id, bandwidth_limit_rule): + # validate that we have access to the policy + policy = self._get_policy_obj(context, policy_id) rule = rule_object.QosBandwidthLimitRule( context, **bandwidth_limit_rule['bandwidth_limit_rule']) rule.id = rule_id rule.update() - self._update_policy_on_driver(context, policy_id) + self.notification_driver_manager.update_policy(policy) return rule.to_dict() def delete_policy_bandwidth_limit_rule(self, context, rule_id, policy_id): + # validate that we have access to the policy + policy = self._get_policy_obj(context, policy_id) rule = rule_object.QosBandwidthLimitRule(context) rule.id = rule_id rule.delete() - self._update_policy_on_driver(context, policy_id) + self.notification_driver_manager.update_policy(policy) @db_base_plugin_common.filter_fields def get_policy_bandwidth_limit_rule(self, context, rule_id, policy_id, fields=None): + # validate that we have access to the policy + self._get_policy_obj(context, policy_id) return rule_object.QosBandwidthLimitRule.get_by_id(context, rule_id).to_dict() @@ -123,6 +131,8 @@ class QoSPlugin(qos.QoSPluginBase): sorts=None, limit=None, marker=None, page_reverse=False): #TODO(QoS): Support all the optional parameters + # validate that we have access to the policy + self._get_policy_obj(context, policy_id) return [rule_obj.to_dict() for rule_obj in rule_object.QosBandwidthLimitRule.get_objects(context)] diff --git a/neutron/tests/api/test_qos.py b/neutron/tests/api/test_qos.py index 3683b4628..5332b45d1 100644 --- a/neutron/tests/api/test_qos.py +++ b/neutron/tests/api/test_qos.py @@ -76,7 +76,7 @@ class QosTestJSON(base.BaseAdminNetworkTest): self.assertEqual('test-policy', retrieved_policy['name']) self.admin_client.delete_qos_policy(policy['id']) - self.assertRaises(exceptions.ServerFault, + self.assertRaises(exceptions.NotFound, self.admin_client.show_qos_policy, policy['id']) @test.attr(type='smoke') diff --git a/neutron/tests/unit/services/qos/test_qos_plugin.py b/neutron/tests/unit/services/qos/test_qos_plugin.py index d4927b677..8254da635 100644 --- a/neutron/tests/unit/services/qos/test_qos_plugin.py +++ b/neutron/tests/unit/services/qos/test_qos_plugin.py @@ -15,6 +15,7 @@ from oslo_config import cfg from neutron.api.rpc.callbacks import events from neutron.api.rpc.callbacks import resources +from neutron.common import exceptions as n_exc from neutron import context from neutron import manager from neutron.objects.qos import policy as policy_object @@ -74,30 +75,84 @@ class TestQosPlugin(base.BaseTestCase): self.assertIsInstance( self.registry_m.call_args[0][2], policy_object.QosPolicy) - def test_qos_plugin_add_policy(self): + def test_add_policy(self): self.qos_plugin.create_policy(self.ctxt, self.policy_data) self.assertFalse(self.registry_m.called) - def test_qos_plugin_update_policy(self): + def test_update_policy(self): self.qos_plugin.update_policy( self.ctxt, self.policy.id, self.policy_data) self._validate_registry_params(events.UPDATED) - def test_qos_plugin_delete_policy(self): + def test_delete_policy(self): self.qos_plugin.delete_policy(self.ctxt, self.policy.id) self._validate_registry_params(events.DELETED) - def test_qos_plugin_create_policy_rule(self): - self.qos_plugin.create_policy_bandwidth_limit_rule( - self.ctxt, self.policy.id, self.rule_data) - self._validate_registry_params(events.UPDATED) - - def test_qos_plugin_update_policy_rule(self): - self.qos_plugin.update_policy_bandwidth_limit_rule( - self.ctxt, self.rule.id, self.policy.id, self.rule_data) - self._validate_registry_params(events.UPDATED) - - def test_qos_plugin_delete_policy_rule(self): - self.qos_plugin.delete_policy_bandwidth_limit_rule( - self.ctxt, self.rule.id, self.policy.id) - self._validate_registry_params(events.UPDATED) + def test_create_policy_rule(self): + with mock.patch('neutron.objects.qos.policy.QosPolicy.get_by_id', + return_value=self.policy): + self.qos_plugin.create_policy_bandwidth_limit_rule( + self.ctxt, self.policy.id, self.rule_data) + self._validate_registry_params(events.UPDATED) + + def test_update_policy_rule(self): + with mock.patch('neutron.objects.qos.policy.QosPolicy.get_by_id', + return_value=self.policy): + self.qos_plugin.update_policy_bandwidth_limit_rule( + self.ctxt, self.rule.id, self.policy.id, self.rule_data) + self._validate_registry_params(events.UPDATED) + + def test_delete_policy_rule(self): + with mock.patch('neutron.objects.qos.policy.QosPolicy.get_by_id', + return_value=self.policy): + self.qos_plugin.delete_policy_bandwidth_limit_rule( + self.ctxt, self.rule.id, self.policy.id) + self._validate_registry_params(events.UPDATED) + + def test_get_policy_for_nonexistent_policy(self): + with mock.patch('neutron.objects.qos.policy.QosPolicy.get_by_id', + return_value=None): + self.assertRaises( + n_exc.QosPolicyNotFound, + self.qos_plugin.get_policy, + self.ctxt, self.policy.id) + + def test_get_policy_bandwidth_limit_rule_for_nonexistent_policy(self): + with mock.patch('neutron.objects.qos.policy.QosPolicy.get_by_id', + return_value=None): + self.assertRaises( + n_exc.QosPolicyNotFound, + self.qos_plugin.get_policy_bandwidth_limit_rule, + self.ctxt, self.rule.id, self.policy.id) + + def test_get_policy_bandwidth_limit_rules_for_nonexistent_policy(self): + with mock.patch('neutron.objects.qos.policy.QosPolicy.get_by_id', + return_value=None): + self.assertRaises( + n_exc.QosPolicyNotFound, + self.qos_plugin.get_policy_bandwidth_limit_rules, + self.ctxt, self.policy.id) + + def test_create_policy_rule_for_nonexistent_policy(self): + with mock.patch('neutron.objects.qos.policy.QosPolicy.get_by_id', + return_value=None): + self.assertRaises( + n_exc.QosPolicyNotFound, + self.qos_plugin.create_policy_bandwidth_limit_rule, + self.ctxt, self.policy.id, self.rule_data) + + def test_update_policy_rule_for_nonexistent_policy(self): + with mock.patch('neutron.objects.qos.policy.QosPolicy.get_by_id', + return_value=None): + self.assertRaises( + n_exc.QosPolicyNotFound, + self.qos_plugin.update_policy_bandwidth_limit_rule, + self.ctxt, self.rule.id, self.policy.id, self.rule_data) + + def test_delete_policy_rule_for_nonexistent_policy(self): + with mock.patch('neutron.objects.qos.policy.QosPolicy.get_by_id', + return_value=None): + self.assertRaises( + n_exc.QosPolicyNotFound, + self.qos_plugin.delete_policy_bandwidth_limit_rule, + self.ctxt, self.rule.id, self.policy.id)