openstack
/
neutron
Code Issues Proposed changes

QoS core extension: fixed dict extension when QoS policy is unset 

Previously, if QoS policy was detached from a core resource, core
resource extension didn't update the resource dict with the new QoS
policy value (None), and that resulted in no notification sent to the
agent about the change, so QoS rules were not flushed from the affected
ports.

Change-Id: I22397af3a43254d146abaa4a4429ac654b4c3c50
Partially-Implements: quantum-qos-api
This commit is contained in:
Ihar Hrachyshka 2015-08-08 00:21:09 +02:00
parent 088289acd2
commit 5f5be37899
2 changed files with 46 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
neutron/core_extensions/qos.py
View File

@ -46,7 +46,7 @@ class QosCoreResourceExtension(base.CoreResourceExtension):
# 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
port[qos_consts.QOS_POLICY_ID] = qos_policy_id
def _update_network_policy(self, context, network, network_changes):
old_policy = policy_object.QosPolicy.get_network_policy(
@ -55,13 +55,13 @@ class QosCoreResourceExtension(base.CoreResourceExtension):
old_policy.detach_network(network['id'])
qos_policy_id = network_changes.get(qos_consts.QOS_POLICY_ID)
if 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
network[qos_consts.QOS_POLICY_ID] = qos_policy_id
def _exec(self, method_name, context, kwargs):
with db_api.autonested_transaction(context.session):

46
neutron/tests/unit/core_extensions/test_qos.py
View File

@ -72,8 +72,29 @@ class QosCoreResourceExtensionTestCase(base.BaseTestCase):
def test_process_fields_port_updated_policy(self):
with self._mock_plugin_loaded(True):
qos_policy_id = mock.Mock()
qos_policy1_id = mock.Mock()
qos_policy2_id = mock.Mock()
port_id = mock.Mock()
actual_port = {'id': port_id,
qos_consts.QOS_POLICY_ID: qos_policy1_id}
old_qos_policy = mock.MagicMock()
self.policy_m.get_port_policy = mock.Mock(
return_value=old_qos_policy)
new_qos_policy = mock.MagicMock()
self.policy_m.get_by_id = mock.Mock(return_value=new_qos_policy)
self.core_extension.process_fields(
self.context, base_core.PORT,
{qos_consts.QOS_POLICY_ID: qos_policy2_id},
actual_port)
old_qos_policy.detach_port.assert_called_once_with(port_id)
new_qos_policy.attach_port.assert_called_once_with(port_id)
self.assertEqual(qos_policy2_id, actual_port['qos_policy_id'])
def test_process_resource_port_updated_no_policy(self):
with self._mock_plugin_loaded(True):
port_id = mock.Mock()
qos_policy_id = mock.Mock()
actual_port = {'id': port_id,
qos_consts.QOS_POLICY_ID: qos_policy_id}
old_qos_policy = mock.MagicMock()
@ -83,11 +104,30 @@ class QosCoreResourceExtensionTestCase(base.BaseTestCase):
self.policy_m.get_by_id = mock.Mock(return_value=new_qos_policy)
self.core_extension.process_fields(
self.context, base_core.PORT,
{qos_consts.QOS_POLICY_ID: qos_policy_id},
{qos_consts.QOS_POLICY_ID: None},
actual_port)
old_qos_policy.detach_port.assert_called_once_with(port_id)
new_qos_policy.attach_port.assert_called_once_with(port_id)
self.assertIsNone(actual_port['qos_policy_id'])
def test_process_resource_network_updated_no_policy(self):
with self._mock_plugin_loaded(True):
network_id = mock.Mock()
qos_policy_id = mock.Mock()
actual_network = {'id': network_id,
qos_consts.QOS_POLICY_ID: qos_policy_id}
old_qos_policy = mock.MagicMock()
self.policy_m.get_network_policy = mock.Mock(
return_value=old_qos_policy)
new_qos_policy = mock.MagicMock()
self.policy_m.get_by_id = mock.Mock(return_value=new_qos_policy)
self.core_extension.process_fields(
self.context, base_core.NETWORK,
{qos_consts.QOS_POLICY_ID: None},
actual_network)
old_qos_policy.detach_network.assert_called_once_with(network_id)
self.assertIsNone(actual_network['qos_policy_id'])
def test_process_fields_network_new_policy(self):
with self._mock_plugin_loaded(True):