objects.qos.policy: fixed get_*_policy and attach_* methods

Added sql unit tests for those methods.

Change-Id: I6e95aa6cb61d5cc36600394b2198587793da8a0e

diff --git a/neutron/db/api.py b/neutron/db/api.py
index 6de77700059999ec8849632b043b87f1a8226083..f3cb3a84a4d50eee7c544c6ce515aadf48a3bf24 100644 (file)
--- a/neutron/db/api.py
+++ b/neutron/db/api.py
@@ -93,7 +93,7 @@ class convert_db_exception_to_retry(object):
 # Common database operation implementations
 # TODO(QoS): consider handling multiple objects found, or no objects at all
 # TODO(QoS): consider changing the name and making it public, officially
-def _find_object(context, model, *kwargs):
+def _find_object(context, model, **kwargs):
     with context.session.begin(subtransactions=True):
         return (common_db_mixin.model_query(context, model)
                 .filter_by(**kwargs)

diff --git a/neutron/objects/qos/policy.py b/neutron/objects/qos/policy.py
index 21605a555acff5d843848d3bbd9ad82ce9c5ee54..1e34c6809e91cfc0b1ed3bdd6cf0b8003259d674 100644 (file)
--- a/neutron/objects/qos/policy.py
+++ b/neutron/objects/qos/policy.py
@@ -61,11 +61,13 @@ class QosPolicy(base.NeutronObject):
                                       port_id=port_id)
 
     def attach_network(self, network_id):
-        qos_db_api.create_policy_network_binding(policy_id=self.id,
+        qos_db_api.create_policy_network_binding(self._context,
+                                                 policy_id=self.id,
                                                  network_id=network_id)
 
     def attach_port(self, port_id):
-        qos_db_api.create_policy_port_binding(policy_id=self.id,
+        qos_db_api.create_policy_port_binding(self._context,
+                                              policy_id=self.id,
                                               port_id=port_id)
 
     def detach_network(self, network_id):

diff --git a/neutron/tests/unit/objects/qos/test_policy.py b/neutron/tests/unit/objects/qos/test_policy.py
index e88b7915a7d440cfb721dc6a1e4c120e2506f5e4..ca26c5a029fbc986dde255ca436f7ca805f06391 100644 (file)
--- a/neutron/tests/unit/objects/qos/test_policy.py
+++ b/neutron/tests/unit/objects/qos/test_policy.py
@@ -10,6 +10,8 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+from neutron.db import api as db_api
+from neutron.db import models_v2
 from neutron.objects.qos import policy
 from neutron.tests.unit.objects import test_base
 from neutron.tests.unit import testlib_api
@@ -28,4 +30,54 @@ class QosPolicyObjectTestCase(QosPolicyBaseTestCase,
 class QosPolicyDbObjectTestCase(QosPolicyBaseTestCase,
                                 test_base.BaseDbObjectTestCase,
                                 testlib_api.SqlTestCase):
-    pass
+
+    def test_attach_network_get_network_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'})
+
+        policy_obj = policy.QosPolicy.get_network_policy(self.context,
+                                                         network['id'])
+        self.assertIsNone(policy_obj)
+
+        # Now attach policy and repeat
+        obj.attach_network(network['id'])
+
+        policy_obj = policy.QosPolicy.get_network_policy(self.context,
+                                                         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'})
+
+        # 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_port_policy(self.context,
+                                                      port['id'])
+        self.assertIsNone(policy_obj)
+
+        # Now attach policy and repeat
+        obj.attach_port(port['id'])
+
+        policy_obj = policy.QosPolicy.get_port_policy(self.context,
+                                                      port['id'])
+        self.assertEqual(obj, policy_obj)