openstack
/
neutron
Code Issues Proposed changes

tests: introduce update_obj_fields method in base objects test class 

This method can be used by subclasses to inject specific values into
randomly generated values for objects, for example, when we need to link
those automatic test objects to other objects through foreign keys.

Change-Id: Ic32ebf5ef055b25e4c765a9bf71ca4657d3b3d70
Partially-Implements: blueprint adopt-oslo-versioned-objects-for-db
This commit is contained in:
Ihar Hrachyshka 2016-11-29 00:20:44 +00:00
parent 6c46e55232
commit 79e62516af
11 changed files with 79 additions and 101 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
neutron/tests/unit/objects/port/extensions/test_allowedaddresspairs.py
View File

@ -10,8 +10,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import itertools
from neutron import context
from neutron.objects.port.extensions import allowedaddresspairs
from neutron.tests.unit.objects import test_base as obj_test_base
@ -34,5 +32,4 @@ class AllowedAddrPairsDbObjTestCase(obj_test_base.BaseDbObjectTestCase,
self.context = context.get_admin_context()
self._create_test_network()
self._create_test_port(self._network)
for obj in itertools.chain(self.db_objs, self.obj_fields, self.objs):
obj['port_id'] = self._port['id']
self.update_obj_fields({'port_id': self._port['id']})

5
neutron/tests/unit/objects/port/extensions/test_extra_dhcp_opt.py
View File

@ -10,8 +10,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import itertools
from neutron.objects.port.extensions import extra_dhcp_opt
from neutron.tests.unit.objects import test_base as obj_test_base
from neutron.tests.unit import testlib_api
@ -31,5 +29,4 @@ class ExtraDhcpOptDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
super(ExtraDhcpOptDbObjectTestCase, self).setUp()
self._create_test_network()
self._create_test_port(self._network)
for obj in itertools.chain(self.db_objs, self.obj_fields, self.objs):
obj['port_id'] = self._port['id']
self.update_obj_fields({'port_id': self._port['id']})

37
neutron/tests/unit/objects/test_base.py
View File

@ -479,7 +479,7 @@ class _BaseObjectTestCase(object):
valid_field = [f for f in self._test_class.fields
if f not in self._test_class.synthetic_fields][0]
self.valid_field_filter = {valid_field:
self.obj_fields[0][valid_field]}
self.obj_fields[-1][valid_field]}
self.obj_registry = self.useFixture(
fixture.VersionedObjectRegistryFixture())
self.obj_registry.register(FakeSmallNeutronObject)
@ -502,6 +502,41 @@ class _BaseObjectTestCase(object):
fields[field] = get_value(generator, ip_version)
return obj_cls.modify_fields_to_db(fields)
def update_obj_fields(self, values_dict,
db_objs=None, obj_fields=None, objs=None):
'''Update values for test objects with specific values.
The default behaviour is using random values for all fields of test
objects. Sometimes it's not practical, for example, when some fields,
often those referencing other objects, require non-random values (None
or UUIDs of valid objects). If that's the case, a test subclass may
call the method to override some field values for test objects.
Receives a single ``values_dict`` dict argument where keys are names of
test class fields, and values are either actual values for the keys, or
callables that will be used to generate different values for each test
object.
Note: if a value is a dict itself, the method will recursively update
corresponding embedded objects.
'''
for k, v in values_dict.items():
for db_obj, fields, obj in zip(
db_objs or self.db_objs,
obj_fields or self.obj_fields,
objs or self.objs):
val = v() if callable(v) else v
db_obj_key = obj.fields_need_translation.get(k, k)
if isinstance(val, collections.Mapping):
self.update_obj_fields(
val, db_obj[db_obj_key], fields[k], obj[k])
else:
db_obj[db_obj_key] = val
fields[k] = val
obj[k] = val
if k in self.valid_field_filter:
self.valid_field_filter[k] = val
@classmethod
def generate_object_keys(cls, obj_cls, field_names=None):
if field_names is None:

8
neutron/tests/unit/objects/test_floatingip.py
View File

@ -31,9 +31,5 @@ class FloatingIPDNSDbObjectTestcase(obj_test_base.BaseDbObjectTestCase,
def setUp(self):
super(FloatingIPDNSDbObjectTestcase, self).setUp()
for db_obj, obj_field, obj in zip(self.db_objs,
self.obj_fields, self.objs):
fip = self._create_test_fip()
db_obj['floatingip_id'] = fip['id']
obj_field['floatingip_id'] = fip['id']
obj['floatingip_id'] = fip['id']
self.update_obj_fields(
{'floatingip_id': lambda: self._create_test_fip().id})

23
neutron/tests/unit/objects/test_network.py
View File

@ -10,8 +10,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import itertools
from neutron.objects import base as obj_base
from neutron.objects import network
from neutron.objects.qos import policy
@ -30,12 +28,7 @@ class NetworkPortSecurityDbObjTestCase(obj_test_base.BaseDbObjectTestCase,
def setUp(self):
super(NetworkPortSecurityDbObjTestCase, self).setUp()
for db_obj, obj_field, obj in zip(
self.db_objs, self.obj_fields, self.objs):
network = self._create_network()
db_obj['network_id'] = network.id
obj_field['id'] = network.id
obj['id'] = network['id']
self.update_obj_fields({'id': lambda: self._create_network().id})
class NetworkSegmentIfaceObjTestCase(obj_test_base.BaseObjectIfaceTestCase):
@ -59,8 +52,7 @@ class NetworkSegmentDbObjTestCase(obj_test_base.BaseDbObjectTestCase,
def setUp(self):
super(NetworkSegmentDbObjTestCase, self).setUp()
network = self._create_network()
for obj in itertools.chain(self.db_objs, self.obj_fields, self.objs):
obj['network_id'] = network.id
self.update_obj_fields({'network_id': network.id})
class NetworkObjectIfaceTestCase(obj_test_base.BaseObjectIfaceTestCase):
@ -171,8 +163,7 @@ class SegmentHostMappingDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
super(SegmentHostMappingDbObjectTestCase, self).setUp()
self._create_test_network()
self._create_test_segment(network=self._network)
for obj in itertools.chain(self.db_objs, self.obj_fields, self.objs):
obj['segment_id'] = self._segment['id']
self.update_obj_fields({'segment_id': self._segment['id']})
class NetworkDNSDomainIfaceObjectTestcase(
@ -188,9 +179,5 @@ class NetworkDNSDomainDbObjectTestcase(obj_test_base.BaseDbObjectTestCase,
def setUp(self):
super(NetworkDNSDomainDbObjectTestcase, self).setUp()
for db_obj, obj_field, obj in zip(self.db_objs,
self.obj_fields, self.objs):
network = self._create_network()
db_obj['network_id'] = network['id']
obj_field['network_id'] = network['id']
obj['network_id'] = network['id']
self.update_obj_fields(
{'network_id': lambda: self._create_network().id})

43
neutron/tests/unit/objects/test_ports.py
View File

@ -10,8 +10,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import itertools
from oslo_utils import uuidutils
import testscenarios
@ -31,12 +29,8 @@ class BasePortBindingDbObjectTestCase(obj_test_base._BaseObjectTestCase,
def setUp(self):
super(BasePortBindingDbObjectTestCase, self).setUp()
self._create_test_network()
for db_obj, fields, obj in zip(
self.db_objs, self.obj_fields, self.objs):
port = self._create_port(network_id=self._network['id'])
db_obj['port_id'] = port.id
fields['port_id'] = port.id
obj.port_id = port.id
getter = lambda: self._create_port(network_id=self._network['id']).id
self.update_obj_fields({'port_id': getter})
class PortBindingIfaceObjTestCase(obj_test_base.BaseObjectIfaceTestCase):
@ -77,12 +71,8 @@ class PortBindingVifDetailsTestCase(testscenarios.WithScenarios,
def setUp(self):
super(PortBindingVifDetailsTestCase, self).setUp()
self._create_test_network()
for db_obj, fields, obj in zip(
self.db_objs, self.obj_fields, self.objs):
port = self._create_port(network_id=self._network['id'])
db_obj['port_id'] = port.id
fields['port_id'] = port.id
obj.port_id = port.id
getter = lambda: self._create_port(network_id=self._network['id']).id
self.update_obj_fields({'port_id': getter})
def _create_port(self, **port_attrs):
attrs = {'project_id': uuidutils.generate_uuid(),
@ -165,10 +155,9 @@ class IPAllocationDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
self._create_test_network()
self._create_test_subnet(self._network)
self._create_test_port(self._network)
for obj in itertools.chain(self.db_objs, self.obj_fields, self.objs):
obj['port_id'] = self._port.id
obj['network_id'] = self._network.id
obj['subnet_id'] = self._subnet.id
self.update_obj_fields({'port_id': self._port.id,
'network_id': self._network.id,
'subnet_id': self._subnet.id})
class PortDNSIfaceObjTestCase(obj_test_base.BaseObjectIfaceTestCase):
@ -184,12 +173,8 @@ class PortDNSDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
def setUp(self):
super(PortDNSDbObjectTestCase, self).setUp()
self._create_test_network()
for db_obj, fields, obj in zip(
self.db_objs, self.obj_fields, self.objs):
port_ = self._create_port(network_id=self._network['id'])
db_obj['port_id'] = port_.id
fields['port_id'] = port_.id
obj.port_id = port_.id
getter = lambda: self._create_port(network_id=self._network['id']).id
self.update_obj_fields({'port_id': getter})
class PortBindingLevelIfaceObjTestCase(
@ -239,12 +224,10 @@ class PortDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
super(PortDbObjectTestCase, self).setUp()
self._create_test_network()
self._create_test_subnet(self._network)
for obj in itertools.chain(self.db_objs, self.obj_fields, self.objs):
obj['network_id'] = self._network['id']
for obj in self.db_objs:
for ipalloc in obj['fixed_ips']:
ipalloc['subnet_id'] = self._subnet.id
ipalloc['network_id'] = self._network['id']
self.update_obj_fields(
{'network_id': self._network.id,
'fixed_ips': {'subnet_id': self._subnet.id,
'network_id': self._network['id']}})
def test_security_group_ids(self):
sg1 = self._create_test_security_group()

6
neutron/tests/unit/objects/test_provisioning_blocks.py
View File

@ -10,8 +10,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import itertools
from neutron.objects import provisioning_blocks
from neutron.tests.unit.objects import test_base as obj_test_base
from neutron.tests.unit import testlib_api
@ -31,5 +29,5 @@ class ProvisioningBlockDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
def setUp(self):
super(ProvisioningBlockDbObjectTestCase, self).setUp()
self._create_test_standard_attribute()
for obj in itertools.chain(self.db_objs, self.obj_fields, self.objs):
obj['standard_attr_id'] = self._standard_attribute['id']
self.update_obj_fields(
{'standard_attr_id': self._standard_attribute['id']})

19
neutron/tests/unit/objects/test_router.py
View File

@ -31,12 +31,12 @@ class RouterRouteDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
def setUp(self):
super(RouterRouteDbObjectTestCase, self).setUp()
for db_obj, obj_field, obj in zip(
self.db_objs, self.obj_fields, self.objs):
def getter():
self._create_test_router()
db_obj['router_id'] = self._router['id']
obj_field['router_id'] = self._router['id']
obj['router_id'] = self._router['id']
return self._router['id']
self.update_obj_fields({'router_id': getter})
class RouterExtraAttrsIfaceObjTestCase(obj_test_base.
@ -51,9 +51,8 @@ class RouterExtraAttrsDbObjTestCase(obj_test_base.BaseDbObjectTestCase,
def setUp(self):
super(RouterExtraAttrsDbObjTestCase, self).setUp()
for db_obj, obj_field, obj in zip(
self.db_objs, self.obj_fields, self.objs):
def getter():
self._create_test_router()
db_obj['router_id'] = self._router['id']
obj_field['router_id'] = self._router['id']
obj['router_id'] = self._router['id']
return self._router['id']
self.update_obj_fields({'router_id': getter})

10
neutron/tests/unit/objects/test_securitygroup.py
View File

@ -10,8 +10,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import itertools
from neutron.objects import securitygroup
from neutron.tests.unit.objects import test_base
from neutron.tests.unit import testlib_api
@ -117,8 +115,7 @@ class DefaultSecurityGroupDbObjTestCase(test_base.BaseDbObjectTestCase,
sg_db_obj)
self.sg_obj = securitygroup.SecurityGroup(self.context, **sg_fields)
self.sg_obj.create()
for obj in itertools.chain(self.db_objs, self.obj_fields, self.objs):
obj['security_group_id'] = self.sg_obj['id']
self.update_obj_fields({'security_group_id': self.sg_obj['id']})
class SecurityGroupRuleIfaceObjTestCase(test_base.BaseObjectIfaceTestCase):
@ -138,6 +135,5 @@ class SecurityGroupRuleDbObjTestCase(test_base.BaseDbObjectTestCase,
sg_db_obj)
self.sg_obj = securitygroup.SecurityGroup(self.context, **sg_fields)
self.sg_obj.create()
for obj in itertools.chain(self.db_objs, self.obj_fields, self.objs):
obj['security_group_id'] = self.sg_obj['id']
obj['remote_group_id'] = self.sg_obj['id']
self.update_obj_fields({'security_group_id': self.sg_obj['id'],
'remote_group_id': self.sg_obj['id']})

19
neutron/tests/unit/objects/test_subnet.py
View File

@ -10,8 +10,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import itertools
from oslo_utils import uuidutils
from neutron import context
@ -38,8 +36,7 @@ class IPAllocationPoolDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
super(IPAllocationPoolDbObjectTestCase, self).setUp()
self._create_test_network()
self._create_test_subnet(self._network)
for obj in itertools.chain(self.db_objs, self.obj_fields, self.objs):
obj['subnet_id'] = self._subnet['id']
self.update_obj_fields({'subnet_id': self._subnet['id']})
class DNSNameServerObjectIfaceTestCase(obj_test_base.BaseObjectIfaceTestCase):
@ -61,8 +58,7 @@ class DNSNameServerDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
super(DNSNameServerDbObjectTestCase, self).setUp()
self._create_test_network()
self._create_test_subnet(self._network)
for obj in itertools.chain(self.db_objs, self.obj_fields, self.objs):
obj['subnet_id'] = self._subnet['id']
self.update_obj_fields({'subnet_id': self._subnet['id']})
def _create_dnsnameservers(self):
for obj in self.obj_fields:
@ -106,8 +102,7 @@ class RouteDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
super(RouteDbObjectTestCase, self).setUp()
self._create_test_network()
self._create_test_subnet(self._network)
for obj in itertools.chain(self.db_objs, self.obj_fields, self.objs):
obj['subnet_id'] = self._subnet['id']
self.update_obj_fields({'subnet_id': self._subnet['id']})
class SubnetServiceTypeObjectIfaceTestCase(
@ -125,8 +120,7 @@ class SubnetServiceTypeDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
super(SubnetServiceTypeDbObjectTestCase, self).setUp()
self._create_test_network()
self._create_test_subnet(self._network)
for obj in itertools.chain(self.db_objs, self.obj_fields, self.objs):
obj['subnet_id'] = self._subnet['id']
self.update_obj_fields({'subnet_id': self._subnet['id']})
class SubnetObjectIfaceTestCase(obj_test_base.BaseObjectIfaceTestCase):
@ -148,9 +142,8 @@ class SubnetDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
super(SubnetDbObjectTestCase, self).setUp()
self._create_test_network()
self._create_test_segment(self._network)
for obj in itertools.chain(self.db_objs, self.obj_fields, self.objs):
obj['network_id'] = self._network['id']
obj['segment_id'] = self._segment['id']
self.update_obj_fields({'network_id': self._network['id'],
'segment_id': self._segment['id']})
def test_get_dns_nameservers_in_order(self):
obj = self._make_object(self.obj_fields[0])

5
neutron/tests/unit/objects/test_subnetpool.py
View File

@ -13,8 +13,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import itertools
from oslo_utils import uuidutils
from neutron.objects import subnetpool
@ -87,5 +85,4 @@ class SubnetPoolPrefixDbObjectTestCase(
def setUp(self):
super(SubnetPoolPrefixDbObjectTestCase, self).setUp()
self._create_test_subnetpool()
for obj in itertools.chain(self.db_objs, self.obj_fields, self.objs):
obj['subnetpool_id'] = self._pool.id
self.update_obj_fields({'subnetpool_id': self._pool.id})