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

diff --git a/neutron/db/qos/api.py b/neutron/db/qos/api.py
index 632c57e9efbb71edfdecd97f215fdfa740d277d2..40b8ab77b8e80395beb7fac87a7a1c6b6be331cd 100644 (file)
--- 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 e421023bdb541352c2178b388650622357b1a50c..83c481a02b1fd87d0cbe872544fb1c9f35366543 100644 (file)
--- 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 ca26c5a029fbc986dde255ca436f7ca805f06391..9c208b994959ed850425cb3525683833a9411b17 100644 (file)
--- 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)