openstack
/
neutron
Code Issues Proposed changes

Merge "Check if pool update is needed in reference driver"

This commit is contained in:
Jenkins 2016-05-18 12:13:34 +00:00 committed by Gerrit Code Review
parent 0eed94a777 310074b2d4
commit acca136721
2 changed files with 50 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
neutron/ipam/drivers/neutrondb_ipam/driver.py
View File

@ -370,11 +370,20 @@ class NeutronDbSubnet(ipam_base.Subnet):
subnet_id=self.subnet_manager.neutron_id,
ip_address=address)
def _no_pool_changes(self, session, pools):
"""Check if pool updates in db are required."""
db_pools = self.subnet_manager.list_pools(session)
iprange_pools = [netaddr.IPRange(pool.first_ip, pool.last_ip)
for pool in db_pools]
return pools == iprange_pools
def update_allocation_pools(self, pools, cidr):
# Pools have already been validated in the subnet request object which
# was sent to the subnet pool driver. Further validation should not be
# required.
session = self._context.session
if self._no_pool_changes(session, pools):
return
self.subnet_manager.delete_allocation_pools(session)
self.create_allocation_pools(self.subnet_manager, session, pools, cidr)
self._pools = pools

41
neutron/tests/unit/ipam/drivers/neutrondb_ipam/test_driver.py
View File

@ -21,6 +21,7 @@ from neutron_lib import exceptions as n_exc
from neutron.common import constants as n_const
from neutron import context
from neutron.db import api as ndb_api
from neutron.ipam.drivers.neutrondb_ipam import db_models
from neutron.ipam.drivers.neutrondb_ipam import driver
from neutron.ipam import exceptions as ipam_exc
from neutron.ipam import requests as ipam_req
@ -460,3 +461,43 @@ class TestNeutronDbIpamSubnet(testlib_api.SqlTestCase,
ipam_subnet._allocate_specific_ip(self.ctx.session, ip)
self.assertRaises(ipam_exc.IPAllocationFailed, go)
def test_update_allocation_pools_with_no_pool_change(self):
cidr = '10.0.0.0/24'
ipam_subnet = self._create_and_allocate_ipam_subnet(
cidr)[0]
ipam_subnet.subnet_manager.delete_allocation_pools = mock.Mock()
ipam_subnet.create_allocation_pools = mock.Mock()
alloc_pools = [netaddr.IPRange('10.0.0.2', '10.0.0.254')]
# Make sure allocation pools recreation does not happen in case of
# unchanged allocation pools
ipam_subnet.update_allocation_pools(alloc_pools, cidr)
self.assertFalse(
ipam_subnet.subnet_manager.delete_allocation_pools.called)
self.assertFalse(ipam_subnet.create_allocation_pools.called)
def _test__no_pool_changes(self, new_pools):
id = 'some-id'
ipam_subnet = driver.NeutronDbSubnet(id, self.ctx)
pools = [db_models.IpamAllocationPool(ipam_subnet_id=id,
first_ip='192.168.10.20',
last_ip='192.168.10.41'),
db_models.IpamAllocationPool(ipam_subnet_id=id,
first_ip='192.168.10.50',
last_ip='192.168.10.60')]
ipam_subnet.subnet_manager.list_pools = mock.Mock(return_value=pools)
return ipam_subnet._no_pool_changes(self.ctx.session, new_pools)
def test__no_pool_changes_negative(self):
pool_list = [[netaddr.IPRange('192.168.10.2', '192.168.10.254')],
[netaddr.IPRange('192.168.10.20', '192.168.10.41')],
[netaddr.IPRange('192.168.10.20', '192.168.10.41'),
netaddr.IPRange('192.168.10.51', '192.168.10.60')]]
for pools in pool_list:
self.assertFalse(self._test__no_pool_changes(pools))
def test__no_pool_changes_positive(self):
pools = [netaddr.IPRange('192.168.10.20', '192.168.10.41'),
netaddr.IPRange('192.168.10.50', '192.168.10.60')]
self.assertTrue(self._test__no_pool_changes(pools))