openstack
/
neutron
Code Issues Proposed changes

Refactoring _create_test methods for OVO UTs 

Most of the _create_test methods create a property and only uses
its id during their execution.  This change pretends to simplify
the returned data and reduce the number of lines used for the
creation of test data used during the execution of Unit Tests for
Oslo-Versioned Objects.

Change-Id: Icfc251a7d460246e91e260a4ac503d10c4031170
Partially-Implements: blueprint adopt-oslo-versioned-objects-for-db
This commit is contained in:
Victor Morales 2017-01-26 08:43:40 -06:00 committed by Ihar Hrachyshka
parent 008bc01ddf
commit 474bfe289a
19 changed files with 193 additions and 209 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
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.
from neutron_lib import context
from neutron.objects.port.extensions import allowedaddresspairs
from neutron.tests.unit.objects import test_base as obj_test_base
from neutron.tests.unit import testlib_api
@ -30,7 +28,5 @@ class AllowedAddrPairsDbObjTestCase(obj_test_base.BaseDbObjectTestCase,
def setUp(self):
super(AllowedAddrPairsDbObjTestCase, self).setUp()
self.context = context.get_admin_context()
self._create_test_network()
self._create_test_port(self._network)
self.update_obj_fields({'port_id': self._port['id']})
self.update_obj_fields(
{'port_id': lambda: self._create_test_port_id()})

4
neutron/tests/unit/objects/port/extensions/test_data_plane_status.py
View File

@ -29,6 +29,6 @@ class DataPlaneStatusDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
def setUp(self):
super(DataPlaneStatusDbObjectTestCase, self).setUp()
self._create_test_network()
getter = lambda: self._create_port(network_id=self._network['id']).id
net = self._create_test_network()
getter = lambda: self._create_test_port(network_id=net.id).id
self.update_obj_fields({'port_id': getter})

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

@ -27,6 +27,5 @@ class ExtraDhcpOptDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
def setUp(self):
super(ExtraDhcpOptDbObjectTestCase, self).setUp()
self._create_test_network()
self._create_test_port(self._network)
self.update_obj_fields({'port_id': self._port['id']})
self.update_obj_fields(
{'port_id': lambda: self._create_test_port_id()})

6
neutron/tests/unit/objects/port/extensions/test_port_security.py
View File

@ -29,7 +29,7 @@ class PortSecurityDbObjTestCase(obj_test_base.BaseDbObjectTestCase,
def setUp(self):
super(PortSecurityDbObjTestCase, self).setUp()
self._create_test_network()
network_id = self._create_test_network_id()
for obj in self.db_objs:
self._create_port(id=obj['port_id'],
network_id=self._network['id'])
self._create_test_port(
id=obj['port_id'], network_id=network_id)

34
neutron/tests/unit/objects/qos/test_policy.py
View File

@ -125,8 +125,8 @@ class QosPolicyDbObjectTestCase(test_base.BaseDbObjectTestCase,
def setUp(self):
super(QosPolicyDbObjectTestCase, self).setUp()
self._create_test_network()
self._create_test_port(self._network)
self._network_id = self._create_test_network_id()
self._port = self._create_test_port(network_id=self._network_id)
def _create_test_policy(self):
self.objs[0].create()
@ -152,14 +152,14 @@ class QosPolicyDbObjectTestCase(test_base.BaseDbObjectTestCase,
obj = self._create_test_policy()
policy_obj = policy.QosPolicy.get_network_policy(self.context,
self._network['id'])
self._network_id)
self.assertIsNone(policy_obj)
# Now attach policy and repeat
obj.attach_network(self._network['id'])
obj.attach_network(self._network_id)
policy_obj = policy.QosPolicy.get_network_policy(self.context,
self._network['id'])
self._network_id)
self.assertEqual(obj, policy_obj)
def test_attach_network_nonexistent_network(self):
@ -171,15 +171,15 @@ class QosPolicyDbObjectTestCase(test_base.BaseDbObjectTestCase,
def test_attach_network_get_policy_network(self):
obj = self._create_test_policy()
obj.attach_network(self._network['id'])
obj.attach_network(self._network_id)
networks = obj.get_bound_networks()
self.assertEqual(1, len(networks))
self.assertEqual(self._network['id'], networks[0])
self.assertEqual(self._network_id, networks[0])
def test_attach_and_get_multiple_policy_networks(self):
net1_id = self._network['id']
net1_id = self._network_id
net2 = net_obj.Network(self.context,
name='test-network2')
net2.create()
@ -204,7 +204,7 @@ class QosPolicyDbObjectTestCase(test_base.BaseDbObjectTestCase,
policy_obj = self._make_object(self.obj_fields[0])
self.assertRaises(n_exc.NetworkQosBindingNotFound,
policy_obj.attach_network, self._network['id'])
policy_obj.attach_network, self._network_id)
def test_attach_port_nonexistent_policy(self):
@ -217,7 +217,7 @@ class QosPolicyDbObjectTestCase(test_base.BaseDbObjectTestCase,
obj = self._create_test_policy()
policy_obj = policy.QosPolicy.get_network_policy(self.context,
self._network['id'])
self._network_id)
self.assertIsNone(policy_obj)
@ -234,7 +234,7 @@ class QosPolicyDbObjectTestCase(test_base.BaseDbObjectTestCase,
port2 = db_api.create_object(self.context, models_v2.Port,
{'tenant_id': 'fake_tenant_id',
'name': 'test-port2',
'network_id': self._network['id'],
'network_id': self._network_id,
'mac_address': 'fake_mac2',
'admin_state_up': True,
'status': 'ACTIVE',
@ -271,11 +271,11 @@ class QosPolicyDbObjectTestCase(test_base.BaseDbObjectTestCase,
def test_detach_network(self):
obj = self._create_test_policy()
obj.attach_network(self._network['id'])
obj.detach_network(self._network['id'])
obj.attach_network(self._network_id)
obj.detach_network(self._network_id)
policy_obj = policy.QosPolicy.get_network_policy(self.context,
self._network['id'])
self._network_id)
self.assertIsNone(policy_obj)
def test_detach_port_nonexistent_port(self):
@ -296,7 +296,7 @@ class QosPolicyDbObjectTestCase(test_base.BaseDbObjectTestCase,
def test_detach_network_nonexistent_policy(self):
policy_obj = self._make_object(self.obj_fields[0])
self.assertRaises(n_exc.NetworkQosBindingNotFound,
policy_obj.detach_network, self._network['id'])
policy_obj.detach_network, self._network_id)
def test_synthetic_rule_fields(self):
policy_obj, rule_obj = self._create_test_policy_with_rules(
@ -346,11 +346,11 @@ class QosPolicyDbObjectTestCase(test_base.BaseDbObjectTestCase,
def test_delete_not_allowed_if_policy_in_use_by_network(self):
obj = self._create_test_policy()
obj.attach_network(self._network['id'])
obj.attach_network(self._network_id)
self.assertRaises(n_exc.QosPolicyInUse, obj.delete)
obj.detach_network(self._network['id'])
obj.detach_network(self._network_id)
obj.delete()
def test_reload_rules_reloads_rules(self):

7
neutron/tests/unit/objects/test_auto_allocate.py
View File

@ -30,7 +30,6 @@ class AutoAllocateTopologyDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
def setUp(self):
super(AutoAllocateTopologyDbObjectTestCase, self).setUp()
self._create_test_network()
self._create_test_router()
self.update_obj_fields({'network_id': self._network['id'],
'router_id': self._router['id']})
self.update_obj_fields({
'network_id': lambda: self._create_test_network_id(),
'router_id': lambda: self._create_test_router_id()})

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

@ -1261,44 +1261,41 @@ class BaseDbObjectTestCase(_BaseObjectTestCase,
objclass.db_model(**objclass_fields)
]
def _create_test_network(self):
self._network = net_obj.Network(self.context,
name='test-network1')
self._network.create()
def _create_network(self):
name = "test-network-%s" % helpers.get_random_string(4)
_network = net_obj.Network(self.context,
name=name)
def _create_test_network(self, name='test-network1'):
_network = net_obj.Network(self.context, name=name)
_network.create()
return _network
def _create_external_network(self):
test_network = self._create_network()
ext_net = net_obj.ExternalNetwork(self.context,
network_id=test_network['id'])
ext_net.create()
return ext_net
def _create_test_network_id(self):
return self._create_test_network(
"test-network-%s" % helpers.get_random_string(4)).id
def _create_test_fip(self):
def _create_external_network_id(self):
test_network_id = self._create_test_network_id()
ext_net = net_obj.ExternalNetwork(self.context,
network_id=test_network_id)
ext_net.create()
return ext_net.network_id
def _create_test_fip_id(self):
fake_fip = '172.23.3.0'
ext_net = self._create_external_network()
test_port = self._create_port(
network_id=ext_net['network_id'])
ext_net_id = self._create_external_network_id()
# TODO(manjeets) replace this with fip ovo
# once it is implemented
return obj_db_api.create_object(
self.context,
l3_model.FloatingIP,
self.context, l3_model.FloatingIP,
{'floating_ip_address': fake_fip,
'floating_network_id': ext_net['network_id'],
'floating_port_id': test_port['id']})
'floating_network_id': ext_net_id,
'floating_port_id': self._create_test_port_id(
network_id=ext_net_id)}).id
def _create_test_subnet(self, network):
def _create_test_subnet_id(self, network_id=None):
if not network_id:
network_id = self._create_test_network_id()
test_subnet = {
'project_id': uuidutils.generate_uuid(),
'name': 'test-subnet1',
'network_id': network['id'],
'network_id': network_id,
'ip_version': 4,
'cidr': netaddr.IPNetwork('10.0.0.0/24'),
'gateway_ip': '10.0.0.1',
@ -1306,10 +1303,17 @@ class BaseDbObjectTestCase(_BaseObjectTestCase,
'ipv6_ra_mode': None,
'ipv6_address_mode': None
}
self._subnet = subnet.Subnet(self.context, **test_subnet)
self._subnet.create()
subnet_obj = subnet.Subnet(self.context, **test_subnet)
subnet_obj.create()
return subnet_obj.id
def _create_test_port_id(self, **port_attrs):
return self._create_test_port(**port_attrs)['id']
def _create_test_port(self, **port_attrs):
if 'network_id' not in port_attrs:
port_attrs['network_id'] = self._create_test_network_id()
def _create_port(self, **port_attrs):
if not hasattr(self, '_mac_address_generator'):
self._mac_address_generator = (
netaddr.EUI(":".join(["%02x" % i] * 6))
@ -1336,46 +1340,55 @@ class BaseDbObjectTestCase(_BaseObjectTestCase,
port.create()
return port
def _create_test_segment(self, network):
def _create_test_segment_id(self, network_id=None):
attr = self.get_random_object_fields(net_obj.NetworkSegment)
attr['network_id'] = network['id']
self._segment = net_obj.NetworkSegment(self.context, **attr)
self._segment.create()
attr['network_id'] = network_id or self._create_test_network_id()
segment = net_obj.NetworkSegment(self.context, **attr)
segment.create()
return segment.id
def _create_test_router(self):
def _create_test_router_id(self):
attrs = {
'name': 'test_router',
}
# TODO(sindhu): Replace with the router object once its ready
self._router = obj_db_api.create_object(self.context,
l3_model.Router,
attrs)
router = obj_db_api.create_object(
self.context, l3_model.Router, attrs)
return router['id']
def _create_test_security_group(self):
def _create_test_security_group_id(self):
sg_fields = self.get_random_object_fields(securitygroup.SecurityGroup)
self._securitygroup = securitygroup.SecurityGroup(self.context,
**sg_fields)
self._securitygroup.create()
return self._securitygroup
_securitygroup = securitygroup.SecurityGroup(
self.context, **sg_fields)
_securitygroup.create()
return _securitygroup.id
def _create_test_agent(self):
def _create_test_agent_id(self):
attrs = self.get_random_object_fields(obj_cls=agent.Agent)
self._agent = agent.Agent(self.context, **attrs)
self._agent.create()
_agent = agent.Agent(self.context, **attrs)
_agent.create()
return _agent['id']
def _create_test_port(self, network):
self._port = self._create_port(network_id=network['id'])
def _create_test_standard_attribute(self):
def _create_test_standard_attribute_id(self):
attrs = {
'id': tools.get_random_integer(),
'resource_type': helpers.get_random_string(4),
'revision_number': tools.get_random_integer()
}
self._standard_attribute = obj_db_api.create_object(
self.context,
standard_attr.StandardAttribute,
attrs)
return obj_db_api.create_object(
self.context, standard_attr.StandardAttribute, attrs)['id']
def _create_test_flavor_id(self):
attrs = self.get_random_object_fields(obj_cls=flavor.Flavor)
flavor_obj = flavor.Flavor(self.context, **attrs)
flavor_obj.create()
return flavor_obj.id
def _create_test_service_profile_id(self):
attrs = self.get_random_object_fields(obj_cls=flavor.ServiceProfile)
service_profile_obj = flavor.ServiceProfile(self.context, **attrs)
service_profile_obj.create()
return service_profile_obj.id
def test_get_standard_attr_id(self):
@ -1397,18 +1410,6 @@ class BaseDbObjectTestCase(_BaseObjectTestCase,
self.assertEqual(
model.standard_attr_id, retrieved_obj.standard_attr_id)
def _create_test_flavor(self):
attrs = self.get_random_object_fields(obj_cls=flavor.Flavor)
self._flavor = flavor.Flavor(self.context, **attrs)
self._flavor.create()
return self._flavor
def _create_test_service_profile(self):
attrs = self.get_random_object_fields(obj_cls=flavor.ServiceProfile)
self._service_profile = flavor.ServiceProfile(self.context, **attrs)
self._service_profile.create()
return self._service_profile
def _make_object(self, fields):
fields = get_non_synthetic_fields(self._test_class, fields)
return self._test_class(self.context,

7
neutron/tests/unit/objects/test_flavor.py
View File

@ -32,10 +32,9 @@ class FlavorServiceProfileBindingDbObjectTestCase(
def setUp(self):
super(FlavorServiceProfileBindingDbObjectTestCase, self).setUp()
self.update_obj_fields(
{'flavor_id': lambda: self._create_test_flavor().id})
self.update_obj_fields(
{'service_profile_id': lambda: self._create_test_service_profile(
).id})
{'flavor_id': lambda: self._create_test_flavor_id(),
'service_profile_id':
lambda: self._create_test_service_profile_id()})
class ServiceProfileIfaceObjectTestCase(obj_test_base.BaseObjectIfaceTestCase):

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

@ -32,4 +32,4 @@ class FloatingIPDNSDbObjectTestcase(obj_test_base.BaseDbObjectTestCase,
def setUp(self):
super(FloatingIPDNSDbObjectTestcase, self).setUp()
self.update_obj_fields(
{'floatingip_id': lambda: self._create_test_fip().id})
{'floatingip_id': lambda: self._create_test_fip_id()})

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

@ -27,9 +27,8 @@ class IpamSubnetDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
def setUp(self):
super(IpamSubnetDbObjectTestCase, self).setUp()
self._create_test_network()
self._create_test_subnet(self._network)
self.update_obj_fields({'neutron_subnet_id': self._subnet['id']})
self.update_obj_fields(
{'neutron_subnet_id': lambda: self._create_test_subnet_id()})
class IpamAllocationPoolObjectIfaceTestCase(

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

@ -29,14 +29,10 @@ class RouterL3AgentBindingDbObjTestCase(test_base.BaseDbObjectTestCase,
def setUp(self):
super(RouterL3AgentBindingDbObjTestCase, self).setUp()
self._create_test_router()
def getter():
self._create_test_agent()
return self._agent['id']
router_id = self._create_test_router_id()
index = iter(range(1, len(self.objs) + 1))
self.update_obj_fields(
{'router_id': self._router.id,
{'router_id': router_id,
'binding_index': lambda: next(index),
'l3_agent_id': getter})
'l3_agent_id': lambda: self._create_test_agent_id()})

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

@ -28,7 +28,7 @@ class NetworkPortSecurityDbObjTestCase(obj_test_base.BaseDbObjectTestCase,
def setUp(self):
super(NetworkPortSecurityDbObjTestCase, self).setUp()
self.update_obj_fields({'id': lambda: self._create_network().id})
self.update_obj_fields({'id': lambda: self._create_test_network_id()})
class NetworkSegmentIfaceObjTestCase(obj_test_base.BaseObjectIfaceTestCase):
@ -51,8 +51,8 @@ class NetworkSegmentDbObjTestCase(obj_test_base.BaseDbObjectTestCase,
def setUp(self):
super(NetworkSegmentDbObjTestCase, self).setUp()
network = self._create_network()
self.update_obj_fields({'network_id': network.id})
self.update_obj_fields(
{'network_id': lambda: self._create_test_network_id()})
def test_hosts(self):
hosts = ['host1', 'host2']
@ -182,9 +182,8 @@ class SegmentHostMappingDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
def setUp(self):
super(SegmentHostMappingDbObjectTestCase, self).setUp()
self._create_test_network()
self._create_test_segment(network=self._network)
self.update_obj_fields({'segment_id': self._segment['id']})
self.update_obj_fields(
{'segment_id': lambda: self._create_test_segment_id()})
class NetworkDNSDomainIfaceObjectTestcase(
@ -201,7 +200,7 @@ class NetworkDNSDomainDbObjectTestcase(obj_test_base.BaseDbObjectTestCase,
def setUp(self):
super(NetworkDNSDomainDbObjectTestcase, self).setUp()
self.update_obj_fields(
{'network_id': lambda: self._create_network().id})
{'network_id': lambda: self._create_test_network_id()})
class ExternalNetworkIfaceObjectTestCase(
@ -218,4 +217,4 @@ class ExternalNetworkDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
def setUp(self):
super(ExternalNetworkDbObjectTestCase, self).setUp()
self.update_obj_fields(
{'network_id': lambda: self._create_network().id})
{'network_id': lambda: self._create_test_network_id()})

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

@ -26,9 +26,8 @@ class BasePortBindingDbObjectTestCase(obj_test_base._BaseObjectTestCase,
testlib_api.SqlTestCase):
def setUp(self):
super(BasePortBindingDbObjectTestCase, self).setUp()
self._create_test_network()
getter = lambda: self._create_port(network_id=self._network['id']).id
self.update_obj_fields({'port_id': getter})
self.update_obj_fields(
{'port_id': lambda: self._create_test_port_id()})
class PortBindingIfaceObjTestCase(obj_test_base.BaseObjectIfaceTestCase):
@ -153,12 +152,11 @@ class IPAllocationDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
def setUp(self):
super(IPAllocationDbObjectTestCase, self).setUp()
self._create_test_network()
self._create_test_subnet(self._network)
self._create_test_port(self._network)
self.update_obj_fields({'port_id': self._port.id,
'network_id': self._network.id,
'subnet_id': self._subnet.id})
network_id = self._create_test_network_id()
port_id = self._create_test_port_id(network_id=network_id)
self.update_obj_fields(
{'port_id': port_id, 'network_id': network_id,
'subnet_id': lambda: self._create_test_subnet_id(network_id)})
class PortDNSIfaceObjTestCase(obj_test_base.BaseObjectIfaceTestCase):
@ -173,9 +171,8 @@ class PortDNSDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
def setUp(self):
super(PortDNSDbObjectTestCase, self).setUp()
self._create_test_network()
getter = lambda: self._create_port(network_id=self._network['id']).id
self.update_obj_fields({'port_id': getter})
self.update_obj_fields(
{'port_id': lambda: self._create_test_port_id()})
class PortBindingLevelIfaceObjTestCase(
@ -219,17 +216,16 @@ class PortDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
def setUp(self):
super(PortDbObjectTestCase, self).setUp()
self._create_test_network()
self._create_test_subnet(self._network)
network_id = self._create_test_network_id()
subnet_id = self._create_test_subnet_id(network_id)
self.update_obj_fields(
{'network_id': self._network.id,
'fixed_ips': {'subnet_id': self._subnet.id,
'network_id': self._network['id']}})
{'network_id': network_id,
'fixed_ips': {'subnet_id': subnet_id, 'network_id': network_id}})
def test_security_group_ids(self):
sg1 = self._create_test_security_group()
sg2 = self._create_test_security_group()
groups = {sg1.id, sg2.id}
sg1_id = self._create_test_security_group_id()
sg2_id = self._create_test_security_group_id()
groups = {sg1_id, sg2_id}
obj = self._make_object(self.obj_fields[0])
obj.security_group_ids = groups
obj.create()
@ -237,12 +233,12 @@ class PortDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
obj = ports.Port.get_object(self.context, id=obj.id)
self.assertEqual(groups, obj.security_group_ids)
sg3 = self._create_test_security_group()
obj.security_group_ids = {sg3.id}
sg3_id = self._create_test_security_group_id()
obj.security_group_ids = {sg3_id}
obj.update()
obj = ports.Port.get_object(self.context, id=obj.id)
self.assertEqual({sg3.id}, obj.security_group_ids)
self.assertEqual({sg3_id}, obj.security_group_ids)
obj.security_group_ids = set()
obj.update()
@ -254,17 +250,17 @@ class PortDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
obj = self._make_object(self.obj_fields[0])
obj.create()
sg = self._create_test_security_group()
obj._attach_security_group(sg.id)
sg_id = self._create_test_security_group_id()
obj._attach_security_group(sg_id)
obj = ports.Port.get_object(self.context, id=obj.id)
self.assertIn(sg.id, obj.security_group_ids)
self.assertIn(sg_id, obj.security_group_ids)
sg2 = self._create_test_security_group()
obj._attach_security_group(sg2.id)
sg2_id = self._create_test_security_group_id()
obj._attach_security_group(sg2_id)
obj = ports.Port.get_object(self.context, id=obj.id)
self.assertIn(sg2.id, obj.security_group_ids)
self.assertIn(sg2_id, obj.security_group_ids)
def test_qos_policy_id(self):
policy_obj = policy.QosPolicy(self.context)
@ -315,7 +311,7 @@ class PortDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
'Port object loads segment info without relationships')
def test_v1_1_to_v1_0_drops_data_plane_status(self):
port_new = self._create_port(network_id=self._network['id'])
port_new = self._create_test_port()
port_v1_0 = port_new.obj_to_primitive(target_version='1.0')
self.assertNotIn('data_plane_status',
port_v1_0['versioned_object.data'])

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

@ -28,6 +28,8 @@ class ProvisioningBlockDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
def setUp(self):
super(ProvisioningBlockDbObjectTestCase, self).setUp()
self._create_test_standard_attribute()
self.update_obj_fields(
{'standard_attr_id': self._standard_attribute['id']})
{
'standard_attr_id':
lambda: self._create_test_standard_attribute_id()
})

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

@ -31,12 +31,8 @@ class RouterRouteDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
def setUp(self):
super(RouterRouteDbObjectTestCase, self).setUp()
def getter():
self._create_test_router()
return self._router['id']
self.update_obj_fields({'router_id': getter})
self.update_obj_fields(
{'router_id': lambda: self._create_test_router_id()})
class RouterExtraAttrsIfaceObjTestCase(obj_test_base.
@ -50,9 +46,5 @@ class RouterExtraAttrsDbObjTestCase(obj_test_base.BaseDbObjectTestCase,
def setUp(self):
super(RouterExtraAttrsDbObjTestCase, self).setUp()
def getter():
self._create_test_router()
return self._router['id']
self.update_obj_fields({'router_id': getter})
self.update_obj_fields(
{'router_id': lambda: self._create_test_router_id()})

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

@ -110,8 +110,11 @@ class DefaultSecurityGroupDbObjTestCase(test_base.BaseDbObjectTestCase,
def setUp(self):
super(DefaultSecurityGroupDbObjTestCase, self).setUp()
self._create_test_security_group()
self.update_obj_fields({'security_group_id': self._securitygroup.id})
self.update_obj_fields(
{
'security_group_id':
lambda: self._create_test_security_group_id()
})
class SecurityGroupRuleIfaceObjTestCase(test_base.BaseObjectIfaceTestCase):
@ -126,6 +129,10 @@ class SecurityGroupRuleDbObjTestCase(test_base.BaseDbObjectTestCase,
def setUp(self):
super(SecurityGroupRuleDbObjTestCase, self).setUp()
self._create_test_security_group()
self.update_obj_fields({'security_group_id': self._securitygroup.id,
'remote_group_id': self._securitygroup.id})
self.update_obj_fields(
{
'security_group_id':
lambda: self._create_test_security_group_id(),
'remote_group_id':
lambda: self._create_test_security_group_id()
})

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

@ -34,9 +34,8 @@ class IPAllocationPoolDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
def setUp(self):
super(IPAllocationPoolDbObjectTestCase, self).setUp()
self._create_test_network()
self._create_test_subnet(self._network)
self.update_obj_fields({'subnet_id': self._subnet['id']})
self.update_obj_fields(
{'subnet_id': lambda: self._create_test_subnet_id()})
class DNSNameServerObjectIfaceTestCase(obj_test_base.BaseObjectIfaceTestCase):
@ -56,9 +55,8 @@ class DNSNameServerDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
def setUp(self):
super(DNSNameServerDbObjectTestCase, self).setUp()
self._create_test_network()
self._create_test_subnet(self._network)
self.update_obj_fields({'subnet_id': self._subnet['id']})
self._subnet_id = self._create_test_subnet_id()
self.update_obj_fields({'subnet_id': self._subnet_id})
def _create_dnsnameservers(self):
for obj in self.obj_fields:
@ -75,7 +73,7 @@ class DNSNameServerDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
self._create_dnsnameservers()
pager = obj_base.Pager(sorts=[('order', False)])
objs = self._test_class.get_objects(self.context, _pager=pager,
subnet_id=self._subnet.id)
subnet_id=self._subnet_id)
fields_sorted = sorted([obj['order'] for obj in self.obj_fields],
reverse=True)
self.assertEqual(fields_sorted, [obj.order for obj in objs])
@ -100,9 +98,8 @@ class RouteDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
def setUp(self):
super(RouteDbObjectTestCase, self).setUp()
self._create_test_network()
self._create_test_subnet(self._network)
self.update_obj_fields({'subnet_id': self._subnet['id']})
self.update_obj_fields(
{'subnet_id': lambda: self._create_test_subnet_id()})
class SubnetServiceTypeObjectIfaceTestCase(
@ -118,9 +115,8 @@ class SubnetServiceTypeDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
def setUp(self):
super(SubnetServiceTypeDbObjectTestCase, self).setUp()
self._create_test_network()
self._create_test_subnet(self._network)
self.update_obj_fields({'subnet_id': self._subnet['id']})
self.update_obj_fields(
{'subnet_id': lambda: self._create_test_subnet_id()})
class SubnetObjectIfaceTestCase(obj_test_base.BaseObjectIfaceTestCase):
@ -140,10 +136,10 @@ class SubnetDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
def setUp(self):
super(SubnetDbObjectTestCase, self).setUp()
self._create_test_network()
self._create_test_segment(self._network)
self.update_obj_fields({'network_id': self._network['id'],
'segment_id': self._segment['id']})
network_id = self._create_test_network_id()
self.update_obj_fields(
{'network_id': network_id,
'segment_id': lambda: self._create_test_segment_id(network_id)})
def test_get_dns_nameservers_in_order(self):
obj = self._make_object(self.obj_fields[0])
@ -170,7 +166,7 @@ class SubnetDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
attrs)
def test_get_subnet_shared_true(self):
network = self._create_network()
network = self._create_test_network()
self._create_shared_network_rbac_entry(network)
subnet_data = dict(self.obj_fields[0])
subnet_data['network_id'] = network['id']
@ -189,7 +185,7 @@ class SubnetDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
self.assertTrue(new.shared)
def test_filter_by_shared(self):
network = self._create_network()
network = self._create_test_network()
self._create_shared_network_rbac_entry(network)
subnet_data = dict(self.obj_fields[0])
@ -202,7 +198,7 @@ class SubnetDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
self.assertEqual(obj, result[0])
def test_get_shared_subnet_with_another_tenant(self):
network_shared = self._create_network()
network_shared = self._create_test_network()
self._create_shared_network_rbac_entry(network_shared)
subnet_data = dict(self.obj_fields[0])

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

@ -27,6 +27,8 @@ class TagDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
def setUp(self):
super(TagDbObjectTestCase, self).setUp()
self._create_test_standard_attribute()
self.update_obj_fields(
{'standard_attr_id': self._standard_attribute['id']})
{
'standard_attr_id':
lambda: self._create_test_standard_attribute_id()
})

33
neutron/tests/unit/objects/test_trunk.py
View File

@ -46,15 +46,14 @@ class SubPortDbObjectTestCase(test_base.BaseDbObjectTestCase,
def setUp(self):
super(SubPortDbObjectTestCase, self).setUp()
self._create_test_network()
self._network_id = self._create_test_network_id()
for obj in self.obj_fields:
self._create_port(id=obj['port_id'],
network_id=self._network['id'])
self._create_test_port(
id=obj['port_id'], network_id=self._network_id)
self._create_trunk(trunk_id=obj['trunk_id'])
def _create_trunk(self, trunk_id):
port_id = uuidutils.generate_uuid()
self._create_port(id=port_id, network_id=self._network['id'])
port_id = self._create_test_port_id(network_id=self._network_id)
trunk = t_obj.Trunk(self.context, id=trunk_id, port_id=port_id)
trunk.create()
@ -86,14 +85,14 @@ class TrunkDbObjectTestCase(test_base.BaseDbObjectTestCase,
def setUp(self):
super(TrunkDbObjectTestCase, self).setUp()
self._create_test_network()
self._network_id = self._create_test_network_id()
sub_ports = []
for obj in self.db_objs:
sub_ports.extend(obj['sub_ports'])
for obj in itertools.chain(self.obj_fields, sub_ports):
self._create_port(id=obj['port_id'],
network_id=self._network['id'])
self._create_test_port(
id=obj['port_id'], network_id=self._network_id)
def test_create_port_not_found(self):
obj = self.obj_fields[0]
@ -107,12 +106,14 @@ class TrunkDbObjectTestCase(test_base.BaseDbObjectTestCase,
sub_ports = []
for vid in vids:
port = self._create_port(network_id=self._network['id'])
sub_ports.append(t_obj.SubPort(self.context, port_id=port['id'],
segmentation_type='vlan',
segmentation_id=vid))
trunk = t_obj.Trunk(self.context, port_id=port_id,
sub_ports=sub_ports, project_id=project_id)
vid_port_id = self._create_test_port_id(
network_id=self._network_id)
sub_ports.append(t_obj.SubPort(
self.context, port_id=vid_port_id, segmentation_type='vlan',
segmentation_id=vid))
trunk = t_obj.Trunk(
self.context, port_id=port_id, sub_ports=sub_ports,
project_id=project_id)
trunk.create()
self.assertEqual(sub_ports, trunk.sub_ports)
return trunk
@ -138,8 +139,8 @@ class TrunkDbObjectTestCase(test_base.BaseDbObjectTestCase,
trunk1 = self._test_create_trunk_with_subports(port_id1, trunk1_vids)
port_id2 = uuidutils.generate_uuid()
self._create_port(id=port_id2,
network_id=self._network['id'])
self._create_test_port(
id=port_id2, network_id=self._network_id)
self._test_create_trunk_with_subports(port_id2, trunk2_vids)
listed_trunk1 = t_obj.Trunk.get_object(