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

_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
(cherry picked from commit cf8f3326be)
changes/66/656066/2
Gabriele Cerami 4 years ago
parent fe35b91595
commit 2c6f733933
  1. 3
      neutron/db/db_base_plugin_v2.py
  2. 13
      neutron/tests/unit/db/test_db_base_plugin_v2.py

@ -1391,6 +1391,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', {})
query = model_query.get_collection_query(context, Port,
@ -1404,6 +1405,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()

@ -1166,7 +1166,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():
@ -1176,9 +1177,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:

Loading…
Cancel
Save