openstack
/
neutron
Code Issues Proposed changes

Merge "Raise error upon deleting subnet with router ports"

This commit is contained in:
Jenkins 2015-03-09 16:43:05 +00:00 committed by Gerrit Code Review
parent ebc1469a7e e80ab8113e
commit 4663a15efd
4 changed files with 90 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
neutron/db/db_base_plugin_v2.py
View File

@ -1221,6 +1221,20 @@ class NeutronDbPluginV2(neutron_plugin_base_v2.NeutronPluginBaseV2,
models_v2.IPAllocation).filter_by(
subnet_id=subnet_id).join(models_v2.Port).first()
def _subnet_check_ip_allocations_internal_router_ports(self, context,
subnet_id):
# Do not delete the subnet if IP allocations for internal
# router ports still exist
allocs = context.session.query(models_v2.IPAllocation).filter_by(
subnet_id=subnet_id).join(models_v2.Port).filter(
models_v2.Port.device_owner.in_(
constants.ROUTER_INTERFACE_OWNERS)
).first()
if allocs:
LOG.debug("Subnet %s still has internal router ports, "
"cannot delete", subnet_id)
raise n_exc.SubnetInUse(subnet_id=id)
def delete_subnet(self, context, id):
with context.session.begin(subtransactions=True):
subnet = self._get_subnet(context, id)
@ -1233,7 +1247,10 @@ class NeutronDbPluginV2(neutron_plugin_base_v2.NeutronPluginBaseV2,
# for IPv6 addresses which were automatically generated
# via SLAAC
is_auto_addr_subnet = ipv6_utils.is_auto_address_subnet(subnet)
if not is_auto_addr_subnet:
if is_auto_addr_subnet:
self._subnet_check_ip_allocations_internal_router_ports(
context, id)
else:
qry_network_ports = (
qry_network_ports.filter(models_v2.Port.device_owner.
in_(AUTO_DELETE_PORT_OWNERS)))

5
neutron/plugins/ml2/plugin.py
View File

@ -842,7 +842,10 @@ class Ml2Plugin(db_base_plugin_v2.NeutronDbPluginV2,
# Remove network owned ports, and delete IP allocations
# for IPv6 addresses which were automatically generated
# via SLAAC
if not is_auto_addr_subnet:
if is_auto_addr_subnet:
self._subnet_check_ip_allocations_internal_router_ports(
context, id)
else:
qry_allocated = (
qry_allocated.filter(models_v2.Port.device_owner.
in_(db_base_plugin_v2.AUTO_DELETE_PORT_OWNERS)))

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

@ -2822,17 +2822,21 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
self.assertEqual(res.status_int,
webob.exc.HTTPNoContent.code)
def test_delete_subnet_ipv6_slaac_port_exists(self):
"""Test IPv6 SLAAC subnet delete when a port is still using subnet."""
def _create_slaac_subnet_and_port(self, port_owner=None):
# Create an IPv6 SLAAC subnet and a port using that subnet
res = self._create_network(fmt=self.fmt, name='net',
admin_state_up=True)
network = self.deserialize(self.fmt, res)
# Create an IPv6 SLAAC subnet and a port using that subnet
subnet = self._make_subnet(self.fmt, network, gateway='fe80::1',
cidr='fe80::/64', ip_version=6,
ipv6_ra_mode=constants.IPV6_SLAAC,
ipv6_address_mode=constants.IPV6_SLAAC)
res = self._create_port(self.fmt, net_id=network['network']['id'])
if port_owner:
res = self._create_port(self.fmt, net_id=network['network']['id'],
device_owner=port_owner)
else:
res = self._create_port(self.fmt, net_id=network['network']['id'])
port = self.deserialize(self.fmt, res)
self.assertEqual(1, len(port['port']['fixed_ips']))
@ -2842,6 +2846,11 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
sport = self.deserialize(self.fmt, req.get_response(self.api))
self.assertEqual(1, len(sport['port']['fixed_ips']))
return subnet, port
def test_delete_subnet_ipv6_slaac_port_exists(self):
"""Test IPv6 SLAAC subnet delete when a port is still using subnet."""
subnet, port = self._create_slaac_subnet_and_port()
# Delete the subnet
req = self.new_delete_request('subnets', subnet['subnet']['id'])
res = req.get_response(self.api)
@ -2852,6 +2861,29 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
sport = self.deserialize(self.fmt, req.get_response(self.api))
self.assertEqual(0, len(sport['port']['fixed_ips']))
def test_delete_subnet_ipv6_slaac_router_port_exists(self):
"""Test IPv6 SLAAC subnet delete with a router port using the subnet"""
subnet, port = self._create_slaac_subnet_and_port(
constants.DEVICE_OWNER_ROUTER_INTF)
# Delete the subnet and assert that we get a HTTP 409 error
req = self.new_delete_request('subnets', subnet['subnet']['id'])
res = req.get_response(self.api)
self.assertEqual(webob.exc.HTTPConflict.code, res.status_int)
# The subnet should still exist and the port should still have an
# address from the subnet
req = self.new_show_request('subnets', subnet['subnet']['id'],
self.fmt)
res = req.get_response(self.api)
ssubnet = self.deserialize(self.fmt, req.get_response(self.api))
self.assertIsNotNone(ssubnet)
req = self.new_show_request('ports', port['port']['id'], self.fmt)
res = req.get_response(self.api)
sport = self.deserialize(self.fmt, req.get_response(self.api))
self.assertEqual(1, len(sport['port']['fixed_ips']))
port_subnet_ids = [fip['subnet_id'] for fip in
sport['port']['fixed_ips']]
self.assertIn(subnet['subnet']['id'], port_subnet_ids)
def test_delete_network(self):
gateway_ip = '10.0.0.1'
cidr = '10.0.0.0/24'

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

@ -2023,21 +2023,41 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
break
self.assertTrue(found)
def _test_router_delete_subnet_inuse_returns_409(self, router, subnet):
r, s = router, subnet
self._router_interface_action('add',
r['router']['id'],
s['subnet']['id'],
None)
# subnet cannot be deleted as it's attached to a router
self._delete('subnets', s['subnet']['id'],
expected_code=exc.HTTPConflict.code)
# remove interface so test can exit without errors
self._router_interface_action('remove',
r['router']['id'],
s['subnet']['id'],
None)
def _ipv6_subnet(self, mode):
return self.subnet(cidr='fd00::1/64', gateway_ip='fd00::1',
ip_version=6,
ipv6_ra_mode=mode,
ipv6_address_mode=mode)
def test_router_delete_subnet_inuse_returns_409(self):
with self.router() as r:
with self.subnet() as s:
self._router_interface_action('add',
r['router']['id'],
s['subnet']['id'],
None)
# subnet cannot be delete as it's attached to a router
self._delete('subnets', s['subnet']['id'],
expected_code=exc.HTTPConflict.code)
# remove interface so test can exit without errors
self._router_interface_action('remove',
r['router']['id'],
s['subnet']['id'],
None)
self._test_router_delete_subnet_inuse_returns_409(r, s)
def test_router_delete_ipv6_slaac_subnet_inuse_returns_409(self):
with self.router() as r:
with self._ipv6_subnet(l3_constants.IPV6_SLAAC) as s:
self._test_router_delete_subnet_inuse_returns_409(r, s)
def test_router_delete_dhcpv6_stateless_subnet_inuse_returns_409(self):
with self.router() as r:
with self._ipv6_subnet(l3_constants.DHCPV6_STATELESS) as s:
self._test_router_delete_subnet_inuse_returns_409(r, s)
def test_delete_ext_net_with_disassociated_floating_ips(self):
with self.network() as net: