openstack/neutron
Code Issues Proposed changes

Merge "Add scope ID to the "GROUP BY" clause in `get_scoped_floating_ips`"

Browse Source
This commit is contained in:
Zuul 2023-05-16 16:44:28 +00:00 committed by Gerrit Code Review
commit 7b9e4a9659
3 changed files with 37 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
neutron/objects/router.py
View File

@ -420,8 +420,12 @@ class FloatingIP(base.NeutronDbObject):
# Filter out on router_ids
query = query.filter(l3.FloatingIP.router_id.in_(router_ids))
# Remove duplicate rows based on FIP IDs
query = query.group_by(l3.FloatingIP.id)
# Remove duplicate rows based on FIP IDs and the subnet pool address
# scope. Only one subnet pool (per IP version, 4 in this case) can
# be assigned to a subnet. The subnet pool address scope for a FIP is
# unique.
query = query.group_by(l3.FloatingIP.id,
models_v2.SubnetPool.address_scope_id)
for row in query:
yield (cls._load_object(context, row[0]), row[1])

20
neutron/tests/unit/objects/test_base.py
View File

@ -55,6 +55,7 @@ from neutron.objects import router
from neutron.objects import securitygroup
from neutron.objects import stdattrs
from neutron.objects import subnet
from neutron.objects import subnetpool
from neutron.tests import base as test_base
from neutron.tests import tools
from neutron.tests.unit.db import test_db_base_plugin_v2
@ -1583,7 +1584,7 @@ class BaseDbObjectTestCase(_BaseObjectTestCase,
fip_obj.create()
return fip_obj.id
def _create_test_subnet_id(self, network_id=None):
def _create_test_subnet_id(self, network_id=None, subnet_pool_id=None):
if not network_id:
network_id = self._create_test_network_id()
test_subnet = {
@ -1597,6 +1598,8 @@ class BaseDbObjectTestCase(_BaseObjectTestCase,
'ipv6_ra_mode': None,
'ipv6_address_mode': None
}
if subnet_pool_id:
test_subnet['subnetpool_id'] = subnet_pool_id
subnet_obj = subnet.Subnet(self.context, **test_subnet)
subnet_obj.create()
return subnet_obj.id
@ -1722,6 +1725,21 @@ class BaseDbObjectTestCase(_BaseObjectTestCase,
_qos_policy.create()
return _qos_policy
def _create_test_subnet_pool(
self, prefix, default_prefixlen, min_prefixlen, max_prefixlen,
ip_version):
subnet_pool = {
'prefixes': [prefix],
'default_prefixlen': default_prefixlen,
'min_prefixlen': min_prefixlen,
'max_prefixlen': max_prefixlen,
'ip_version': ip_version,
'address_scope_id': uuidutils.generate_uuid(),
}
subnet_pool_obj = subnetpool.SubnetPool(self.context, **subnet_pool)
subnet_pool_obj.create()
return subnet_pool_obj.id, subnet_pool_obj.address_scope_id
def test_get_standard_attr_id(self):
if not self._test_class.has_standard_attributes():

21
neutron/tests/unit/objects/test_router.py
View File

@ -329,17 +329,19 @@ class FloatingIPDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
def test_get_scoped_floating_ips(self):
def compare_results(router_ids, original_fips):
self.assertCountEqual(
original_fips,
[
fip[0].id
for fip in router.FloatingIP.get_scoped_floating_ips(
self.context, router_ids)
]
)
fips_scope = [fip for fip in
router.FloatingIP.get_scoped_floating_ips(
self.context, router_ids)]
fip_ids = [fip[0].id for fip in fips_scope]
as_ids = {fip[1] for fip in fips_scope}
self.assertCountEqual(original_fips, fip_ids)
self.assertEqual(1, len(as_ids))
self.assertEqual(address_scope_id, as_ids.pop())
# Setup three routers, networks and external networks
routers = {}
subnet_pool_id, address_scope_id = self._create_test_subnet_pool(
'10.0.0.0/16', 24, 16, 28, constants.IP_VERSION_4)
for i in range(3):
router_id = self._create_test_router_id(name=f'router-{i}')
routers[router_id] = []
@ -349,7 +351,8 @@ class FloatingIPDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
# Create three subnets and three FIPs using the
# aforementioned networks and routers
for j in range(3):
self._create_test_subnet_id(net_id)
self._create_test_subnet_id(network_id=net_id,
subnet_pool_id=subnet_pool_id)
fip = router.FloatingIP(
self.context,
floating_ip_address=netaddr.IPAddress(f'10.{i}.{j}.3'),