x/vmware-nsx
Code Issues Proposed changes

UTs: Disable auto deletion of ports/subnets/nets

Browse Source 
Change the behavior of the unit test context managers
for ports, networks, and subnets to skip the deletion
of the objects they create.

Adds explicit delete calls to the tests that require
deletion of the objects.

This is the first patch in a series to eliminate the
automatic deletion of objects created during tests to
improve test performance and make testing more explicit.

Implements: blueprint remove-unit-test-autodeletion
Change-Id: Ifc4490b121310ad64e18936e6fd94a2c9a1e36f2
This commit is contained in:
Kevin Benton 2014-06-25 06:12:43 +00:00
parent c0e55e9194
commit a48c5d9838
21 changed files with 129 additions and 86 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
neutron/tests/unit/bigswitch/test_restproxy_plugin.py
View File

@ -85,8 +85,8 @@ class TestBigSwitchProxyPortsV2(test_plugin.TestPortsV2,
def test_router_port_status_active(self):
# router ports screw up port auto-deletion so it has to be
# disabled for this test
with self.network(do_delete=False) as net:
with self.subnet(network=net, do_delete=False) as sub:
with self.network() as net:
with self.subnet(network=net) as sub:
with self.port(
subnet=sub,
do_delete=False,
@ -214,6 +214,8 @@ class TestBigSwitchProxyPortsV2(test_plugin.TestPortsV2,
with self.port(**kwargs) as port:
self.assertEqual(port['port']['binding:vif_type'],
portbindings.VIF_TYPE_IVS)
self._delete('ports', port['port']['id'])
self._delete('networks', port['port']['network_id'])
kwargs = {'name': 'name2', 'binding:host_id': 'someotherhost',
'device_id': 'other_dev'}
with self.port(**kwargs) as port:

8
neutron/tests/unit/bigswitch/test_router_db.py
View File

@ -97,7 +97,7 @@ class RouterDBTestCase(RouterDBTestBase,
with self.router() as r:
with self.subnet() as s:
with self.subnet(cidr='10.0.10.0/24') as s1:
with self.port(subnet=s1, do_delete=False) as p:
with self.port(subnet=s1) as p:
self._router_interface_action('add',
r['router']['id'],
None,
@ -116,7 +116,7 @@ class RouterDBTestCase(RouterDBTestBase,
def test_router_remove_router_interface_wrong_port_returns_404(self):
with self.router() as r:
with self.subnet() as s:
with self.port(subnet=s, do_delete=False) as p:
with self.port(subnet=s) as p:
self._router_interface_action('add',
r['router']['id'],
None,
@ -262,7 +262,7 @@ class RouterDBTestCase(RouterDBTestBase,
def test_router_remove_interface_wrong_subnet_returns_400(self):
with self.router() as r:
with self.subnet(cidr='10.0.10.0/24') as s:
with self.port(do_delete=False) as p:
with self.port() as p:
self._router_interface_action('add',
r['router']['id'],
None,
@ -281,7 +281,7 @@ class RouterDBTestCase(RouterDBTestBase,
def test_router_remove_interface_wrong_port_returns_404(self):
with self.router() as r:
with self.subnet(cidr='10.0.10.0/24'):
with self.port(do_delete=False) as p:
with self.port() as p:
self._router_interface_action('add',
r['router']['id'],
None,

7
neutron/tests/unit/bigswitch/test_servermanager.py
View File

@ -116,14 +116,14 @@ class ServerManagerTests(test_rp.BigSwitchProxyPluginV2TestCase):
rv.getresponse.return_value.getheader.return_value = 'HASHHEADER'
rv.getresponse.return_value.status = 200
rv.getresponse.return_value.read.return_value = ''
with self.network():
with self.network() as network:
callheaders = rv.request.mock_calls[0][1][3]
self.assertIn('X-BSN-BVS-HASH-MATCH', callheaders)
# first call will be empty to indicate no previous state hash
self.assertEqual(callheaders['X-BSN-BVS-HASH-MATCH'], '')
# change the header that will be received on delete call
rv.getresponse.return_value.getheader.return_value = 'HASH2'
self._delete('networks', network['network']['id'])
# net delete should have used header received on create
callheaders = rv.request.mock_calls[1][1][3]
self.assertEqual(callheaders['X-BSN-BVS-HASH-MATCH'], 'HASHHEADER')
@ -142,10 +142,11 @@ class ServerManagerTests(test_rp.BigSwitchProxyPluginV2TestCase):
rv.getresponse.return_value.getheader.return_value = 'HASHHEADER'
rv.getresponse.return_value.status = 200
rv.getresponse.return_value.read.return_value = ''
with self.network():
with self.network() as net:
# change the header that will be received on delete call
rv.getresponse.return_value.getheader.return_value = 'EVIL'
rv.getresponse.return_value.status = 'GARBAGE'
self._delete('networks', net['network']['id'])
# create again should not use header from delete call
with self.network():

3
neutron/tests/unit/cisco/n1kv/test_n1kv_plugin.py
View File

@ -1052,6 +1052,9 @@ class TestN1kvSubnets(test_plugin.TestSubnetsV2,
self.skipTest("Unsupported test")
super(TestN1kvSubnets, self).setUp()
def test_port_prevents_network_deletion(self):
self.skipTest("plugin does not return standard conflict code")
def test_create_subnet_with_invalid_parameters(self):
"""Test subnet creation with invalid parameters sent to the VSM"""
with self.network() as network:

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

@ -307,6 +307,8 @@ class TestCiscoPortsV2(CiscoNetworkPluginV2TestCase,
yield res
if do_delete:
self._delete('ports', port['port']['id'])
self._delete('subnets', subnet['subnet']['id'])
self._delete('networks', network['network']['id'])
def test_create_ports_bulk_emulated_plugin_failure(self):
real_has_attr = hasattr
@ -399,12 +401,13 @@ class TestCiscoPortsV2(CiscoNetworkPluginV2TestCase,
'device_id': device_id,
'device_owner': DEVICE_OWNER}
with self.port(subnet=subnet, fmt=self.fmt,
arg_list=arg_list, **port_dict):
arg_list=arg_list, **port_dict) as port:
self.assertTrue(self._is_vlan_configured(
vlan_creation_expected=vlan_creation_expected,
add_keyword_expected=False))
self.mock_ncclient.reset_mock()
yield
self._delete('ports', port['port']['id'])
# Create network and subnet
with self.network(name=NETWORK_NAME) as network:

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

@ -271,6 +271,9 @@ class VPNTestMixin(object):
if res.status_int >= 400:
raise webob.exc.HTTPClientError(
code=res.status_int, detail=vpnservice)
self._delete('subnets', public_sub['subnet']['id'])
if not subnet:
self._delete('subnets', tmp_subnet['subnet']['id'])
def _create_ipsec_site_connection(self, fmt, name='test',
peer_address='192.168.1.10',
@ -1132,6 +1135,7 @@ class TestVpnaas(VPNPluginDbTestCase):
dpd)
except webob.exc.HTTPClientError as ce:
self.assertEqual(ce.code, expected_status_int)
self._delete('subnets', subnet['subnet']['id'])
def test_create_ipsec_site_connection(self, **extras):
"""Test case to create an ipsec_site_connection."""
@ -1256,6 +1260,7 @@ class TestVpnaas(VPNPluginDbTestCase):
self.assertEqual(v, sorted(actual[k]))
else:
self.assertEqual(v, actual[k])
self._delete('networks', subnet['subnet']['network_id'])
def test_show_ipsec_site_connection(self):
"""Test case to show a ipsec_site_connection."""

3
neutron/tests/unit/ml2/drivers/cisco/nexus/test_cisco_mech.py
View File

@ -268,6 +268,8 @@ class TestCiscoPortsV2(CiscoML2MechanismTestCase,
yield req.get_response(self.api)
if expected_failure:
self._create_deviceowner_mock()
self._delete('ports', port['port']['id'])
self._delete('networks', network['network']['id'])
def _assertExpectedHTTP(self, status, exc):
"""Confirm that an HTTP status corresponds to an expected exception.
@ -480,6 +482,7 @@ class TestCiscoPortsV2(CiscoML2MechanismTestCase,
add_keyword_expected=False))
self.mock_ncclient.reset_mock()
yield
self._delete('ports', port['port']['id'])
# Create network and subnet
with self.network(name=NETWORK_NAME) as network:

4
neutron/tests/unit/ml2/drivers/test_l2population.py
View File

@ -522,7 +522,7 @@ class TestL2PopulationRpcTestCase(test_plugin.NeutronDbPluginV2TestCase):
self.callbacks.update_device_up(self.adminContext,
agent_id=HOST,
device=device1)
self._delete('ports', port2['port']['id'])
p2_ips = [p['ip_address'] for p in p2['fixed_ips']]
expected = {'args':
{'fdb_entries':
@ -558,7 +558,7 @@ class TestL2PopulationRpcTestCase(test_plugin.NeutronDbPluginV2TestCase):
self.callbacks.update_device_up(self.adminContext,
agent_id=HOST,
device=device)
self._delete('ports', port['port']['id'])
p1_ips = [p['ip_address'] for p in p1['fixed_ips']]
expected = {'args':
{'fdb_entries':

2
neutron/tests/unit/mlnx/test_mlnx_plugin.py
View File

@ -93,6 +93,8 @@ class TestMlnxPortBinding(MlnxPluginV2TestCase,
**profile_arg) as port:
self._check_default_port_binding_profole(
port, expected_vif_type=vif_type)
self._delete('ports', port['port']['id'])
self._delete('networks', port['port']['network_id'])
def test_create_port_binding_profile_with_empty_dict(self):
profile_arg = {portbindings.PROFILE: {}}

30
neutron/tests/unit/nec/test_nec_plugin.py
View File

@ -169,6 +169,7 @@ class TestNecPortsV2Callback(NecPluginV2TestCase):
self.rpcapi_update_ports(removed=[port_id])
self.assertEqual(self.ofc.delete_ofc_port.call_count, 1)
self.assertIsNone(self._get_portinfo(port_id))
self._delete('ports', port['port']['id'])
# The port and portinfo is expected to delete when exiting with-clause.
self.assertEqual(self.ofc.delete_ofc_port.call_count, 1)
@ -216,6 +217,7 @@ class TestNecPortsV2Callback(NecPluginV2TestCase):
self.assertEqual(self.ofc.delete_ofc_port.call_count, 1)
self.assertEqual(ndb.get_portinfo(self.context.session,
port_id).port_no, 456)
self._delete('ports', port['port']['id'])
if not portinfo_change_first:
# The port is expected to delete when exiting with-clause.
@ -355,6 +357,7 @@ class TestNecPluginOfcManager(NecPluginV2TestCase):
with self.network() as network:
net = network['network']
self.assertEqual(network['network']['status'], 'ACTIVE')
self._delete('networks', network['network']['id'])
expected = [
mock.call.exists_ofc_tenant(ctx, self._tenant_id),
@ -374,6 +377,7 @@ class TestNecPluginOfcManager(NecPluginV2TestCase):
with self.network(admin_state_up=False) as network:
net = network['network']
self.assertEqual(network['network']['status'], 'DOWN')
self._delete('networks', network['network']['id'])
expected = [
mock.call.exists_ofc_tenant(ctx, self._tenant_id),
@ -396,7 +400,8 @@ class TestNecPluginOfcManager(NecPluginV2TestCase):
with self.network() as net2:
nets.append(net2['network'])
self.assertEqual(net2['network']['status'], 'ACTIVE')
self._delete('networks', net2['network']['id'])
self._delete('networks', net1['network']['id'])
expected = [
mock.call.exists_ofc_tenant(ctx, self._tenant_id),
mock.call.create_ofc_tenant(ctx, self._tenant_id),
@ -424,7 +429,7 @@ class TestNecPluginOfcManager(NecPluginV2TestCase):
# tearDown(). When OFCManager has failed to create a network on OFC,
# it does not keeps ofc_network entry and will fail to delete this
# network from OFC. Deletion of network is not the scope of this test.
with self.network(do_delete=False) as network:
with self.network() as network:
net = network['network']
self.assertEqual(net['status'], 'ERROR')
net_ref = self._show('networks', net['id'])
@ -465,6 +470,7 @@ class TestNecPluginOfcManager(NecPluginV2TestCase):
net_ref = self._show('networks', net['id'])
self.assertEqual(net_ref['network']['status'], 'ACTIVE')
self._delete('networks', network['network']['id'])
expected = [
mock.call.exists_ofc_tenant(ctx, self._tenant_id),
@ -492,7 +498,8 @@ class TestNecPluginOfcManager(NecPluginV2TestCase):
p1_ref = self._show('ports', p1['id'])
self.assertEqual(p1_ref['port']['status'], 'DOWN')
self._delete('ports', port['port']['id'])
self._delete('networks', port['port']['network_id'])
expected = [
mock.call.exists_ofc_tenant(ctx, self._tenant_id),
mock.call.create_ofc_tenant(ctx, self._tenant_id),
@ -532,6 +539,8 @@ class TestNecPluginOfcManager(NecPluginV2TestCase):
p1_ref = self._show('ports', p1['id'])
self.assertEqual(p1_ref['port']['status'], 'ACTIVE')
self._delete('ports', port['port']['id'])
self._delete('networks', port['port']['network_id'])
expected = [
mock.call.exists_ofc_tenant(ctx, self._tenant_id),
@ -567,6 +576,7 @@ class TestNecPluginOfcManager(NecPluginV2TestCase):
self.rpcapi_update_ports(added=[portinfo])
# In a case of dhcp port, the port is deleted automatically
# when delete_network.
self._delete('networks', network['network']['id'])
expected = [
mock.call.exists_ofc_tenant(ctx, self._tenant_id),
@ -592,7 +602,7 @@ class TestNecPluginOfcManager(NecPluginV2TestCase):
net_id = net['network']['id']
net_ref = self._show('networks', net_id)
self.assertEqual(net_ref['network']['status'], 'ERROR')
self._delete('networks', net['network']['id'])
ctx = mock.ANY
tenant_id = self._tenant_id
net_name = mock.ANY
@ -622,6 +632,7 @@ class TestNecPluginOfcManager(NecPluginV2TestCase):
self.assertEqual(net_ref['network']['status'], 'ERROR')
self.ofc.set_raise_exc('delete_ofc_network', None)
self._delete('networks', net['network']['id'])
ctx = mock.ANY
tenant = mock.ANY
@ -641,7 +652,7 @@ class TestNecPluginOfcManager(NecPluginV2TestCase):
self.ofc.set_raise_exc('delete_ofc_port',
nexc.OFCException(reason='hoge'))
with self.network(do_delete=False) as net:
with self.network() as net:
net_id = net['network']['id']
device_owner = db_base_plugin_v2.AUTO_DELETE_PORT_OWNERS[0]
@ -728,6 +739,8 @@ class TestNecPluginOfcManager(NecPluginV2TestCase):
{'admin_state_up': False})
self.assertEqual(res['status'], 'DOWN')
self.assertEqual(self.ofc.delete_ofc_port.call_count, 1)
self._delete('ports', port['port']['id'])
self._delete('networks', port['port']['network_id'])
expected = [
mock.call.exists_ofc_tenant(ctx, self._tenant_id),
@ -777,7 +790,7 @@ class TestNecPluginOfcManager(NecPluginV2TestCase):
self.assertEqual(res['port']['status'], 'ACTIVE')
port_ref = self._show('ports', port_id)
self.assertEqual(port_ref['port']['status'], 'ACTIVE')
self._delete('ports', port['port']['id'])
ctx = mock.ANY
port = mock.ANY
expected = [
@ -820,6 +833,7 @@ class TestNecPluginOfcManager(NecPluginV2TestCase):
self.assertEqual(res['port']['status'], 'DOWN')
port_ref = self._show('ports', port_id)
self.assertEqual(port_ref['port']['status'], 'DOWN')
self._delete('ports', port['port']['id'])
ctx = mock.ANY
port = mock.ANY
@ -845,6 +859,7 @@ class TestNecPluginOfcManager(NecPluginV2TestCase):
self.rpcapi_update_ports(added=[portinfo])
port_ref = self._show('ports', port_id)
self.assertEqual(port_ref['port']['status'], 'ERROR')
self._delete('ports', port['port']['id'])
ctx = mock.ANY
port = mock.ANY
@ -873,6 +888,7 @@ class TestNecPluginOfcManager(NecPluginV2TestCase):
self.assertEqual(port_ref['port']['status'], 'ERROR')
self.ofc.set_raise_exc('delete_ofc_port', None)
self._delete('ports', port['port']['id'])
ctx = mock.ANY
port = mock.ANY
@ -890,7 +906,7 @@ class TestNecPluginOfcManager(NecPluginV2TestCase):
def _test_delete_port_for_disappeared_ofc_port(self, raised_exc):
self.ofc.set_raise_exc('delete_ofc_port', raised_exc)
with self.port(do_delete=False) as port:
with self.port() as port:
port_id = port['port']['id']
portinfo = {'id': port_id, 'port_no': 123}

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

@ -100,6 +100,8 @@ class TestNecPluginPacketFilterBase(test_nec_plugin.NecPluginV2TestCase):
yield pf
if do_delete:
self._delete('packet_filters', pf['packet_filter']['id'])
if not network:
self._delete('networks', network_to_use['network']['id'])
@contextlib.contextmanager
def packet_filter_on_port(self, port=None, fmt=None, do_delete=True,
@ -552,7 +554,7 @@ class TestNecPluginPacketFilter(TestNecPluginPacketFilterBase):
expected_code=webob.exc.HTTPNotFound.code)
def test_auto_delete_pf_in_port_deletion(self):
with self.port(do_delete=False) as port:
with self.port() as port:
network = self._show('networks', port['port']['network_id'])
with self.packet_filter_on_network(network=network) as pfn:

3
neutron/tests/unit/nuage/test_nuage_plugin.py
View File

@ -248,6 +248,7 @@ class NuagePluginV2TestCase(test_db_plugin.NeutronDbPluginV2TestCase):
self.assertEqual(
fip2['floatingip']['port_id'],
body['floatingip']['port_id'])
self._delete('ports', p['port']['id'])
# Test that port has been successfully deleted.
body = self._show('ports', p['port']['id'],
@ -430,7 +431,7 @@ class TestNuageExtrarouteTestCase(NuagePluginV2TestCase,
def test_router_update_with_dup_destination_address(self):
with self.router() as r:
with self.subnet(cidr='10.0.1.0/24') as s:
with self.port(subnet=s, do_delete=False) as p:
with self.port(subnet=s) as p:
self._router_interface_action('add',
r['router']['id'],
None,

24
neutron/tests/unit/openvswitch/test_agent_scheduler.py
View File

@ -464,12 +464,15 @@ class OvsAgentSchedulerTestCase(OvsAgentSchedulerTestCaseBase):
with self.port() as port1:
dhcp_agents = self._list_dhcp_agents_hosting_network(
port1['port']['network_id'])
self._delete('ports', port1['port']['id'])
self._delete('networks', port1['port']['network_id'])
self.assertEqual(1, len(dhcp_agents['agents']))
agents = self._list_agents()
self._disable_agent(agents['agents'][0]['id'])
with self.port() as port2:
dhcp_agents = self._list_dhcp_agents_hosting_network(
port2['port']['network_id'])
self._delete('ports', port2['port']['id'])
self.assertEqual(0, len(dhcp_agents['agents']))
def test_network_scheduler_with_down_agent(self):
@ -488,12 +491,15 @@ class OvsAgentSchedulerTestCase(OvsAgentSchedulerTestCaseBase):
with self.port() as port:
dhcp_agents = self._list_dhcp_agents_hosting_network(
port['port']['network_id'])
self._delete('ports', port['port']['id'])
self._delete('networks', port['port']['network_id'])
self.assertEqual(1, len(dhcp_agents['agents']))
with mock.patch(is_agent_down_str) as mock_is_agent_down:
mock_is_agent_down.return_value = True
with self.port() as port:
dhcp_agents = self._list_dhcp_agents_hosting_network(
port['port']['network_id'])
self._delete('ports', port['port']['id'])
self.assertEqual(0, len(dhcp_agents['agents']))
def test_network_scheduler_with_hosted_network(self):
@ -517,13 +523,12 @@ class OvsAgentSchedulerTestCase(OvsAgentSchedulerTestCaseBase):
mock_hosting_agents.return_value = plugin.get_agents_db(
self.adminContext)
with self.network('test', do_delete=False) as net1:
with self.network('test') as net1:
pass
with self.subnet(network=net1,
cidr='10.0.1.0/24',
do_delete=False) as subnet1:
cidr='10.0.1.0/24') as subnet1:
pass
with self.port(subnet=subnet1, do_delete=False) as port2:
with self.port(subnet=subnet1) as port2:
pass
dhcp_agents = self._list_dhcp_agents_hosting_network(
port2['port']['network_id'])
@ -1149,7 +1154,7 @@ class OvsDhcpAgentNotifierTestCase(test_l3_plugin.L3NatTestCaseMixin,
self._assert_notify(notifications, expected_event_type)
def test_network_remove_from_dhcp_agent_notification(self):
with self.network(do_delete=False) as net1:
with self.network() as net1:
network_id = net1['network']['id']
self._register_agent_states()
hosta_id = self._get_agent_id(constants.AGENT_TYPE_DHCP,
@ -1191,18 +1196,15 @@ class OvsDhcpAgentNotifierTestCase(test_l3_plugin.L3NatTestCaseMixin,
'configurations': {'dhcp_driver': 'dhcp_driver',
'use_namespaces': True, },
'agent_type': constants.AGENT_TYPE_DHCP})
with self.network(do_delete=False) as net1:
with self.network() as net1:
with self.subnet(network=net1,
gateway_ip=gateway,
do_delete=False) as subnet1:
gateway_ip=gateway) as subnet1:
if owner:
with self.port(subnet=subnet1,
do_delete=False,
device_owner=owner) as port:
return [net1, subnet1, port]
else:
with self.port(subnet=subnet1,
do_delete=False) as port:
with self.port(subnet=subnet1) as port:
return [net1, subnet1, port]
def _notification_mocks(self, hosts, net, subnet, port):

2
neutron/tests/unit/services/loadbalancer/drivers/embrane/test_plugin_driver.py
View File

@ -58,7 +58,7 @@ class TestLoadBalancerPlugin(test_db_loadbalancer.TestLoadBalancer,
self.skip("App cookie persistence not supported.")
def test_pool_port(self):
with self.port(do_delete=False) as port:
with self.port() as port:
with self.pool() as pool:
h_db.add_pool_port(context.get_admin_context(),
pool['pool']['id'], port['port']['id'])

8
neutron/tests/unit/services/loadbalancer/drivers/radware/test_plugin_driver.py
View File

@ -215,8 +215,8 @@ class TestLoadBalancerPlugin(TestLoadBalancerPluginBase):
def test_create_vip_failure(self):
"""Test the rest call failure handling by Exception raising."""
with self.network(do_delete=False) as network:
with self.subnet(network=network, do_delete=False) as subnet:
with self.network() as network:
with self.subnet(network=network) as subnet:
with self.pool(do_delete=False,
provider='radware',
subnet_id=subnet['subnet']['id']) as pool:
@ -495,8 +495,8 @@ class TestLoadBalancerPlugin(TestLoadBalancerPluginBase):
def test_delete_vip_failure(self):
plugin = self.plugin_instance
with self.network(do_delete=False) as network:
with self.subnet(network=network, do_delete=False) as subnet:
with self.network() as network:
with self.subnet(network=network) as subnet:
with self.pool(do_delete=False,
provider='radware',
subnet_id=subnet['subnet']['id']) as pool:

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

@ -518,13 +518,10 @@ class NeutronDbPluginV2TestCase(testlib_api.WebTestCase,
def network(self, name='net1',
admin_state_up=True,
fmt=None,
do_delete=True,
**kwargs):
network = self._make_network(fmt or self.fmt, name,
admin_state_up, **kwargs)
yield network
if do_delete:
self._delete('networks', network['network']['id'])
@contextlib.contextmanager
def subnet(self, network=None,
@ -537,7 +534,6 @@ class NeutronDbPluginV2TestCase(testlib_api.WebTestCase,
dns_nameservers=None,
host_routes=None,
shared=None,
do_delete=True,
ipv6_ra_mode=None,
ipv6_address_mode=None):
with optional_ctx(network, self.network) as network_to_use:
@ -554,18 +550,13 @@ class NeutronDbPluginV2TestCase(testlib_api.WebTestCase,
ipv6_ra_mode=ipv6_ra_mode,
ipv6_address_mode=ipv6_address_mode)
yield subnet
if do_delete:
self._delete('subnets', subnet['subnet']['id'])
@contextlib.contextmanager
def port(self, subnet=None, fmt=None, do_delete=True,
**kwargs):
def port(self, subnet=None, fmt=None, **kwargs):
with optional_ctx(subnet, self.subnet) as subnet_to_use:
net_id = subnet_to_use['subnet']['network_id']
port = self._make_port(fmt or self.fmt, net_id, **kwargs)
yield port
if do_delete:
self._delete('ports', port['port']['id'])
def _test_list_with_sort(self, resource,
items, sorts, resources=None, query_params=''):
@ -783,7 +774,7 @@ class TestPortsV2(NeutronDbPluginV2TestCase):
self.assertEqual('myname', port['port']['name'])
def test_create_port_as_admin(self):
with self.network(do_delete=False) as network:
with self.network() as network:
self._create_port(self.fmt,
network['network']['id'],
webob.exc.HTTPCreated.code,
@ -1058,7 +1049,7 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s
self.assertEqual(port['port']['id'], sport['port']['id'])
def test_delete_port(self):
with self.port(do_delete=False) as port:
with self.port() as port:
self._delete('ports', port['port']['id'])
self._show('ports', port['port']['id'],
expected_code=webob.exc.HTTPNotFound.code)
@ -1684,8 +1675,8 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s
ctx = context.get_admin_context()
with self.subnet() as subnet:
with contextlib.nested(
self.port(subnet=subnet, device_id='owner1', do_delete=False),
self.port(subnet=subnet, device_id='owner1', do_delete=False),
self.port(subnet=subnet, device_id='owner1'),
self.port(subnet=subnet, device_id='owner1'),
self.port(subnet=subnet, device_id='owner2'),
) as (p1, p2, p3):
network_id = subnet['subnet']['network_id']
@ -1702,7 +1693,7 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s
ctx = context.get_admin_context()
with self.subnet() as subnet:
with contextlib.nested(
self.port(subnet=subnet, device_id='owner1', do_delete=False),
self.port(subnet=subnet, device_id='owner1'),
self.port(subnet=subnet, device_id='owner1'),
self.port(subnet=subnet, device_id='owner2'),
) as (p1, p2, p3):
@ -2296,7 +2287,8 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
sorted(expected[k]))
else:
self.assertEqual(subnet['subnet'][k], expected[k])
return subnet
self._delete('subnets', subnet['subnet']['id'])
return subnet
def test_create_subnet(self):
gateway_ip = '10.0.0.1'
@ -2438,6 +2430,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
res = self._create_subnet_bulk(self.fmt, 2,
net['network']['id'],
'test')
self._delete('networks', net['network']['id'])
# We expect a 500 as we injected a fault in the plugin
self._validate_behavior_on_bulk_failure(
res, 'subnets', webob.exc.HTTPServerError.code
@ -2541,8 +2534,8 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
with self.network() as network:
with contextlib.nested(
self.subnet(network=network),
self.subnet(network=network, cidr='10.0.1.0/24',
do_delete=False)) as (subnet1, subnet2):
self.subnet(network=network, cidr='10.0.1.0/24'),
) as (subnet1, subnet2):
subnet1_id = subnet1['subnet']['id']
subnet2_id = subnet2['subnet']['id']
with self.port(
@ -2578,7 +2571,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
set_context=True)
def test_create_subnet_as_admin(self):
with self.network(do_delete=False) as network:
with self.network() as network:
self._create_subnet(self.fmt,
network['network']['id'],
'10.0.2.0/24',
@ -3849,6 +3842,16 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
'max_subnet_host_routes',
n_exc.HostRoutesExhausted)
def test_port_prevents_network_deletion(self):
with self.port() as p:
self._delete('networks', p['port']['network_id'],
expected_code=webob.exc.HTTPConflict.code)
def test_port_prevents_subnet_deletion(self):
with self.port() as p:
self._delete('subnets', p['port']['fixed_ips'][0]['subnet_id'],
expected_code=webob.exc.HTTPConflict.code)
class DbModelTestCase(base.BaseTestCase):
"""DB model tests."""

2
neutron/tests/unit/test_extension_ext_net.py
View File

@ -174,7 +174,7 @@ class ExtNetDBTestCase(test_db_plugin.NeutronDbPluginV2TestCase):
'get_service_plugins') as srv_plugins:
l3_mock = mock.Mock()
srv_plugins.return_value = {'L3_ROUTER_NAT': l3_mock}
with self.network(do_delete=False) as net:
with self.network() as net:
req = self.new_delete_request('networks', net['network']['id'])
res = req.get_response(self.api)
self.assertEqual(res.status_int, exc.HTTPNoContent.code)

28
neutron/tests/unit/test_extension_extraroute.py
View File

@ -76,7 +76,7 @@ class ExtraRouteDBTestCaseBase(object):
routes = [{'destination': '135.207.0.0/16', 'nexthop': '10.0.1.3'}]
with self.router() as r:
with self.subnet(cidr='10.0.1.0/24') as s:
with self.port(subnet=s, do_delete=False) as p:
with self.port(subnet=s) as p:
body = self._routes_update_prepare(r['router']['id'],
None, p['port']['id'],
routes)
@ -93,7 +93,7 @@ class ExtraRouteDBTestCaseBase(object):
'nexthop': '10.0.1.5'}]
with self.router() as r:
with self.subnet(cidr='10.0.1.0/24') as s:
with self.port(subnet=s, do_delete=False) as p:
with self.port(subnet=s) as p:
self._routes_update_prepare(r['router']['id'],
None, p['port']['id'], routes)
body = self._update('routers', r['router']['id'],
@ -107,7 +107,7 @@ class ExtraRouteDBTestCaseBase(object):
'nexthop': '10.0.1.3'}]
with self.router() as r:
with self.subnet(cidr='10.0.1.0/24') as s:
with self.port(subnet=s, do_delete=False) as p:
with self.port(subnet=s) as p:
body = self._routes_update_prepare(r['router']['id'],
None, p['port']['id'],
routes)
@ -131,7 +131,7 @@ class ExtraRouteDBTestCaseBase(object):
'nexthop': '10.0.1.5'}]
with self.router() as r:
with self.subnet(cidr='10.0.1.0/24') as s:
with self.port(subnet=s, do_delete=False) as p:
with self.port(subnet=s) as p:
body = self._routes_update_prepare(r['router']['id'],
None, p['port']['id'],
routes)
@ -150,8 +150,8 @@ class ExtraRouteDBTestCaseBase(object):
self.router(),
self.subnet(cidr='10.0.0.0/24')) as (r1, r2, s):
with contextlib.nested(
self.port(subnet=s, do_delete=False),
self.port(subnet=s, do_delete=False)) as (p1, p2):
self.port(subnet=s),
self.port(subnet=s)) as (p1, p2):
body = self._routes_update_prepare(r1['router']['id'],
None, p1['port']['id'],
routes1)
@ -180,7 +180,7 @@ class ExtraRouteDBTestCaseBase(object):
'nexthop': '10.0.1.5'}]
with self.router() as r:
with self.subnet(cidr='10.0.1.0/24') as s:
with self.port(subnet=s, do_delete=False) as p:
with self.port(subnet=s) as p:
body = self._routes_update_prepare(r['router']['id'],
None, p['port']['id'],
routes_orig)
@ -198,7 +198,7 @@ class ExtraRouteDBTestCaseBase(object):
def _test_malformed_route(self, routes):
with self.router() as r:
with self.subnet(cidr='10.0.1.0/24') as s:
with self.port(subnet=s, do_delete=False) as p:
with self.port(subnet=s) as p:
self._router_interface_action('add',
r['router']['id'],
None,
@ -230,7 +230,7 @@ class ExtraRouteDBTestCaseBase(object):
def test_nexthop_is_port_ip(self):
with self.router() as r:
with self.subnet(cidr='10.0.1.0/24') as s:
with self.port(subnet=s, do_delete=False) as p:
with self.port(subnet=s) as p:
self._router_interface_action('add',
r['router']['id'],
None,
@ -252,7 +252,7 @@ class ExtraRouteDBTestCaseBase(object):
def test_router_update_with_too_many_routes(self):
with self.router() as r:
with self.subnet(cidr='10.0.1.0/24') as s:
with self.port(subnet=s, do_delete=False) as p:
with self.port(subnet=s) as p:
self._router_interface_action('add',
r['router']['id'],
None,
@ -281,7 +281,7 @@ class ExtraRouteDBTestCaseBase(object):
def test_router_update_with_dup_address(self):
with self.router() as r:
with self.subnet(cidr='10.0.1.0/24') as s:
with self.port(subnet=s, do_delete=False) as p:
with self.port(subnet=s) as p:
self._router_interface_action('add',
r['router']['id'],
None,
@ -306,7 +306,7 @@ class ExtraRouteDBTestCaseBase(object):
def test_router_update_with_invalid_ip_address(self):
with self.router() as r:
with self.subnet(cidr='10.0.1.0/24') as s:
with self.port(subnet=s, do_delete=False) as p:
with self.port(subnet=s) as p:
self._router_interface_action('add',
r['router']['id'],
None,
@ -345,7 +345,7 @@ class ExtraRouteDBTestCaseBase(object):
def test_router_update_with_invalid_nexthop_ip(self):
with self.router() as r:
with self.subnet(cidr='10.0.1.0/24') as s:
with self.port(subnet=s, do_delete=False) as p:
with self.port(subnet=s) as p:
self._router_interface_action('add',
r['router']['id'],
None,
@ -368,7 +368,7 @@ class ExtraRouteDBTestCaseBase(object):
def test_router_update_with_nexthop_is_outside_port_subnet(self):
with self.router() as r:
with self.subnet(cidr='10.0.1.0/24') as s:
with self.port(subnet=s, do_delete=False) as p:
with self.port(subnet=s) as p:
self._router_interface_action('add',
r['router']['id'],
None,

4
neutron/tests/unit/test_extension_portsecurity.py
View File

@ -371,8 +371,8 @@ class TestPortSecurity(PortSecurityDBTestCase):
self.assertEqual(res.status_int, 403)
def test_update_port_security_off_shared_network(self):
with self.network(shared=True, do_delete=False) as net:
with self.subnet(network=net, do_delete=False):
with self.network(shared=True) as net:
with self.subnet(network=net):
res = self._create_port('json', net['network']['id'],
tenant_id='not_network_owner',
set_context=True)

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

@ -833,7 +833,7 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
def test_router_add_interface_port(self):
with self.router() as r:
with self.port(do_delete=False) as p:
with self.port() as p:
body = self._router_interface_action('add',
r['router']['id'],
None,
@ -865,7 +865,7 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
'roles': []}
tdict.return_value = admin_context
with self.router() as r:
with self.port(do_delete=False) as p:
with self.port() as p:
tdict.return_value = tenant_context
err_code = exc.HTTPNotFound.code
self._router_interface_action('add',
@ -914,7 +914,7 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
def test_router_add_interface_dup_subnet2_returns_400(self):
with self.router() as r:
with self.subnet() as s:
with self.port(subnet=s, do_delete=False) as p1:
with self.port(subnet=s) as p1:
with self.port(subnet=s) as p2:
self._router_interface_action('add',
r['router']['id'],
@ -1136,7 +1136,7 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
def test_router_remove_interface_wrong_subnet_returns_400(self):
with self.router() as r:
with self.subnet() as s:
with self.port(do_delete=False) as p:
with self.port() as p:
self._router_interface_action('add',
r['router']['id'],
None,
@ -1154,7 +1154,7 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
def test_router_remove_interface_returns_200(self):
with self.router() as r:
with self.port(do_delete=False) as p:
with self.port() as p:
body = self._router_interface_action('add',
r['router']['id'],
None,
@ -1168,7 +1168,7 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
def test_router_remove_interface_wrong_port_returns_404(self):
with self.router() as r:
with self.subnet():
with self.port(do_delete=False) as p:
with self.port() as p:
self._router_interface_action('add',
r['router']['id'],
None,
@ -1410,7 +1410,7 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
self.assertEqual(
body['floatingip']['port_id'],
fip2['floatingip']['port_id'])
self._delete('ports', p['port']['id'])
# Test that port has been successfully deleted.
body = self._show('ports', p['port']['id'],
expected_code=exc.HTTPNotFound.code)
@ -1525,6 +1525,7 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
# note: once this port goes out of scope, the port will be
# deleted, which is what we want to test. We want to confirm
# that the fields are set back to None
self._delete('ports', p['port']['id'])
body = self._show('floatingips', fip['floatingip']['id'])
self.assertEqual(body['floatingip']['id'],
fip['floatingip']['id'])
@ -1770,7 +1771,7 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
with self.network() as net:
net_id = net['network']['id']
self._set_net_external(net_id)
with self.subnet(network=net, do_delete=False):
with self.subnet(network=net):
self._make_floatingip(self.fmt, net_id)
@ -1780,7 +1781,7 @@ class L3AgentDbTestCaseBase(L3NatTestCaseMixin):
def test_l3_agent_routers_query_interfaces(self):
with self.router() as r:
with self.port(do_delete=False) as p:
with self.port() as p:
self._router_interface_action('add',
r['router']['id'],
None,
@ -1804,7 +1805,6 @@ class L3AgentDbTestCaseBase(L3NatTestCaseMixin):
with self.router() as r:
with self.subnet(cidr='9.0.1.0/24') as subnet:
with self.port(subnet=subnet,
do_delete=False,
fixed_ips=[{'ip_address': '9.0.1.3'}]) as p:
self._router_interface_action('add',
r['router']['id'],
@ -1892,7 +1892,7 @@ class L3AgentDbTestCaseBase(L3NatTestCaseMixin):
self._test_notify_op_agent(self._test_router_gateway_op_agent)
def _test_interfaces_op_agent(self, r, notifyApi):
with self.port(do_delete=False) as p:
with self.port() as p:
self._router_interface_action('add',
r['router']['id'],
None,

4
neutron/tests/unit/vmware/test_nsx_plugin.py
View File

@ -937,14 +937,14 @@ class TestL3NatTestCase(L3NatTest,
subnets = self._list('subnets')['subnets']
with self.subnet() as s:
with self.port(subnet=s, device_id='1234',
device_owner=constants.DEVICE_OWNER_DHCP):
device_owner=constants.DEVICE_OWNER_DHCP) as port:
subnets = self._list('subnets')['subnets']
self.assertEqual(len(subnets), 1)
self.assertEqual(subnets[0]['host_routes'][0]['nexthop'],
'10.0.0.2')
self.assertEqual(subnets[0]['host_routes'][0]['destination'],
'169.254.169.254/32')
self._delete('ports', port['port']['id'])
subnets = self._list('subnets')['subnets']
# Test that route is deleted after dhcp port is removed
self.assertEqual(len(subnets[0]['host_routes']), 0)