Bug
1111572 was filed about a failed update (PUT) on
'allocation_pools' of subnet. This is currently not allowed by the
neutron API (hence DocImpact below). Following discussion on the
bug and subsequently, it seems this is a desirable feature.
This review makes the allocation_pools attribute of subnet
updateable by PUT. The semantics are that the entire allocation
pools attribute is replaced by the provided parameter (see
provided tests for details).
Unit tests added that exercise successful update of
allocation_pools with sane params and update using erroneous
allocation_pools that fall outside the subnet cidr.
DocImpact
Closes-Bug:
1111572
Change-Id: I47a3a71d0d196b76eda46b1d960193fb60417ba9
Co-Authored-By: Robert Collins <rbtcollins@hp.com>
'default': ATTR_NOT_SPECIFIED,
'validate': {'type:ip_address_or_none': None},
'is_visible': True},
- #TODO(salvatore-orlando): Enable PUT on allocation_pools
- 'allocation_pools': {'allow_post': True, 'allow_put': False,
+ 'allocation_pools': {'allow_post': True, 'allow_put': True,
'default': ATTR_NOT_SPECIFIED,
'validate': {'type:ip_pools': None},
'is_visible': True},
return self._make_subnet_dict(subnet)
+ def _update_subnet_allocation_pools(self, context, id, s):
+ context.session.query(models_v2.IPAllocationPool).filter_by(
+ subnet_id=id).delete()
+ new_pools = [models_v2.IPAllocationPool(
+ first_ip=p['start'], last_ip=p['end'],
+ subnet_id=id) for p in s['allocation_pools']]
+ context.session.add_all(new_pools)
+ NeutronDbPluginV2._rebuild_availability_ranges(context, [s])
+ #Gather new pools for result:
+ result_pools = [{'start': pool['start'],
+ 'end': pool['end']}
+ for pool in s['allocation_pools']]
+ del s['allocation_pools']
+ return result_pools
+
def update_subnet(self, context, id, subnet):
"""Update the subnet with new info.
s = subnet['subnet']
changed_host_routes = False
changed_dns = False
+ changed_allocation_pools = False
db_subnet = self._get_subnet(context, id)
# Fill 'ip_version' and 'allocation_pools' fields with the current
# value since _validate_subnet() expects subnet spec has 'ip_version'
'nexthop': route_str.partition("_")[2]})
del s["host_routes"]
+ if "allocation_pools" in s:
+ self._validate_allocation_pools(s['allocation_pools'],
+ s['cidr'])
+ changed_allocation_pools = True
+ new_pools = self._update_subnet_allocation_pools(context,
+ id, s)
subnet = self._get_subnet(context, id)
subnet.update(s)
result = self._make_subnet_dict(subnet)
result['dns_nameservers'] = new_dns
if changed_host_routes:
result['host_routes'] = new_routes
+ if changed_allocation_pools:
+ result['allocation_pools'] = new_pools
return result
def delete_subnet(self, context, id):
available_ranges = orm.relationship(IPAvailabilityRange,
backref='ipallocationpool',
lazy="joined",
- cascade='delete')
+ cascade='all, delete-orphan')
def __repr__(self):
return "%s - %s" % (self.first_ip, self.last_ip)
class TestPlumgridPluginSubnetsV2(test_plugin.TestSubnetsV2,
PLUMgridPluginV2TestCase):
- def test_create_subnet_default_gw_conflict_allocation_pool_returns_409(
- self):
- self.skipTest("Plugin does not support Neutron allocation process")
-
- def test_create_subnet_defaults(self):
- self.skipTest("Plugin does not support Neutron allocation process")
+ _unsupported = (
+ 'test_create_subnet_default_gw_conflict_allocation_pool_returns_409',
+ 'test_create_subnet_defaults', 'test_create_subnet_gw_values',
+ 'test_update_subnet_gateway_in_allocation_pool_returns_409',
+ 'test_update_subnet_allocation_pools',
+ 'test_update_subnet_allocation_pools_invalid_pool_for_cidr')
- def test_create_subnet_gw_values(self):
- self.skipTest("Plugin does not support Neutron allocation process")
-
- def test_update_subnet_gateway_in_allocation_pool_returns_409(self):
- self.skipTest("Plugin does not support Neutron allocation process")
+ def setUp(self):
+ if self._testMethodName in self._unsupported:
+ self.skipTest("Plugin does not support Neutron allocation process")
+ super(TestPlumgridPluginSubnetsV2, self).setUp()
class TestPlumgridPluginPortBinding(PLUMgridPluginV2TestCase,
self.assertEqual(res.status_int,
webob.exc.HTTPClientError.code)
+ def test_update_subnet_allocation_pools(self):
+ """Test that we can successfully update with sane params.
+
+ This will create a subnet with specified allocation_pools
+ Then issue an update (PUT) to update these using correct
+ (i.e. non erroneous) params. Finally retrieve the updated
+ subnet and verify.
+ """
+ allocation_pools = [{'start': '192.168.0.2', 'end': '192.168.0.254'}]
+ with self.network() as network:
+ with self.subnet(network=network,
+ allocation_pools=allocation_pools,
+ cidr='192.168.0.0/24') as subnet:
+ data = {'subnet': {'allocation_pools': [
+ {'start': '192.168.0.10', 'end': '192.168.0.20'},
+ {'start': '192.168.0.30', 'end': '192.168.0.40'}]}}
+ req = self.new_update_request('subnets', data,
+ subnet['subnet']['id'])
+ #check res code but then do GET on subnet for verification
+ res = req.get_response(self.api)
+ self.assertEqual(res.status_code, 200)
+ req = self.new_show_request('subnets', subnet['subnet']['id'],
+ self.fmt)
+ res = self.deserialize(self.fmt, req.get_response(self.api))
+ self.assertEqual(len(res['subnet']['allocation_pools']), 2)
+ res_vals = res['subnet']['allocation_pools'][0].values() +\
+ res['subnet']['allocation_pools'][1].values()
+ for pool_val in ['10', '20', '30', '40']:
+ self.assertTrue('192.168.0.%s' % (pool_val) in res_vals)
+
+ #updating alloc pool to something outside subnet.cidr
+ def test_update_subnet_allocation_pools_invalid_pool_for_cidr(self):
+ """Test update alloc pool to something outside subnet.cidr.
+
+ This makes sure that an erroneous allocation_pool specified
+ in a subnet update (outside subnet cidr) will result in an error.
+ """
+ allocation_pools = [{'start': '192.168.0.2', 'end': '192.168.0.254'}]
+ with self.network() as network:
+ with self.subnet(network=network,
+ allocation_pools=allocation_pools,
+ cidr='192.168.0.0/24') as subnet:
+ data = {'subnet': {'allocation_pools': [
+ {'start': '10.0.0.10', 'end': '10.0.0.20'}]}}
+ req = self.new_update_request('subnets', data,
+ subnet['subnet']['id'])
+ res = req.get_response(self.api)
+ self.assertEqual(res.status_int,
+ webob.exc.HTTPClientError.code)
+
def test_show_subnet(self):
with self.network() as network:
with self.subnet(network=network) as subnet:
Code Review / openstack-build / neutron-build.git / commitdiff