openstack/neutron-fwaas
Code Issues Proposed changes

UT: do not hide an original error in test resource ctxtmgr

Browse Source 
In unit tests, resource contextmanagers such as network(), subnet()
try to delete themselves after returning from yield even if an
exception occurs. However when an exception occurs, there is a case
where deletion fails. In this case original exception will be hidden
and it makes difficult to debug test failures.

Before each test starts, resources like database entries will be
recreated, so there is no need to try to delete resources even
when an exception occurs. This commit removes try-finally clause
from resource contextmanagers to make original errors visible.

Closes-Bug: #1295887
Change-Id: Ia844d2aa2c9fc036e643068c5284f64798963ee3
This commit is contained in:
Akihiro Motoki 2014-03-22 08:17:20 +09:00 committed by Gerrit Code Review
parent 721830c541
commit 9730c18584
14 changed files with 210 additions and 256 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
neutron/tests/unit/cisco/test_network_plugin.py
View File

@ -303,11 +303,9 @@ class TestCiscoPortsV2(CiscoNetworkPluginV2TestCase,
res = self._create_port(self.fmt, net_id, arg_list=args, res = self._create_port(self.fmt, net_id, arg_list=args,
context=ctx, **port_dict) context=ctx, **port_dict)
port = self.deserialize(self.fmt, res) port = self.deserialize(self.fmt, res)
try: yield res
yield res if do_delete:
finally: self._delete('ports', port['port']['id'])
if do_delete:
self._delete('ports', port['port']['id'])
def test_create_ports_bulk_emulated_plugin_failure(self): def test_create_ports_bulk_emulated_plugin_failure(self):
real_has_attr = hasattr real_has_attr = hasattr
@ -961,11 +959,9 @@ class TestCiscoNetworksV2(CiscoNetworkPluginV2TestCase,
res = self._create_network(self.fmt, net_name, True, res = self._create_network(self.fmt, net_name, True,
arg_list=arg_list, **provider_attrs) arg_list=arg_list, **provider_attrs)
network = self.deserialize(self.fmt, res)['network'] network = self.deserialize(self.fmt, res)['network']
try: yield network
yield network req = self.new_delete_request('networks', network['id'])
finally: req.get_response(self.api)
req = self.new_delete_request('networks', network['id'])
req.get_response(self.api)
def test_create_provider_vlan_network(self): def test_create_provider_vlan_network(self):
with self._provider_vlan_network(PHYS_NET, '1234', with self._provider_vlan_network(PHYS_NET, '1234',
@ -1073,10 +1069,8 @@ class TestCiscoRouterInterfacesV2(CiscoNetworkPluginV2TestCase):
request = self.new_create_request('routers', data, self.fmt) request = self.new_create_request('routers', data, self.fmt)
response = request.get_response(self.ext_api) response = request.get_response(self.ext_api)
router = self.deserialize(self.fmt, response) router = self.deserialize(self.fmt, response)
try: yield network, subnet, router
yield network, subnet, router self._delete('routers', router['router']['id'])
finally:
self._delete('routers', router['router']['id'])
@contextlib.contextmanager @contextlib.contextmanager
def _router_interface(self, router, subnet, **kwargs): def _router_interface(self, router, subnet, **kwargs):
@ -1089,15 +1083,15 @@ class TestCiscoRouterInterfacesV2(CiscoNetworkPluginV2TestCase):
router['router']['id'], router['router']['id'],
'add_router_interface') 'add_router_interface')
response = request.get_response(self.ext_api) response = request.get_response(self.ext_api)
try:
yield response yield response
finally:
# If router interface was created successfully, delete it now. # If router interface was created successfully, delete it now.
if response.status_int == wexc.HTTPOk.code: if response.status_int == wexc.HTTPOk.code:
request = self.new_action_request('routers', interface_data, request = self.new_action_request('routers', interface_data,
router['router']['id'], router['router']['id'],
'remove_router_interface') 'remove_router_interface')
request.get_response(self.ext_api) request.get_response(self.ext_api)
@contextlib.contextmanager @contextlib.contextmanager
def _network_subnet_router_interface(self, **kwargs): def _network_subnet_router_interface(self, **kwargs):

28
neutron/tests/unit/db/firewall/test_db_firewall.py
View File

@ -188,12 +188,10 @@ class FirewallPluginDbTestCase(test_db_plugin.NeutronDbPluginV2TestCase):
if res.status_int >= 400: if res.status_int >= 400:
raise webob.exc.HTTPClientError(code=res.status_int) raise webob.exc.HTTPClientError(code=res.status_int)
firewall_policy = self.deserialize(fmt or self.fmt, res) firewall_policy = self.deserialize(fmt or self.fmt, res)
try: yield firewall_policy
yield firewall_policy if not no_delete:
finally: self._delete('firewall_policies',
if not no_delete: firewall_policy['firewall_policy']['id'])
self._delete('firewall_policies',
firewall_policy['firewall_policy']['id'])
def _create_firewall_rule(self, fmt, name, shared, protocol, def _create_firewall_rule(self, fmt, name, shared, protocol,
ip_version, source_ip_address, ip_version, source_ip_address,
@ -240,12 +238,10 @@ class FirewallPluginDbTestCase(test_db_plugin.NeutronDbPluginV2TestCase):
if res.status_int >= 400: if res.status_int >= 400:
raise webob.exc.HTTPClientError(code=res.status_int) raise webob.exc.HTTPClientError(code=res.status_int)
firewall_rule = self.deserialize(fmt or self.fmt, res) firewall_rule = self.deserialize(fmt or self.fmt, res)
try: yield firewall_rule
yield firewall_rule if not no_delete:
finally: self._delete('firewall_rules',
if not no_delete: firewall_rule['firewall_rule']['id'])
self._delete('firewall_rules',
firewall_rule['firewall_rule']['id'])
def _create_firewall(self, fmt, name, description, firewall_policy_id, def _create_firewall(self, fmt, name, description, firewall_policy_id,
admin_state_up=True, expected_res_status=None, admin_state_up=True, expected_res_status=None,
@ -275,11 +271,9 @@ class FirewallPluginDbTestCase(test_db_plugin.NeutronDbPluginV2TestCase):
if res.status_int >= 400: if res.status_int >= 400:
raise webob.exc.HTTPClientError(code=res.status_int) raise webob.exc.HTTPClientError(code=res.status_int)
firewall = self.deserialize(fmt or self.fmt, res) firewall = self.deserialize(fmt or self.fmt, res)
try: yield firewall
yield firewall if not no_delete:
finally: self._delete('firewalls', firewall['firewall']['id'])
if not no_delete:
self._delete('firewalls', firewall['firewall']['id'])
def _rule_action(self, action, id, firewall_rule_id, insert_before=None, def _rule_action(self, action, id, firewall_rule_id, insert_before=None,
insert_after=None, expected_code=webob.exc.HTTPOk.code, insert_after=None, expected_code=webob.exc.HTTPOk.code,

38
neutron/tests/unit/db/loadbalancer/test_db_loadbalancer.py
View File

@ -227,12 +227,10 @@ class LoadBalancerTestMixin(object):
explanation=_("Unexpected error code: %s") % explanation=_("Unexpected error code: %s") %
res.status_int res.status_int
) )
try: vip = self.deserialize(fmt or self.fmt, res)
vip = self.deserialize(fmt or self.fmt, res) yield vip
yield vip if not no_delete:
finally: self._delete('vips', vip['vip']['id'])
if not no_delete:
self._delete('vips', vip['vip']['id'])
@contextlib.contextmanager @contextlib.contextmanager
def pool(self, fmt=None, name='pool1', lb_method='ROUND_ROBIN', def pool(self, fmt=None, name='pool1', lb_method='ROUND_ROBIN',
@ -250,12 +248,10 @@ class LoadBalancerTestMixin(object):
raise webob.exc.HTTPClientError( raise webob.exc.HTTPClientError(
explanation=_("Unexpected error code: %s") % res.status_int explanation=_("Unexpected error code: %s") % res.status_int
) )
try: pool = self.deserialize(fmt or self.fmt, res)
pool = self.deserialize(fmt or self.fmt, res) yield pool
yield pool if not no_delete:
finally: self._delete('pools', pool['pool']['id'])
if not no_delete:
self._delete('pools', pool['pool']['id'])
@contextlib.contextmanager @contextlib.contextmanager
def member(self, fmt=None, address='192.168.1.100', protocol_port=80, def member(self, fmt=None, address='192.168.1.100', protocol_port=80,
@ -271,12 +267,10 @@ class LoadBalancerTestMixin(object):
raise webob.exc.HTTPClientError( raise webob.exc.HTTPClientError(
explanation=_("Unexpected error code: %s") % res.status_int explanation=_("Unexpected error code: %s") % res.status_int
) )
try: member = self.deserialize(fmt or self.fmt, res)
member = self.deserialize(fmt or self.fmt, res) yield member
yield member if not no_delete:
finally: self._delete('members', member['member']['id'])
if not no_delete:
self._delete('members', member['member']['id'])
@contextlib.contextmanager @contextlib.contextmanager
def health_monitor(self, fmt=None, type='TCP', def health_monitor(self, fmt=None, type='TCP',
@ -310,11 +304,9 @@ class LoadBalancerTestMixin(object):
else: else:
for arg in http_related_attributes: for arg in http_related_attributes:
self.assertIsNone(the_health_monitor.get(arg)) self.assertIsNone(the_health_monitor.get(arg))
try: yield health_monitor
yield health_monitor if not no_delete:
finally: self._delete('health_monitors', the_health_monitor['id'])
if not no_delete:
self._delete('health_monitors', the_health_monitor['id'])
class LoadBalancerPluginDbTestCase(LoadBalancerTestMixin, class LoadBalancerPluginDbTestCase(LoadBalancerTestMixin,

20
neutron/tests/unit/db/metering/test_db_metering.py
View File

@ -96,12 +96,10 @@ class MeteringPluginDbTestCaseMixin(object):
fmt = self.fmt fmt = self.fmt
metering_label = self._make_metering_label(fmt, name, metering_label = self._make_metering_label(fmt, name,
description, **kwargs) description, **kwargs)
try: yield metering_label
yield metering_label if not no_delete:
finally: self._delete('metering-labels',
if not no_delete: metering_label['metering_label']['id'])
self._delete('metering-labels',
metering_label['metering_label']['id'])
@contextlib.contextmanager @contextlib.contextmanager
def metering_label_rule(self, metering_label_id=None, direction='ingress', def metering_label_rule(self, metering_label_id=None, direction='ingress',
@ -114,12 +112,10 @@ class MeteringPluginDbTestCaseMixin(object):
direction, direction,
remote_ip_prefix, remote_ip_prefix,
excluded) excluded)
try: yield metering_label_rule
yield metering_label_rule if not no_delete:
finally: self._delete('metering-label-rules',
if not no_delete: metering_label_rule['metering_label_rule']['id'])
self._delete('metering-label-rules',
metering_label_rule['metering_label_rule']['id'])
class MeteringPluginDbTestCase(test_db_plugin.NeutronDbPluginV2TestCase, class MeteringPluginDbTestCase(test_db_plugin.NeutronDbPluginV2TestCase,

101
neutron/tests/unit/db/vpn/test_db_vpnaas.py
View File

@ -125,12 +125,10 @@ class VPNTestMixin(object):
**kwargs) **kwargs)
if res.status_int >= 400: if res.status_int >= 400:
raise webob.exc.HTTPClientError(code=res.status_int) raise webob.exc.HTTPClientError(code=res.status_int)
try: ikepolicy = self.deserialize(fmt or self.fmt, res)
ikepolicy = self.deserialize(fmt or self.fmt, res) yield ikepolicy
yield ikepolicy if not no_delete:
finally: self._delete('ikepolicies', ikepolicy['ikepolicy']['id'])
if not no_delete:
self._delete('ikepolicies', ikepolicy['ikepolicy']['id'])
def _create_ipsecpolicy(self, fmt, def _create_ipsecpolicy(self, fmt,
name='ipsecpolicy1', name='ipsecpolicy1',
@ -188,12 +186,10 @@ class VPNTestMixin(object):
**kwargs) **kwargs)
if res.status_int >= 400: if res.status_int >= 400:
raise webob.exc.HTTPClientError(code=res.status_int) raise webob.exc.HTTPClientError(code=res.status_int)
try: ipsecpolicy = self.deserialize(fmt or self.fmt, res)
ipsecpolicy = self.deserialize(fmt or self.fmt, res) yield ipsecpolicy
yield ipsecpolicy if not no_delete:
finally: self._delete('ipsecpolicies', ipsecpolicy['ipsecpolicy']['id'])
if not no_delete:
self._delete('ipsecpolicies', ipsecpolicy['ipsecpolicy']['id'])
def _create_vpnservice(self, fmt, name, def _create_vpnservice(self, fmt, name,
admin_state_up, admin_state_up,
@ -250,36 +246,37 @@ class VPNTestMixin(object):
'add', 'add',
tmp_router['router']['id'], tmp_router['router']['id'],
tmp_subnet['subnet']['id'], None) tmp_subnet['subnet']['id'], None)
try:
res = self._create_vpnservice(fmt, res = self._create_vpnservice(fmt,
name, name,
admin_state_up, admin_state_up,
router_id=(tmp_router['router'] router_id=(tmp_router['router']
['id']), ['id']),
subnet_id=(tmp_subnet['subnet'] subnet_id=(tmp_subnet['subnet']
['id']), ['id']),
**kwargs) **kwargs)
vpnservice = self.deserialize(fmt or self.fmt, res) vpnservice = self.deserialize(fmt or self.fmt, res)
if res.status_int >= 400: if res.status_int < 400:
raise webob.exc.HTTPClientError(
code=res.status_int, detail=vpnservice)
yield vpnservice yield vpnservice
finally:
if not no_delete and vpnservice.get('vpnservice'): if not no_delete and vpnservice.get('vpnservice'):
self._delete('vpnservices', self._delete('vpnservices',
vpnservice['vpnservice']['id']) vpnservice['vpnservice']['id'])
if plug_subnet: if plug_subnet:
self._router_interface_action( self._router_interface_action(
'remove', 'remove',
tmp_router['router']['id'], tmp_router['router']['id'],
tmp_subnet['subnet']['id'], None) tmp_subnet['subnet']['id'], None)
if external_router: if external_router:
external_gateway = tmp_router['router'].get( external_gateway = tmp_router['router'].get(
'external_gateway_info') 'external_gateway_info')
if external_gateway: if external_gateway:
network_id = external_gateway['network_id'] network_id = external_gateway['network_id']
self._remove_external_gateway_from_router( self._remove_external_gateway_from_router(
tmp_router['router']['id'], network_id) tmp_router['router']['id'], network_id)
if res.status_int >= 400:
raise webob.exc.HTTPClientError(
code=res.status_int, detail=vpnservice)
def _create_ipsec_site_connection(self, fmt, name='test', def _create_ipsec_site_connection(self, fmt, name='test',
peer_address='192.168.1.10', peer_address='192.168.1.10',
@ -379,18 +376,18 @@ class VPNTestMixin(object):
**kwargs) **kwargs)
if res.status_int >= 400: if res.status_int >= 400:
raise webob.exc.HTTPClientError(code=res.status_int) raise webob.exc.HTTPClientError(code=res.status_int)
try:
ipsec_site_connection = self.deserialize( ipsec_site_connection = self.deserialize(
fmt or self.fmt, res fmt or self.fmt, res
)
yield ipsec_site_connection
if not no_delete:
self._delete(
'ipsec-site-connections',
ipsec_site_connection[
'ipsec_site_connection']['id']
) )
yield ipsec_site_connection
finally:
if not no_delete:
self._delete(
'ipsec-site-connections',
ipsec_site_connection[
'ipsec_site_connection']['id']
)
def _check_ipsec_site_connection(self, ipsec_site_connection, keys, dpd): def _check_ipsec_site_connection(self, ipsec_site_connection, keys, dpd):
self.assertEqual( self.assertEqual(

16
neutron/tests/unit/nec/test_packet_filter.py
View File

@ -97,11 +97,9 @@ class TestNecPluginPacketFilterBase(test_nec_plugin.NecPluginV2TestCase):
with test_plugin.optional_ctx(network, self.network) as network_to_use: with test_plugin.optional_ctx(network, self.network) as network_to_use:
net_id = network_to_use['network']['id'] net_id = network_to_use['network']['id']
pf = self._make_packet_filter(fmt or self.fmt, net_id, **kwargs) pf = self._make_packet_filter(fmt or self.fmt, net_id, **kwargs)
try: yield pf
yield pf if do_delete:
finally: self._delete('packet_filters', pf['packet_filter']['id'])
if do_delete:
self._delete('packet_filters', pf['packet_filter']['id'])
@contextlib.contextmanager @contextlib.contextmanager
def packet_filter_on_port(self, port=None, fmt=None, do_delete=True, def packet_filter_on_port(self, port=None, fmt=None, do_delete=True,
@ -121,11 +119,9 @@ class TestNecPluginPacketFilterBase(test_nec_plugin.NecPluginV2TestCase):
kwargs['in_port'] = port_id kwargs['in_port'] = port_id
pf = self._make_packet_filter(fmt or self.fmt, net_id, **kwargs) pf = self._make_packet_filter(fmt or self.fmt, net_id, **kwargs)
self.assertEqual(port_id, pf['packet_filter']['in_port']) self.assertEqual(port_id, pf['packet_filter']['in_port'])
try: yield pf
yield pf if do_delete:
finally: self._delete('packet_filters', pf['packet_filter']['id'])
if do_delete:
self._delete('packet_filters', pf['packet_filter']['id'])
class TestNecPluginPacketFilter(TestNecPluginPacketFilterBase): class TestNecPluginPacketFilter(TestNecPluginPacketFilterBase):

8
neutron/tests/unit/nuage/test_netpartition.py
View File

@ -63,11 +63,9 @@ class NetPartitionTestCase(test_nuage_plugin.NuagePluginV2TestCase):
**kwargs): **kwargs):
netpart = self._make_netpartition(fmt or self.fmt, name) netpart = self._make_netpartition(fmt or self.fmt, name)
try: yield netpart
yield netpart if do_delete:
finally: self._del_netpartition(netpart['net_partition']['id'])
if do_delete:
self._del_netpartition(netpart['net_partition']['id'])
def test_create_netpartition(self): def test_create_netpartition(self):
name = 'netpart1' name = 'netpart1'

32
neutron/tests/unit/test_db_plugin.py
View File

@ -523,15 +523,13 @@ class NeutronDbPluginV2TestCase(testlib_api.WebTestCase):
**kwargs): **kwargs):
network = self._make_network(fmt or self.fmt, name, network = self._make_network(fmt or self.fmt, name,
admin_state_up, **kwargs) admin_state_up, **kwargs)
try: yield network
yield network if do_delete:
finally: # The do_delete parameter allows you to control whether the
if do_delete: # created network is immediately deleted again. Therefore, this
# The do_delete parameter allows you to control whether the # function is also usable in tests, which require the creation
# created network is immediately deleted again. Therefore, this # of many networks.
# function is also usable in tests, which require the creation self._delete('networks', network['network']['id'])
# of many networks.
self._delete('networks', network['network']['id'])
@contextlib.contextmanager @contextlib.contextmanager
def subnet(self, network=None, def subnet(self, network=None,
@ -560,11 +558,9 @@ class NeutronDbPluginV2TestCase(testlib_api.WebTestCase):
shared=shared, shared=shared,
ipv6_ra_mode=ipv6_ra_mode, ipv6_ra_mode=ipv6_ra_mode,
ipv6_address_mode=ipv6_address_mode) ipv6_address_mode=ipv6_address_mode)
try: yield subnet
yield subnet if do_delete:
finally: self._delete('subnets', subnet['subnet']['id'])
if do_delete:
self._delete('subnets', subnet['subnet']['id'])
@contextlib.contextmanager @contextlib.contextmanager
def port(self, subnet=None, fmt=None, no_delete=False, def port(self, subnet=None, fmt=None, no_delete=False,
@ -572,11 +568,9 @@ class NeutronDbPluginV2TestCase(testlib_api.WebTestCase):
with optional_ctx(subnet, self.subnet) as subnet_to_use: with optional_ctx(subnet, self.subnet) as subnet_to_use:
net_id = subnet_to_use['subnet']['network_id'] net_id = subnet_to_use['subnet']['network_id']
port = self._make_port(fmt or self.fmt, net_id, **kwargs) port = self._make_port(fmt or self.fmt, net_id, **kwargs)
try: yield port
yield port if not no_delete:
finally: self._delete('ports', port['port']['id'])
if not no_delete:
self._delete('ports', port['port']['id'])
def _test_list_with_sort(self, resource, def _test_list_with_sort(self, resource,
items, sorts, resources=None, query_params=''): items, sorts, resources=None, query_params=''):

20
neutron/tests/unit/test_extension_security_group.py
View File

@ -126,12 +126,10 @@ class SecurityGroupsTestCase(test_db_plugin.NeutronDbPluginV2TestCase):
if not fmt: if not fmt:
fmt = self.fmt fmt = self.fmt
security_group = self._make_security_group(fmt, name, description) security_group = self._make_security_group(fmt, name, description)
try: yield security_group
yield security_group if not no_delete:
finally: self._delete('security-groups',
if not no_delete: security_group['security_group']['id'])
self._delete('security-groups',
security_group['security_group']['id'])
@contextlib.contextmanager @contextlib.contextmanager
def security_group_rule(self, security_group_id='4cd70774-cc67-4a87-9b39-7' def security_group_rule(self, security_group_id='4cd70774-cc67-4a87-9b39-7'
@ -150,12 +148,10 @@ class SecurityGroupsTestCase(test_db_plugin.NeutronDbPluginV2TestCase):
remote_group_id, remote_group_id,
ethertype=ethertype) ethertype=ethertype)
security_group_rule = self._make_security_group_rule(self.fmt, rule) security_group_rule = self._make_security_group_rule(self.fmt, rule)
try: yield security_group_rule
yield security_group_rule if not no_delete:
finally: self._delete('security-group-rules',
if not no_delete: security_group_rule['security_group_rule']['id'])
self._delete('security-group-rules',
security_group_rule['security_group_rule']['id'])
def _delete_default_security_group_egress_rules(self, security_group_id): def _delete_default_security_group_egress_rules(self, security_group_id):
"""Deletes default egress rules given a security group ID.""" """Deletes default egress rules given a security group ID."""

98
neutron/tests/unit/test_l3_plugin.py
View File

@ -358,10 +358,8 @@ class L3NatTestCaseMixin(object):
router = self._make_router(fmt or self.fmt, tenant_id, name, router = self._make_router(fmt or self.fmt, tenant_id, name,
admin_state_up, external_gateway_info, admin_state_up, external_gateway_info,
set_context, **kwargs) set_context, **kwargs)
try: yield router
yield router self._delete('routers', router['router']['id'])
finally:
self._delete('routers', router['router']['id'])
def _set_net_external(self, net_id): def _set_net_external(self, net_id):
self._update('networks', net_id, self._update('networks', net_id,
@ -413,31 +411,31 @@ class L3NatTestCaseMixin(object):
sid = private_port['port']['fixed_ips'][0]['subnet_id'] sid = private_port['port']['fixed_ips'][0]['subnet_id']
private_sub = {'subnet': {'id': sid}} private_sub = {'subnet': {'id': sid}}
floatingip = None floatingip = None
try:
self._add_external_gateway_to_router(
r['router']['id'],
public_sub['subnet']['network_id'])
self._router_interface_action(
'add', r['router']['id'],
private_sub['subnet']['id'], None)
floatingip = self._make_floatingip( self._add_external_gateway_to_router(
fmt or self.fmt, r['router']['id'],
public_sub['subnet']['network_id'], public_sub['subnet']['network_id'])
port_id=private_port['port']['id'], self._router_interface_action(
fixed_ip=fixed_ip, 'add', r['router']['id'],
set_context=False) private_sub['subnet']['id'], None)
yield floatingip
finally: floatingip = self._make_floatingip(
if floatingip: fmt or self.fmt,
self._delete('floatingips', public_sub['subnet']['network_id'],
floatingip['floatingip']['id']) port_id=private_port['port']['id'],
self._router_interface_action( fixed_ip=fixed_ip,
'remove', r['router']['id'], set_context=False)
private_sub['subnet']['id'], None) yield floatingip
self._remove_external_gateway_from_router(
r['router']['id'], if floatingip:
public_sub['subnet']['network_id']) self._delete('floatingips',
floatingip['floatingip']['id'])
self._router_interface_action(
'remove', r['router']['id'],
private_sub['subnet']['id'], None)
self._remove_external_gateway_from_router(
r['router']['id'],
public_sub['subnet']['network_id'])
@contextlib.contextmanager @contextlib.contextmanager
def floatingip_no_assoc_with_public_sub( def floatingip_no_assoc_with_public_sub(
@ -445,29 +443,29 @@ class L3NatTestCaseMixin(object):
self._set_net_external(public_sub['subnet']['network_id']) self._set_net_external(public_sub['subnet']['network_id'])
with self.router() as r: with self.router() as r:
floatingip = None floatingip = None
try:
self._add_external_gateway_to_router(
r['router']['id'],
public_sub['subnet']['network_id'])
self._router_interface_action('add', r['router']['id'],
private_sub['subnet']['id'],
None)
floatingip = self._make_floatingip( self._add_external_gateway_to_router(
fmt or self.fmt, r['router']['id'],
public_sub['subnet']['network_id'], public_sub['subnet']['network_id'])
set_context=set_context) self._router_interface_action('add', r['router']['id'],
yield floatingip, r private_sub['subnet']['id'],
finally: None)
if floatingip:
self._delete('floatingips', floatingip = self._make_floatingip(
floatingip['floatingip']['id']) fmt or self.fmt,
self._router_interface_action('remove', r['router']['id'], public_sub['subnet']['network_id'],
private_sub['subnet']['id'], set_context=set_context)
None) yield floatingip, r
self._remove_external_gateway_from_router(
r['router']['id'], if floatingip:
public_sub['subnet']['network_id']) self._delete('floatingips',
floatingip['floatingip']['id'])
self._router_interface_action('remove', r['router']['id'],
private_sub['subnet']['id'],
None)
self._remove_external_gateway_from_router(
r['router']['id'],
public_sub['subnet']['network_id'])
@contextlib.contextmanager @contextlib.contextmanager
def floatingip_no_assoc(self, private_sub, fmt=None, set_context=False): def floatingip_no_assoc(self, private_sub, fmt=None, set_context=False):

12
neutron/tests/unit/test_l3_schedulers.py
View File

@ -117,12 +117,12 @@ class L3SchedulerTestCase(l3_agentschedulers_db.L3AgentSchedulerDbMixin,
self._add_external_gateway_to_router( self._add_external_gateway_to_router(
router['router']['id'], router['router']['id'],
subnet['subnet']['network_id']) subnet['subnet']['network_id'])
try:
yield router yield router
finally:
self._remove_external_gateway_from_router( self._remove_external_gateway_from_router(
router['router']['id'], subnet['subnet']['network_id']) router['router']['id'], subnet['subnet']['network_id'])
self._delete('routers', router['router']['id']) self._delete('routers', router['router']['id'])
class L3AgentChanceSchedulerTestCase(L3SchedulerTestCase): class L3AgentChanceSchedulerTestCase(L3SchedulerTestCase):

12
neutron/tests/unit/vmware/extensions/test_qosqueues.py
View File

@ -76,12 +76,12 @@ class TestQoSQueue(NsxPluginV2TestCase):
qos_queue = self.deserialize('json', res) qos_queue = self.deserialize('json', res)
if res.status_int >= 400: if res.status_int >= 400:
raise webob.exc.HTTPClientError(code=res.status_int) raise webob.exc.HTTPClientError(code=res.status_int)
try:
yield qos_queue yield qos_queue
finally:
if not no_delete: if not no_delete:
self._delete('qos-queues', self._delete('qos-queues',
qos_queue['qos_queue']['id']) qos_queue['qos_queue']['id'])
def test_create_qos_queue(self): def test_create_qos_queue(self):
with self.qos_queue(name='fake_lqueue', min=34, max=44, with self.qos_queue(name='fake_lqueue', min=34, max=44,

21
neutron/tests/unit/vmware/test_nsx_sync.py
View File

@ -355,17 +355,16 @@ class SyncTestCase(base.BaseTestCase):
routers.append(self._plugin.create_router(ctx, router(i))) routers.append(self._plugin.create_router(ctx, router(i)))
# Do not return anything as the user does need the actual # Do not return anything as the user does need the actual
# data created # data created
try: yield
yield
finally: # Remove everything
# Remove everything for router in routers:
for router in routers: self._plugin.delete_router(ctx, router['id'])
self._plugin.delete_router(ctx, router['id']) for port in ports:
for port in ports: self._plugin.delete_port(ctx, port['id'])
self._plugin.delete_port(ctx, port['id']) # This will remove networks and subnets
# This will remove networks and subnets for network in networks:
for network in networks: self._plugin.delete_network(ctx, network['id'])
self._plugin.delete_network(ctx, network['id'])
def _get_tag_dict(self, tags): def _get_tag_dict(self, tags):
return dict((tag['scope'], tag['tag']) for tag in tags) return dict((tag['scope'], tag['tag']) for tag in tags)

20
neutron/tests/unit/vmware/vshield/test_vpnaas_plugin.py
View File

@ -85,16 +85,16 @@ class TestVpnPlugin(test_db_vpnaas.VPNTestMixin,
data = {'router': {'tenant_id': self._tenant_id}} data = {'router': {'tenant_id': self._tenant_id}}
data['router']['service_router'] = True data['router']['service_router'] = True
router_req = self.new_create_request('routers', data, self.fmt) router_req = self.new_create_request('routers', data, self.fmt)
try:
res = router_req.get_response(self.ext_api) res = router_req.get_response(self.ext_api)
router = self.deserialize(self.fmt, res) router = self.deserialize(self.fmt, res)
self._add_external_gateway_to_router( self._add_external_gateway_to_router(
router['router']['id'], router['router']['id'],
s['subnet']['network_id']) s['subnet']['network_id'])
router = self._show('routers', router['router']['id']) router = self._show('routers', router['router']['id'])
yield router yield router
finally:
self._delete('routers', router['router']['id']) self._delete('routers', router['router']['id'])
def test_create_vpnservice(self, **extras): def test_create_vpnservice(self, **extras):
"""Test case to create a vpnservice.""" """Test case to create a vpnservice."""