openstack
/
neutron
Code Issues Proposed changes

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
This commit is contained in:
John Schwarz 2015-08-03 18:55:31 +03:00 committed by Ihar Hrachyshka
parent 08f0bb9ce5
commit bb1546df15
2 changed files with 45 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
neutron/core_extensions/qos.py
View File

@ -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

66
neutron/tests/api/test_qos.py
View File

@ -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')