x/vmware-nsx
Code Issues Proposed changes

Merge "Releasing resources of context manager functions if exceptions occur"

Browse Source
This commit is contained in:
Jenkins 2012-12-06 15:52:36 +00:00 committed by Gerrit Code Review
commit 2f8ccfa879
3 changed files with 137 additions and 103 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

90
quantum/tests/unit/test_db_plugin.py
View File

@ -302,6 +302,16 @@ class QuantumDbPluginV2TestCase(unittest2.TestCase):
'tenant_id': self._tenant_id}}
return self._create_bulk(fmt, number, 'port', base_data, **kwargs)
def _make_network(self, fmt, name, admin_status_up, **kwargs):
res = self._create_network(fmt, name, admin_status_up, **kwargs)
# TODO(salvatore-orlando): do exception handling in this test module
# in a uniform way (we do it differently for ports, subnets, and nets
# Things can go wrong - raise HTTP exc with res code only
# so it can be caught by unit tests
if res.status_int >= 400:
raise webob.exc.HTTPClientError(code=res.status_int)
return self.deserialize(fmt, res)
def _make_subnet(self, fmt, network, gateway, cidr,
allocation_pools=None, ip_version=4, enable_dhcp=True,
dns_nameservers=None, host_routes=None, shared=None):
@ -322,6 +332,14 @@ class QuantumDbPluginV2TestCase(unittest2.TestCase):
raise webob.exc.HTTPClientError(code=res.status_int)
return self.deserialize(fmt, res)
def _make_port(self, fmt, net_id, expected_res_status=None, **kwargs):
res = self._create_port(fmt, net_id, expected_res_status, **kwargs)
# Things can go wrong - raise HTTP exc with res code only
# so it can be caught by unit tests
if res.status_int >= 400:
raise webob.exc.HTTPClientError(code=res.status_int)
return self.deserialize(fmt, res)
def _api_for_resource(self, resource):
if resource in ['networks', 'subnets', 'ports']:
return self.api
@ -393,25 +411,16 @@ class QuantumDbPluginV2TestCase(unittest2.TestCase):
fmt='json',
do_delete=True,
**kwargs):
res = self._create_network(fmt,
name,
admin_status_up,
**kwargs)
network = self.deserialize(fmt, res)
# TODO(salvatore-orlando): do exception handling in this test module
# in a uniform way (we do it differently for ports, subnets, and nets
# Things can go wrong - raise HTTP exc with res code only
# so it can be caught by unit tests
if res.status_int >= 400:
raise webob.exc.HTTPClientError(code=res.status_int)
yield network
if do_delete:
# The do_delete parameter allows you to control whether the
# created network is immediately deleted again. Therefore, this
# function is also usable in tests, which require the creation
# of many networks.
self._delete('networks', network['network']['id'])
network = self._make_network(fmt, name, admin_status_up, **kwargs)
try:
yield network
finally:
if do_delete:
# The do_delete parameter allows you to control whether the
# created network is immediately deleted again. Therefore, this
# function is also usable in tests, which require the creation
# of many networks.
self._delete('networks', network['network']['id'])
@contextlib.contextmanager
def subnet(self, network=None,
@ -439,8 +448,10 @@ class QuantumDbPluginV2TestCase(unittest2.TestCase):
dns_nameservers,
host_routes,
shared=shared)
yield subnet
self._delete('subnets', subnet['subnet']['id'])
try:
yield subnet
finally:
self._delete('subnets', subnet['subnet']['id'])
else:
subnet = self._make_subnet(fmt,
network,
@ -452,8 +463,10 @@ class QuantumDbPluginV2TestCase(unittest2.TestCase):
dns_nameservers,
host_routes,
shared=shared)
yield subnet
self._delete('subnets', subnet['subnet']['id'])
try:
yield subnet
finally:
self._delete('subnets', subnet['subnet']['id'])
@contextlib.contextmanager
def port(self, subnet=None, fmt='json', no_delete=False,
@ -461,27 +474,20 @@ class QuantumDbPluginV2TestCase(unittest2.TestCase):
if not subnet:
with self.subnet() as subnet:
net_id = subnet['subnet']['network_id']
res = self._create_port(fmt, net_id, **kwargs)
port = self.deserialize(fmt, res)
# Things can go wrong - raise HTTP exc with res code only
# so it can be caught by unit tests
if res.status_int >= 400:
raise webob.exc.HTTPClientError(code=res.status_int)
yield port
if not no_delete:
self._delete('ports', port['port']['id'])
port = self._make_port(fmt, net_id, **kwargs)
try:
yield port
finally:
if not no_delete:
self._delete('ports', port['port']['id'])
else:
net_id = subnet['subnet']['network_id']
res = self._create_port(fmt, net_id, **kwargs)
port = self.deserialize(fmt, res)
# Things can go wrong - raise HTTP exc with res code only
# so it can be caught by unit tests
if res.status_int >= 400:
raise webob.exc.HTTPClientError(code=res.status_int)
yield port
if not no_delete:
self._delete('ports', port['port']['id'])
port = self._make_port(fmt, net_id, **kwargs)
try:
yield port
finally:
if not no_delete:
self._delete('ports', port['port']['id'])
class TestBasicGet(QuantumDbPluginV2TestCase):

53
quantum/tests/unit/test_extension_security_group.py
View File

@ -109,18 +109,31 @@ class SecurityGroupsTestCase(test_db_plugin.QuantumDbPluginV2TestCase):
context.Context('', kwargs['tenant_id']))
return security_group_rule_req.get_response(self.ext_api)
def _make_security_group(self, fmt, name, description, external_id=None,
**kwargs):
res = self._create_security_group(fmt, name, description,
external_id, **kwargs)
if res.status_int >= 400:
raise webob.exc.HTTPClientError(code=res.status_int)
return self.deserialize(fmt, res)
def _make_security_group_rule(self, fmt, rules, **kwargs):
res = self._create_security_group_rule('json', rules)
if res.status_int >= 400:
raise webob.exc.HTTPClientError(code=res.status_int)
return self.deserialize(fmt, res)
@contextlib.contextmanager
def security_group(self, name='webservers', description='webservers',
external_id=None, fmt='json', no_delete=False):
res = self._create_security_group(fmt, name, description,
external_id)
security_group = self.deserialize(fmt, res)
if res.status_int >= 400:
raise webob.exc.HTTPClientError(code=res.status_int)
yield security_group
if not no_delete:
self._delete('security-groups',
security_group['security_group']['id'])
security_group = self._make_security_group(fmt, name, description,
external_id)
try:
yield security_group
finally:
if not no_delete:
self._delete('security-groups',
security_group['security_group']['id'])
@contextlib.contextmanager
def security_group_rule(self, security_group_id='4cd70774-cc67-4a87-9b39-7'
@ -129,20 +142,20 @@ class SecurityGroupsTestCase(test_db_plugin.QuantumDbPluginV2TestCase):
port_range_min='22', port_range_max='22',
source_ip_prefix=None, source_group_id=None,
external_id=None, fmt='json', no_delete=False):
rule = self._build_security_group_rule(security_group_id, direction,
rule = self._build_security_group_rule(security_group_id,
direction,
protocol, port_range_min,
port_range_max,
source_ip_prefix,
source_group_id, external_id)
res = self._create_security_group_rule('json', rule)
security_group_rule = self.deserialize(fmt, res)
if res.status_int >= 400:
raise webob.exc.HTTPClientError(code=res.status_int)
yield security_group_rule
if not no_delete:
self._delete('security-group-rules',
security_group_rule['security_group_rule']['id'])
source_group_id,
external_id)
security_group_rule = self._make_security_group_rule('json', rule)
try:
yield security_group_rule
finally:
if not no_delete:
self._delete('security-group-rules',
security_group_rule['security_group_rule']['id'])
class SecurityGroupTestPlugin(db_base_plugin_v2.QuantumDbPluginV2,

97
quantum/tests/unit/test_l3_plugin.py
View File

@ -301,6 +301,10 @@ class L3NatDBTestCase(test_db_plugin.QuantumDbPluginV2TestCase):
router_req = self.new_create_request('routers', data, fmt)
return router_req.get_response(self.ext_api)
def _make_router(self, fmt, tenant_id, name=None, admin_state_up=None):
res = self._create_router(fmt, tenant_id, name, admin_state_up)
return self.deserialize(fmt, res)
def _add_external_gateway_to_router(self, router_id, network_id,
expected_code=exc.HTTPOk.code):
return self._update('routers', router_id,
@ -332,11 +336,11 @@ class L3NatDBTestCase(test_db_plugin.QuantumDbPluginV2TestCase):
@contextlib.contextmanager
def router(self, name='router1', admin_status_up=True,
fmt='json', tenant_id=_uuid()):
res = self._create_router(fmt, tenant_id, name=name,
admin_state_up=admin_status_up)
router = self.deserialize(fmt, res)
yield router
self._delete('routers', router['router']['id'])
router = self._make_router(fmt, tenant_id, name, admin_status_up)
try:
yield router
finally:
self._delete('routers', router['router']['id'])
def test_router_create(self):
name = 'router1'
@ -907,6 +911,12 @@ class L3NatDBTestCase(test_db_plugin.QuantumDbPluginV2TestCase):
floatingip_req = self.new_create_request('floatingips', data, fmt)
return floatingip_req.get_response(self.ext_api)
def _make_floatingip(self, fmt, network_id, port_id=None,
fixed_ip=None):
res = self._create_floatingip(fmt, network_id, port_id, fixed_ip)
self.assertEqual(res.status_int, exc.HTTPCreated.code)
return self.deserialize(fmt, res)
def _validate_floating_ip(self, fip):
body = self._list('floatingips')
self.assertEquals(len(body['floatingips']), 1)
@ -925,6 +935,38 @@ class L3NatDBTestCase(test_db_plugin.QuantumDbPluginV2TestCase):
with self.router() as r:
sid = private_port['port']['fixed_ips'][0]['subnet_id']
private_sub = {'subnet': {'id': sid}}
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(
fmt,
public_sub['subnet']['network_id'],
port_id=private_port['port']['id'])
yield floatingip
finally:
if floatingip:
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
def floatingip_no_assoc(self, private_sub, fmt='json'):
with self.subnet(cidr='12.0.0.0/24') as public_sub:
self._set_net_external(public_sub['subnet']['network_id'])
with self.router() as r:
floatingip = None
try:
self._add_external_gateway_to_router(
r['router']['id'],
public_sub['subnet']['network_id'])
@ -932,47 +974,20 @@ class L3NatDBTestCase(test_db_plugin.QuantumDbPluginV2TestCase):
private_sub['subnet']['id'],
None)
res = self._create_floatingip(
floatingip = self._make_floatingip(
fmt,
public_sub['subnet']['network_id'],
port_id=private_port['port']['id'])
self.assertEqual(res.status_int, exc.HTTPCreated.code)
floatingip = self.deserialize(fmt, res)
public_sub['subnet']['network_id'])
yield floatingip
self._delete('floatingips', floatingip['floatingip']['id'])
finally:
if floatingip:
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'])
self._router_interface_action('remove',
r['router']['id'],
private_sub['subnet']['id'],
None)
@contextlib.contextmanager
def floatingip_no_assoc(self, private_sub, fmt='json'):
with self.subnet(cidr='12.0.0.0/24') as public_sub:
self._set_net_external(public_sub['subnet']['network_id'])
with self.router() as r:
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)
res = self._create_floatingip(
fmt,
public_sub['subnet']['network_id'])
self.assertEqual(res.status_int, exc.HTTPCreated.code)
floatingip = self.deserialize(fmt, res)
yield floatingip
self._delete('floatingips', floatingip['floatingip']['id'])
self._remove_external_gateway_from_router(
r['router']['id'],
public_sub['subnet']['network_id'])
self._router_interface_action('remove', r['router']['id'],
private_sub['subnet']['id'],
None)
def test_floatingip_crd_ops(self):
with self.floatingip_with_assoc() as fip: