Merge "policy: made attach_* and detach_* methods more robust" into feature/qos

neutron/common/exceptions.py

@@ -94,6 +94,16 @@ class PortNotFoundOnNetwork(NotFound):
                 "on network %(net_id)s")
 
 
+class PortQosBindingNotFound(NotFound):
+    message = _("QoS binding for port %(port_id)s and policy %(policy_id)s "
+                "could not be found")
+
+
+class NetworkQosBindingNotFound(NotFound):
+    message = _("QoS binding for network %(net_id)s and policy %(policy_id)s "
+                "could not be found")
+
+
 class PolicyFileNotFound(NotFound):
     message = _("Policy configuration policy.json could not be found")
 

neutron/db/qos/api.py

@@ -10,35 +10,56 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+from oslo_db import exception as oslo_db_exception
+from sqlalchemy.orm import exc as orm_exc
+
+from neutron.common import exceptions as n_exc
 from neutron.db import common_db_mixin as db
 from neutron.db.qos import models
 
 
 def create_policy_network_binding(context, policy_id, network_id):
-    with context.session.begin(subtransactions=True):
-        db_obj = models.QosNetworkPolicyBinding(policy_id=policy_id,
-                                                network_id=network_id)
-        context.session.add(db_obj)
+    try:
+        with context.session.begin(subtransactions=True):
+            db_obj = models.QosNetworkPolicyBinding(policy_id=policy_id,
+                                                    network_id=network_id)
+            context.session.add(db_obj)
+    except oslo_db_exception.DBReferenceError:
+        raise n_exc.NetworkQosBindingNotFound(net_id=network_id,
+                                              policy_id=policy_id)
 
 
 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)
+    try:
+        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)
+    except orm_exc.NoResultFound:
+        raise n_exc.NetworkQosBindingNotFound(net_id=network_id,
+                                              policy_id=policy_id)
 
 
 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)
+    try:
+        with context.session.begin(subtransactions=True):
+            db_obj = models.QosPortPolicyBinding(policy_id=policy_id,
+                                                 port_id=port_id)
+            context.session.add(db_obj)
+    except oslo_db_exception.DBReferenceError:
+        raise n_exc.PortQosBindingNotFound(port_id=port_id,
+                                           policy_id=policy_id)
 
 
 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)
+    try:
+        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)
+    except orm_exc.NoResultFound:
+        raise n_exc.PortQosBindingNotFound(port_id=port_id,
+                                           policy_id=policy_id)

neutron/objects/qos/policy.py

@@ -103,7 +103,6 @@ class QosPolicy(base.NeutronObject):
     def _get_object_policy(cls, context, model, **kwargs):
         with db_api.autonested_transaction(context.session):
             binding_db_obj = db_api.get_object(context, model, **kwargs)
-            # TODO(QoS): rethink handling missing binding case
             if binding_db_obj:
                 return cls.get_by_id(context, binding_db_obj['policy_id'])
 

neutron/tests/unit/objects/qos/test_policy.py

@@ -12,6 +12,7 @@
 
 import mock
 
+from neutron.common import exceptions as n_exc
 from neutron.db import api as db_api
 from neutron.db import models_v2
 from neutron.objects.qos import policy
@@ -78,10 +79,6 @@ class QosPolicyDbObjectTestCase(test_base.BaseDbObjectTestCase,
         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)
@@ -135,6 +132,30 @@ class QosPolicyDbObjectTestCase(test_base.BaseDbObjectTestCase,
                                                          self._network['id'])
         self.assertEqual(obj, policy_obj)
 
+    def test_attach_network_nonexistent_network(self):
+
+        obj = self._create_test_policy()
+        self.assertRaises(n_exc.NetworkQosBindingNotFound,
+                          obj.attach_network, 'non-existent-network')
+
+    def test_attach_port_nonexistent_port(self):
+
+        obj = self._create_test_policy()
+        self.assertRaises(n_exc.PortQosBindingNotFound,
+                          obj.attach_port, 'non-existent-port')
+
+    def test_attach_network_nonexistent_policy(self):
+
+        policy_obj = policy.QosPolicy(self.context, **self.db_obj)
+        self.assertRaises(n_exc.NetworkQosBindingNotFound,
+                          policy_obj.attach_network, self._network['id'])
+
+    def test_attach_port_nonexistent_policy(self):
+
+        policy_obj = policy.QosPolicy(self.context, **self.db_obj)
+        self.assertRaises(n_exc.PortQosBindingNotFound,
+                          policy_obj.attach_port, self._port['id'])
+
     def test_attach_port_get_port_policy(self):
 
         obj = self._create_test_policy()
@@ -169,6 +190,26 @@ class QosPolicyDbObjectTestCase(test_base.BaseDbObjectTestCase,
                                                          self._network['id'])
         self.assertIsNone(policy_obj)
 
+    def test_detach_port_nonexistent_port(self):
+        obj = self._create_test_policy()
+        self.assertRaises(n_exc.PortQosBindingNotFound,
+                          obj.detach_port, 'non-existent-port')
+
+    def test_detach_network_nonexistent_network(self):
+        obj = self._create_test_policy()
+        self.assertRaises(n_exc.NetworkQosBindingNotFound,
+                          obj.detach_network, 'non-existent-port')
+
+    def test_detach_port_nonexistent_policy(self):
+        policy_obj = policy.QosPolicy(self.context, **self.db_obj)
+        self.assertRaises(n_exc.PortQosBindingNotFound,
+                          policy_obj.detach_port, self._port['id'])
+
+    def test_detach_network_nonexistent_policy(self):
+        policy_obj = policy.QosPolicy(self.context, **self.db_obj)
+        self.assertRaises(n_exc.NetworkQosBindingNotFound,
+                          policy_obj.detach_network, self._network['id'])
+
     def test_synthetic_rule_fields(self):
         policy_obj, rule_obj = self._create_test_policy_with_rule()
         policy_obj = policy.QosPolicy.get_by_id(self.context, policy_obj.id)