openstack
/
neutron
Code Issues Proposed changes

Merge "Get ports query: extract limit and use it only at the end." into stable/rocky

This commit is contained in:
Zuul 2019-05-04 20:08:17 +00:00 committed by Gerrit Code Review
parent a75b6e4696 d9ff4aee20
commit 39fbdfd9c3
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

@ -1414,6 +1414,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,
@ -1427,6 +1428,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

@ -1180,7 +1180,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():
@ -1190,9 +1191,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: