objects: avoid deepcopying models in test_db_obj

SQLAlchemy may asynchronously push models out of session cache in which
case we may receive DetachedInstanceError.

In the test case, instead of deepcopying models to compare, compare
each modified attribute independently.

This change also includes conversion from InstrumentedLists to regular
lists when converting model attributes to object fields. The fact that
we were returning InstrumentedLists was always an oversight but it
revealed itself after the modification of the test case that is the
core of this patch.

When converting object fields to db, convert Port's distributed_binding
None value to a empty list to reflect that the relationship of the Port
database model is a list. It was not an issue before the patch because
we were not comparing model attribute for equality but for in-equality
before, and so None was always != [].

Finally, this patch moves a bunch of TODOs to better reflect where they
belong to.

Closes-Bug: #1770452
Change-Id: I42cdf540129bd4470ec1a59345db9845a6198328
(cherry picked from commit 1a8a15f630)
changes/16/650516/3
Ihar Hrachyshka 4 years ago committed by Doug Wiegley
parent b3a12e15f4
commit 8915febaaf
  1. 5
      neutron/objects/base.py
  2. 17
      neutron/objects/ports.py
  3. 8
      neutron/tests/unit/objects/test_base.py

@ -27,6 +27,7 @@ from oslo_versionedobjects import base as obj_base
from oslo_versionedobjects import exception as obj_exception
from oslo_versionedobjects import fields as obj_fields
import six
from sqlalchemy import orm
from neutron._i18n import _
from neutron.db import api as db_api
@ -496,6 +497,10 @@ class NeutronDbObject(NeutronObject):
for field, field_db in cls.fields_need_translation.items():
if field_db in result:
result[field] = result.pop(field_db)
for k, v in result.items():
# don't allow sqlalchemy lists to propagate outside
if isinstance(v, orm.collections.InstrumentedList):
result[k] = list(v)
return result
@classmethod

@ -395,22 +395,31 @@ class Port(base.NeutronDbObject):
return super(Port, cls).get_objects(context, _pager, validate_filters,
**kwargs)
# TODO(rossella_s): get rid of it once we switch the db model to using
# custom types.
@classmethod
def modify_fields_to_db(cls, fields):
result = super(Port, cls).modify_fields_to_db(fields)
# TODO(rossella_s): get rid of it once we switch the db model to using
# custom types.
if 'mac_address' in result:
result['mac_address'] = cls.filter_to_str(result['mac_address'])
# convert None to []
if 'distributed_port_binding' in result:
result['distributed_port_binding'] = (
result['distributed_port_binding'] or []
)
return result
# TODO(rossella_s): get rid of it once we switch the db model to using
# custom types.
@classmethod
def modify_fields_from_db(cls, db_obj):
fields = super(Port, cls).modify_fields_from_db(db_obj)
# TODO(rossella_s): get rid of it once we switch the db model to using
# custom types.
if 'mac_address' in fields:
fields['mac_address'] = utils.AuthenticEUI(fields['mac_address'])
distributed_port_binding = fields.get('distributed_binding')
if distributed_port_binding:
fields['distributed_binding'] = fields['distributed_binding'][0]

@ -1977,12 +1977,16 @@ class BaseDbObjectTestCase(_BaseObjectTestCase,
fields_to_update = self.get_updatable_fields(self.obj_fields[1])
if fields_to_update:
old_model = copy.deepcopy(obj.db_obj)
old_fields = {}
for key, val in fields_to_update.items():
db_model_attr = (
obj.fields_need_translation.get(key, key))
old_fields[db_model_attr] = obj.db_obj[db_model_attr]
setattr(obj, key, val)
obj.update()
self.assertIsNotNone(obj.db_obj)
self.assertNotEqual(old_model, obj.db_obj)
for k, v in obj.modify_fields_to_db(fields_to_update).items():
self.assertEqual(v, obj.db_obj[k], '%s attribute differs' % k)
obj.delete()
self.assertIsNone(obj.db_obj)

Loading…
Cancel
Save