From bb1546df15b57923fdbb9057407274bdcce59c50 Mon Sep 17 00:00:00 2001 From: John Schwarz Date: Mon, 3 Aug 2015 18:55:31 +0300 Subject: [PATCH] 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 --- neutron/core_extensions/qos.py | 12 +++---- neutron/tests/api/test_qos.py | 66 ++++++++++++++++++++-------------- 2 files changed, 45 insertions(+), 33 deletions(-) diff --git a/neutron/core_extensions/qos.py b/neutron/core_extensions/qos.py index c2caae0cf..72fb89883 100644 --- 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 81f598244..d281094b3 100644 --- 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') -- 2.45.2