Merge "Set standard_attr_id as property in DeclarativeObject."

This commit is contained in:
Jenkins 2016-12-21 15:04:53 +00:00 committed by Gerrit Code Review
commit 46594d1d1f
3 changed files with 28 additions and 5 deletions

View File

@ -259,6 +259,9 @@ class DeclarativeObject(abc.ABCMeta):
if (hasattr(cls, 'has_standard_attributes') and
cls.has_standard_attributes()):
setattr(cls, 'standard_attr_id',
property(lambda x: x.db_obj.standard_attr_id
if x.db_obj else None))
standardattributes.add_standard_attributes(cls)
# Instantiate extra filters per class
cls.extra_filter_names = set(cls.extra_filter_names)

View File

@ -202,13 +202,13 @@ class QosPolicyDbObjectTestCase(test_base.BaseDbObjectTestCase,
def test_attach_network_nonexistent_policy(self):
policy_obj = policy.QosPolicy(self.context, **self.db_objs[0])
policy_obj = self._make_object(self.obj_fields[0])
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_objs[0])
policy_obj = self._make_object(self.obj_fields[0])
self.assertRaises(n_exc.PortQosBindingNotFound,
policy_obj.attach_port, self._port['id'])
@ -289,12 +289,12 @@ class QosPolicyDbObjectTestCase(test_base.BaseDbObjectTestCase,
obj.detach_network, 'non-existent-port')
def test_detach_port_nonexistent_policy(self):
policy_obj = policy.QosPolicy(self.context, **self.db_objs[0])
policy_obj = self._make_object(self.obj_fields[0])
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_objs[0])
policy_obj = self._make_object(self.obj_fields[0])
self.assertRaises(n_exc.NetworkQosBindingNotFound,
policy_obj.detach_network, self._network['id'])
@ -332,7 +332,7 @@ class QosPolicyDbObjectTestCase(test_base.BaseDbObjectTestCase,
self.assertEqual(rule_dict, obj_dict['rules'][0])
def test_shared_default(self):
obj = self._test_class(self.context, **self.db_objs[0])
obj = self._make_object(self.obj_fields[0])
self.assertFalse(obj.shared)
def test_delete_not_allowed_if_policy_in_use_by_port(self):

View File

@ -1318,6 +1318,26 @@ class BaseDbObjectTestCase(_BaseObjectTestCase,
standard_attr.StandardAttribute,
attrs)
def test_get_standard_attr_id(self):
if not self._test_class.has_standard_attributes():
self.skipTest(
'No standard attributes found in test class %r'
% self._test_class)
obj = self._make_object(self.obj_fields[0])
obj.create()
model = self.context.session.query(obj.db_model).filter_by(
**obj._get_composite_keys()).one()
retrieved_obj = self._test_class.get_object(
self.context, **obj._get_composite_keys())
self.assertIsNotNone(retrieved_obj.standard_attr_id)
self.assertEqual(
model.standard_attr_id, retrieved_obj.standard_attr_id)
def _make_object(self, fields):
fields = get_non_synthetic_fields(self._test_class, fields)
return self._test_class(self.context,