openstack/neutron
Code Issues Proposed changes

Get ports query: extract limit and use it only at the end.

Browse Source 
_get_ports_query add filters after a get_collection. If limits are
passed to applied to the query, then no additional filter is allowed.
This patch extracts an eventual limit argument, to apply it only after
the additional filters.

Change-Id: I83394394860d10e27379efe0356d0fa9c567140e
Closes-Bug: #1826186
This commit is contained in:
Gabriele Cerami 2019-04-25 11:28:17 +01:00
parent fb518b1cb9
commit cf8f3326be
2 changed files with 15 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
neutron/db/db_base_plugin_v2.py
View File

@ -1475,6 +1475,7 @@ class NeutronDbPluginV2(db_base_plugin_common.DbBasePluginCommon,
Port = models_v2.Port
IPAllocation = models_v2.IPAllocation
limit = kwargs.pop('limit', None)
filters = filters or {}
fixed_ips = filters.pop('fixed_ips', {})
vif_type = filters.pop(portbindings_def.VIF_TYPE, None)
@ -1491,6 +1492,8 @@ class NeutronDbPluginV2(db_base_plugin_common.DbBasePluginCommon,
if subnet_ids:
query = query.filter(
Port.fixed_ips.any(IPAllocation.subnet_id.in_(subnet_ids)))
if limit:
query = query.limit(limit)
return query
@db_api.retry_if_session_inactive()

13
neutron/tests/unit/db/test_db_base_plugin_v2.py
View File

@ -1203,7 +1203,8 @@ class TestPortsV2(NeutronDbPluginV2TestCase):
ports = (v1, v2, v3)
self._test_list_resources('port', ports)
def test_list_ports_filtered_by_fixed_ip(self):
def _test_list_ports_filtered_by_fixed_ip(self, **kwargs):
# for this test we need to enable overlapping ips
cfg.CONF.set_default('allow_overlapping_ips', True)
with self.port() as port1, self.port():
@ -1213,9 +1214,19 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s
""".strip() % (fixed_ips['ip_address'],
'192.168.126.5',
fixed_ips['subnet_id'])
extra_params = "&".join(["{}={}".format(k, v)
for k, v in kwargs.items()])
if extra_params:
query_params = "{}&{}".format(query_params, extra_params)
self._test_list_resources('port', [port1],
query_params=query_params)
def test_list_ports_filtered_by_fixed_ip(self):
self._test_list_ports_filtered_by_fixed_ip()
def test_list_ports_filtered_by_fixed_ip_with_limit(self):
self._test_list_ports_filtered_by_fixed_ip(limit=500)
def test_list_ports_public_network(self):
with self.network(shared=True) as network:
with self.subnet(network) as subnet: