openstack/neutron
Code Issues Proposed changes

Merge "Use normalized cidrs in address groups"

Browse Source
This commit is contained in:
Zuul 2020-10-14 05:44:59 +00:00 committed by Gerrit Code Review
commit 2b3f20b4d9
2 changed files with 34 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
neutron/db/address_group_db.py
View File

@ -41,22 +41,31 @@ class AddressGroupDbMixin(ag_ext.AddressGroupPluginBase):
raise ag_exc.AddressGroupNotFound(address_group_id=id)
return obj
def _dedup_and_compare_addresses(self, ag_obj, req_addrs):
def _process_requested_addresses(self, ag_obj, req_addrs):
"""Process the requested addresses.
Deduplicate, normalize and compare the requested addresses
with existing addresses in the address group.
"""
ag_addrs = set(self._make_address_group_dict(
ag_obj, fields=['addresses'])['addresses'])
req_addrs = set(str(netaddr.IPNetwork(addr)) for addr in req_addrs)
normalized_addrs = set()
for addr in req_addrs:
addr = netaddr.IPNetwork(addr)
normalized_addr = "%s/%s" % (addr.network, addr.prefixlen)
normalized_addrs.add(normalized_addr)
addrs_in_ag = []
addrs_not_in_ag = []
for req_addr in req_addrs:
if req_addr in ag_addrs:
addrs_in_ag.append(req_addr)
for addr in normalized_addrs:
if addr in ag_addrs:
addrs_in_ag.append(addr)
else:
addrs_not_in_ag.append(req_addr)
addrs_not_in_ag.append(addr)
return addrs_in_ag, addrs_not_in_ag
def add_addresses(self, context, address_group_id, addresses):
ag = self._get_address_group(context, address_group_id)
addrs_in_ag, addrs_not_in_ag = self._dedup_and_compare_addresses(
addrs_in_ag, addrs_not_in_ag = self._process_requested_addresses(
ag, addresses['addresses'])
if addrs_in_ag:
raise ag_exc.AddressesAlreadyExist(
@ -72,7 +81,7 @@ class AddressGroupDbMixin(ag_ext.AddressGroupPluginBase):
def remove_addresses(self, context, address_group_id, addresses):
ag = self._get_address_group(context, address_group_id)
addrs_in_ag, addrs_not_in_ag = self._dedup_and_compare_addresses(
addrs_in_ag, addrs_not_in_ag = self._process_requested_addresses(
ag, addresses['addresses'])
if addrs_not_in_ag:
raise ag_exc.AddressesNotFound(
@ -86,7 +95,7 @@ class AddressGroupDbMixin(ag_ext.AddressGroupPluginBase):
def create_address_group(self, context, address_group):
"""Create an address group."""
fields = address_group['address_group']
args = {'project_id': fields['tenant_id'],
args = {'project_id': fields['project_id'],
'id': uuidutils.generate_uuid(),
'name': fields['name'],
'description': fields['description']}

29
neutron/tests/unit/extensions/test_address_group.py
View File

@ -122,9 +122,9 @@ class TestAddressGroup(AddressGroupTestCase):
expected_ag = {'name': 'foo',
'description': 'bar',
'tenant_id': self._tenant_id,
'addresses': ['10.0.1.255/28', '192.168.0.1/32']}
'addresses': ['10.0.1.0/24', '192.168.0.1/32']}
self._test_create_address_group(name='foo', description='bar',
addresses=['10.0.1.255/28',
addresses=['10.0.1.0/24',
'192.168.0.1/32'],
expected=expected_ag)
@ -164,11 +164,14 @@ class TestAddressGroup(AddressGroupTestCase):
self._show('address-groups', ag['address_group']['id'],
expected_code=webob.exc.HTTPNotFound.code)
def test_add_valid_addresses(self):
def test_normalize_and_deduplicate_in_add_addresses(self):
ag = self._test_create_address_group(name='foo')
data = {'addresses': ['10.0.0.1/32', '2001::/32']}
data = {'addresses': ['10.0.1.0/24', '10.0.1.2/24', '2001:db8::/16']}
self._test_address_group_actions(ag['address_group']['id'], data,
'add_addresses', expected=data)
'add_addresses', expected={
'addresses': ['10.0.1.0/24',
'2001::/16']
})
def test_add_invalid_addresses(self):
ag = self._test_create_address_group(name='foo')
@ -179,27 +182,27 @@ class TestAddressGroup(AddressGroupTestCase):
def test_add_duplicated_addresses(self):
ag = self._test_create_address_group(name='foo',
addresses=['10.0.0.1/32'])
data = {'addresses': ['10.0.0.1/32']}
addresses=['10.0.1.0/24'])
data = {'addresses': ['10.0.1.2/24']}
res = self._test_address_group_actions(ag['address_group']['id'],
data, 'add_addresses')
self.assertEqual(webob.exc.HTTPBadRequest.code, res.status_int)
def test_remove_valid_addresses(self):
def test_normalize_and_deduplicate_in_remove_addresses(self):
ag = self._test_create_address_group(name='foo',
addresses=['10.0.0.1/32',
'2001::/32'])
data = {'addresses': ['10.0.0.1/32']}
addresses=['10.0.1.0/24',
'2001::/16'])
data = {'addresses': ['10.0.1.0/24', '10.0.1.2/24', '2001:db8::/16']}
self._test_address_group_actions(ag['address_group']['id'],
data, 'remove_addresses',
expected={
'addresses': ['2001::/32']
'addresses': []
})
def test_remove_absent_addresses(self):
ag = self._test_create_address_group(name='foo',
addresses=['10.0.0.1/32'])
data = {'addresses': ['2001::/32']}
data = {'addresses': ['2001::/16']}
res = self._test_address_group_actions(ag['address_group']['id'],
data, 'remove_addresses')
self.assertEqual(webob.exc.HTTPNotFound.code, res.status_int)