openstack/neutron
Code Issues Proposed changes

Merge "UTs: Disable auto deletion of ports/subnets/nets"

Browse Source
This commit is contained in:
Jenkins 2014-09-12 12:51:41 +00:00 committed by Gerrit Code Review
commit 7c86ec9b73
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)