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 b0c434050952351c652a068660f77ce6505dc80f..b4d3f5a4b2500ef69fd11275330a44f611f67eb1 100644 (file)
--- 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 d5434f5bf9df9a0098c1a51fafde631eb0744089..23135bf82be7a942c580d4137c5ecd6d9a1fbfd7 100644 (file)
--- 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 3683b4628882b7459acc6723b78ef9420f280be5..5332b45d19ae9b5070646faa1b109ed06ad61f70 100644 (file)
--- 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 d4927b67778c4a4e186ec419ec070bc12cf08dd3..8254da6356f3bd4a71e844bda81b1ea9be2fb14c 100644 (file)
--- 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)