Forbid attaching rules if policy isn't accessible

Following up patch If06de416dfe0eb7115fd4be9feb461fae8e8358d, this patch
continues to make sure all access to QoS policies are attempted safely -
if the policy doesn't exist or it's not accessible (for tenant_id
reasons), then an exception will be raised instead.

Change-Id: Id7e64c745cdd63d650a3f69572635dc10197259c
Partially-Implements: quantum-qos-api

diff --git a/neutron/core_extensions/qos.py b/neutron/core_extensions/qos.py
index c2caae0cf8f3a9adb18ccdbaf5419a95248a88b5..72fb898836c29c34ee41355b80c81a7ad285ddc9 100644 (file)
--- a/neutron/core_extensions/qos.py
+++ b/neutron/core_extensions/qos.py
@@ -13,6 +13,7 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+from neutron.common import exceptions as n_exc
 from neutron.core_extensions import base
 from neutron.db import api as db_api
 from neutron import manager
@@ -31,7 +32,10 @@ class QosCoreResourceExtension(base.CoreResourceExtension):
         return self._plugin_loaded
 
     def _get_policy_obj(self, context, policy_id):
-        return policy_object.QosPolicy.get_by_id(context, policy_id)
+        obj = policy_object.QosPolicy.get_by_id(context, policy_id)
+        if obj is None:
+            raise n_exc.QosPolicyNotFound(policy_id=policy_id)
+        return obj
 
     def _update_port_policy(self, context, port, port_changes):
         old_policy = policy_object.QosPolicy.get_port_policy(
@@ -42,9 +46,6 @@ class QosCoreResourceExtension(base.CoreResourceExtension):
         qos_policy_id = port_changes.get(qos_consts.QOS_POLICY_ID)
         if qos_policy_id is not None:
             policy = self._get_policy_obj(context, qos_policy_id)
-            #TODO(QoS): If the policy doesn't exist (or if it is not shared and
-            #           the tenant id doesn't match the context's), this will
-            #           raise an exception (policy is None).
             policy.attach_port(port['id'])
         port[qos_consts.QOS_POLICY_ID] = qos_policy_id
 
@@ -57,9 +58,6 @@ class QosCoreResourceExtension(base.CoreResourceExtension):
         qos_policy_id = network_changes.get(qos_consts.QOS_POLICY_ID)
         if qos_policy_id is not None:
             policy = self._get_policy_obj(context, qos_policy_id)
-            #TODO(QoS): If the policy doesn't exist (or if it is not shared and
-            #           the tenant id doesn't match the context's), this will
-            #           raise an exception (policy is None).
             policy.attach_network(network['id'])
         network[qos_consts.QOS_POLICY_ID] = qos_policy_id
 

diff --git a/neutron/tests/api/test_qos.py b/neutron/tests/api/test_qos.py
index 81f5982449537c4a46c108247d98b3abcb91dbe2..d281094b36d66a889b4fe209565f11e687919426 100644 (file)
--- a/neutron/tests/api/test_qos.py
+++ b/neutron/tests/api/test_qos.py
@@ -157,19 +157,25 @@ class QosTestJSON(base.BaseAdminNetworkTest):
 
         self._disassociate_network(self.client, network['id'])
 
-#    @test.attr(type='smoke')
-#    @test.idempotent_id('1aa55a79-324f-47d9-a076-894a8fc2448b')
-#    def test_policy_association_with_network_non_shared_policy(self):
-#        policy = self.create_qos_policy(name='test-policy',
-#                                        description='test policy',
-#                                        shared=False)
-#        #TODO(QoS): This currently raises an exception on the server side. See
-#        #           core_extensions/qos.py for comments on this subject.
-#        network = self.create_network('test network',
-#                                      qos_policy_id=policy['id'])
-#
-#        retrieved_network = self.admin_client.show_network(network['id'])
-#        self.assertIsNone(retrieved_network['network']['qos_policy_id'])
+    @test.attr(type='smoke')
+    @test.idempotent_id('9efe63d0-836f-4cc2-b00c-468e63aa614e')
+    def test_policy_association_with_network_nonexistent_policy(self):
+        self.assertRaises(
+            exceptions.NotFound,
+            self.create_network,
+            'test network',
+            qos_policy_id='9efe63d0-836f-4cc2-b00c-468e63aa614e')
+
+    @test.attr(type='smoke')
+    @test.idempotent_id('1aa55a79-324f-47d9-a076-894a8fc2448b')
+    def test_policy_association_with_network_non_shared_policy(self):
+        policy = self.create_qos_policy(name='test-policy',
+                                        description='test policy',
+                                        shared=False)
+        self.assertRaises(
+            exceptions.NotFound,
+            self.create_network,
+            'test network', qos_policy_id=policy['id'])
 
     @test.attr(type='smoke')
     @test.idempotent_id('09a9392c-1359-4cbb-989f-fb768e5834a8')
@@ -209,19 +215,27 @@ class QosTestJSON(base.BaseAdminNetworkTest):
 
         self._disassociate_port(port['id'])
 
-#    @test.attr(type='smoke')
-#    @test.idempotent_id('f53d961c-9fe5-4422-8b66-7add972c6031')
-#    def test_policy_association_with_port_non_shared_policy(self):
-#        policy = self.create_qos_policy(name='test-policy',
-#                                        description='test policy',
-#                                        shared=False)
-#        network = self.create_shared_network('test network')
-#        #TODO(QoS): This currently raises an exception on the server side. See
-#        #           core_extensions/qos.py for comments on this subject.
-#        port = self.create_port(network, qos_policy_id=policy['id'])
-#
-#        retrieved_port = self.admin_client.show_port(port['id'])
-#        self.assertIsNone(retrieved_port['port']['qos_policy_id'])
+    @test.attr(type='smoke')
+    @test.idempotent_id('49e02f5a-e1dd-41d5-9855-cfa37f2d195e')
+    def test_policy_association_with_port_nonexistent_policy(self):
+        network = self.create_shared_network('test network')
+        self.assertRaises(
+            exceptions.NotFound,
+            self.create_port,
+            network,
+            qos_policy_id='49e02f5a-e1dd-41d5-9855-cfa37f2d195e')
+
+    @test.attr(type='smoke')
+    @test.idempotent_id('f53d961c-9fe5-4422-8b66-7add972c6031')
+    def test_policy_association_with_port_non_shared_policy(self):
+        policy = self.create_qos_policy(name='test-policy',
+                                        description='test policy',
+                                        shared=False)
+        network = self.create_shared_network('test network')
+        self.assertRaises(
+            exceptions.NotFound,
+            self.create_port,
+            network, qos_policy_id=policy['id'])
 
     @test.attr(type='smoke')
     @test.idempotent_id('f8163237-fba9-4db5-9526-bad6d2343c76')