From dd6cd44b2155605345a6925d3373bb6afe0bcf62 Mon Sep 17 00:00:00 2001 From: Miguel Angel Ajo Date: Thu, 9 Jul 2015 16:39:05 +0200 Subject: [PATCH] Implement QoS policy detach from port and network Includes db.qos.api calls to delete port<->qos_profile and network<->qos_profile bindings. Change-Id: I8ab3e885bdf010fe95529157f3db4f1089326c86 --- neutron/db/qos/api.py | 17 ++++ neutron/objects/qos/policy.py | 10 ++- neutron/tests/unit/objects/qos/test_policy.py | 89 +++++++++++++------ 3 files changed, 83 insertions(+), 33 deletions(-) diff --git a/neutron/db/qos/api.py b/neutron/db/qos/api.py index 632c57e9e..40b8ab77b 100644 --- a/neutron/db/qos/api.py +++ b/neutron/db/qos/api.py @@ -10,6 +10,7 @@ # License for the specific language governing permissions and limitations # under the License. +from neutron.db import common_db_mixin as db from neutron.db.qos import models @@ -20,8 +21,24 @@ def create_policy_network_binding(context, policy_id, network_id): context.session.add(db_obj) +def delete_policy_network_binding(context, policy_id, network_id): + with context.session.begin(subtransactions=True): + db_object = (db.model_query(context, models.QosNetworkPolicyBinding) + .filter_by(policy_id=policy_id, + network_id=network_id).one()) + context.session.delete(db_object) + + def create_policy_port_binding(context, policy_id, port_id): with context.session.begin(subtransactions=True): db_obj = models.QosPortPolicyBinding(policy_id=policy_id, port_id=port_id) context.session.add(db_obj) + + +def delete_policy_port_binding(context, policy_id, port_id): + with context.session.begin(subtransactions=True): + db_object = (db.model_query(context, models.QosPortPolicyBinding) + .filter_by(policy_id=policy_id, + port_id=port_id).one()) + context.session.delete(db_object) diff --git a/neutron/objects/qos/policy.py b/neutron/objects/qos/policy.py index e421023bd..83c481a02 100644 --- a/neutron/objects/qos/policy.py +++ b/neutron/objects/qos/policy.py @@ -73,9 +73,11 @@ class QosPolicy(base.NeutronObject): port_id=port_id) def detach_network(self, network_id): - # TODO(QoS): implement it, in the next life maybe - pass + qos_db_api.delete_policy_network_binding(self._context, + policy_id=self.id, + network_id=network_id) def detach_port(self, port_id): - # TODO(QoS): implement it, in the next life maybe - pass + qos_db_api.delete_policy_port_binding(self._context, + policy_id=self.id, + port_id=port_id) diff --git a/neutron/tests/unit/objects/qos/test_policy.py b/neutron/tests/unit/objects/qos/test_policy.py index ca26c5a02..9c208b994 100644 --- a/neutron/tests/unit/objects/qos/test_policy.py +++ b/neutron/tests/unit/objects/qos/test_policy.py @@ -31,53 +31,84 @@ class QosPolicyDbObjectTestCase(QosPolicyBaseTestCase, test_base.BaseDbObjectTestCase, testlib_api.SqlTestCase): - def test_attach_network_get_network_policy(self): - obj = policy.QosPolicy(self.context, **self.db_obj) - obj.create() - + def setUp(self): + super(QosPolicyDbObjectTestCase, self).setUp() + self._create_test_network() + self._create_test_port(self._network) + #TODO(QoS): move _create_test_policy here, as it's common + # to all. Now the base DB Object test case breaks + # that by introducing a duplicate object colliding + # on PK. + + def _create_test_policy(self): + policy_obj = policy.QosPolicy(self.context, **self.db_obj) + policy_obj.create() + return policy_obj + + def _create_test_network(self): # TODO(ihrachys): replace with network.create() once we get an object # implementation for networks - network = db_api.create_object(self.context, models_v2.Network, - {'name': 'test-network1'}) + self._network = db_api.create_object(self.context, models_v2.Network, + {'name': 'test-network1'}) + + def _create_test_port(self, network): + # TODO(ihrachys): replace with port.create() once we get an object + # implementation for ports + self._port = db_api.create_object(self.context, models_v2.Port, + {'name': 'test-port1', + 'network_id': network['id'], + 'mac_address': 'fake_mac', + 'admin_state_up': True, + 'status': 'ACTIVE', + 'device_id': 'fake_device', + 'device_owner': 'fake_owner'}) + + #TODO(QoS): give a thought on checking detach/attach for invalid values. + def test_attach_network_get_network_policy(self): + + obj = self._create_test_policy() policy_obj = policy.QosPolicy.get_network_policy(self.context, - network['id']) + self._network['id']) self.assertIsNone(policy_obj) # Now attach policy and repeat - obj.attach_network(network['id']) + obj.attach_network(self._network['id']) policy_obj = policy.QosPolicy.get_network_policy(self.context, - network['id']) + self._network['id']) self.assertEqual(obj, policy_obj) def test_attach_port_get_port_policy(self): - obj = policy.QosPolicy(self.context, **self.db_obj) - obj.create() - # TODO(ihrachys): replace with network.create() once we get an object - # implementation for networks - network = db_api.create_object(self.context, models_v2.Network, - {'name': 'test-network1'}) + obj = self._create_test_policy() - # TODO(ihrachys): replace with port.create() once we get an object - # implementation for ports - port = db_api.create_object(self.context, models_v2.Port, - {'name': 'test-port1', - 'network_id': network['id'], - 'mac_address': 'fake_mac', - 'admin_state_up': True, - 'status': 'ACTIVE', - 'device_id': 'fake_device', - 'device_owner': 'fake_owner'}) + policy_obj = policy.QosPolicy.get_network_policy(self.context, + self._network['id']) - policy_obj = policy.QosPolicy.get_port_policy(self.context, - port['id']) self.assertIsNone(policy_obj) # Now attach policy and repeat - obj.attach_port(port['id']) + obj.attach_port(self._port['id']) policy_obj = policy.QosPolicy.get_port_policy(self.context, - port['id']) + self._port['id']) self.assertEqual(obj, policy_obj) + + def test_detach_port(self): + obj = self._create_test_policy() + obj.attach_port(self._port['id']) + obj.detach_port(self._port['id']) + + policy_obj = policy.QosPolicy.get_port_policy(self.context, + self._port['id']) + self.assertIsNone(policy_obj) + + def test_detach_network(self): + obj = self._create_test_policy() + obj.attach_network(self._network['id']) + obj.detach_network(self._network['id']) + + policy_obj = policy.QosPolicy.get_network_policy(self.context, + self._network['id']) + self.assertIsNone(policy_obj) -- 2.45.2