x/vmware-nsx
Code Issues Proposed changes

Use different context for each API request in unit tests

Browse Source 
test_router_add_interface_subnet_with_port_from_other_tenant in
neutron.tests.unit.test_l3_plugin.L3NatTestCaseBase was mocking
neutron.context.Context thus performing multiple API requests
with the same context instance. As a context instance also has
a DB session attribute, this might cause unexpected side effects,
especially for plugins which process request asynchronously.

The plugin neutron.plugins.nicira.NeutronServicePlugin was being
affected.

This patch ensures each request has a different context object
without changing the unit test semantics.

It also refactors slightly test_edge_router.py in the nicira
unit test package to avoid executing twice the same unit tests.

Change-Id: I4895faa00ebd915eb9e259930db2d004a9e78a86
Closes-Bug: #1280035
This commit is contained in:
Salvatore Orlando 2014-02-14 08:17:30 -08:00
parent 9c0fc8e981
commit 8cf94ebe0d
2 changed files with 38 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
neutron/tests/unit/nicira/test_edge_router.py
View File

@ -63,19 +63,6 @@ class ServiceRouterTestExtensionManager(object):
return []
class NvpRouterTestCase(test_nicira_plugin.TestNiciraL3NatTestCase):
def setUp(self, plugin=None, ext_mgr=None, service_plugins=None):
plugin = plugin or SERVICE_PLUGIN_NAME
# Disable the proxying in the tests but functionality will
# be covered separately
mock_proxy = mock.patch(
"%s.%s" % (SERVICE_PLUGIN_NAME, '_set_create_lswitch_proxy'))
mock_proxy.start()
super(NvpRouterTestCase, self).setUp(plugin=plugin, ext_mgr=ext_mgr,
service_plugins=service_plugins)
class ServiceRouterTest(test_nicira_plugin.NiciraL3NatTest,
test_l3_plugin.L3NatTestCaseMixin):
@ -169,7 +156,8 @@ class ServiceRouterTest(test_nicira_plugin.NiciraL3NatTest,
return router_req.get_response(self.ext_api)
class ServiceRouterTestCase(ServiceRouterTest, NvpRouterTestCase):
class ServiceRouterTestCase(ServiceRouterTest,
test_nicira_plugin.TestNiciraL3NatTestCase):
def test_router_create(self):
name = 'router1'

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

@ -379,7 +379,8 @@ class L3NatTestCaseMixin(object):
def _router_interface_action(self, action, router_id, subnet_id, port_id,
expected_code=exc.HTTPOk.code,
expected_body=None):
expected_body=None,
tenant_id=None):
interface_data = {}
if subnet_id:
interface_data.update({'subnet_id': subnet_id})
@ -388,6 +389,10 @@ class L3NatTestCaseMixin(object):
req = self.new_action_request('routers', interface_data, router_id,
"%s_router_interface" % action)
# if tenant_id was specified, create a tenant context for this request
if tenant_id:
req.environ['neutron.context'] = context.Context(
'', tenant_id)
res = req.get_response(self.ext_api)
self.assertEqual(res.status_int, expected_code)
response = self.deserialize(self.fmt, res)
@ -734,42 +739,37 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
def test_router_add_interface_subnet_with_port_from_other_tenant(self):
tenant_id = _uuid()
other_tenant_id = _uuid()
tenant_context = context.Context(user_id=None, tenant_id=tenant_id)
admin_context = context.get_admin_context()
with mock.patch('neutron.context.Context') as ctx:
ctx.return_value = admin_context
with contextlib.nested(
self.router(tenant_id=tenant_id),
self.network(tenant_id=tenant_id),
self.network(tenant_id=other_tenant_id)) as (r, n1, n2):
with contextlib.nested(
self.router(tenant_id=tenant_id),
self.network(tenant_id=tenant_id),
self.network(tenant_id=other_tenant_id)) as (r, n1, n2):
with contextlib.nested(
self.subnet(network=n1, cidr='10.0.0.0/24'),
self.subnet(network=n2, cidr='10.1.0.0/24')) as (s1, s2):
ctx.return_value = admin_context
body = self._router_interface_action(
'add',
r['router']['id'],
s2['subnet']['id'],
None)
self.assertIn('port_id', body)
ctx.return_value = tenant_context
self._router_interface_action(
'add',
r['router']['id'],
s1['subnet']['id'],
None)
self.assertIn('port_id', body)
self._router_interface_action(
'remove',
r['router']['id'],
s1['subnet']['id'],
None)
ctx.return_value = admin_context
body = self._router_interface_action(
'remove',
r['router']['id'],
s2['subnet']['id'],
None)
self.subnet(network=n1, cidr='10.0.0.0/24'),
self.subnet(network=n2, cidr='10.1.0.0/24')) as (s1, s2):
body = self._router_interface_action(
'add',
r['router']['id'],
s2['subnet']['id'],
None)
self.assertIn('port_id', body)
self._router_interface_action(
'add',
r['router']['id'],
s1['subnet']['id'],
None,
tenant_id=tenant_id)
self.assertIn('port_id', body)
self._router_interface_action(
'remove',
r['router']['id'],
s1['subnet']['id'],
None,
tenant_id=tenant_id)
body = self._router_interface_action(
'remove',
r['router']['id'],
s2['subnet']['id'],
None)
def test_router_add_interface_port(self):
with self.router() as r: