Don't specify exactly IP in portforwarding functional test

In Port Forwarding functional tests, port is created in setUp()
method. This port had specified exactly IP address and subnet to use.
This was proposed in [1] to fix issue when new port created in same
subnet got sometimes IP address outside of subnet range.
Unfortunatelly that fix was not enough and caused other issues
as sometimes this "fixed" IP address used for port might be
used e.g. by router_interface port and that caused sometimes
"409 conflict" response from Neutron sometimes.

This patch adds additional method in
neutron.tests.unit.db.test_db_base_plugin_v2 module to find first
free IP address from given subnet.
This method is now used in port forwarding tests module to choose
IP address for second created port in the subnet so there should be
no conflicts with IP addresses anymore.

[1] https://review.openstack.org/#/c/631815

Change-Id: Iee2f72acf6814a0f8f76d240862429a8b63c3b09
Closes-bug: #1813540
This commit is contained in:
Slawek Kaplonski 2019-02-01 14:42:23 +01:00
parent f3810d0da3
commit 414bdd4c59
2 changed files with 17 additions and 13 deletions

View File

@ -13,7 +13,6 @@
import threading
import mock
import netaddr
from neutron_lib.api.definitions import floating_ip_port_forwarding as apidef
from neutron_lib.callbacks import exceptions as c_exc
from neutron_lib import exceptions as lib_exc
@ -99,15 +98,9 @@ class PortForwardingTestCase(PortForwardingTestCaseBase):
self._set_router_gw(self.router['id'], self.ext_net['id'])
self._add_router_interface(self.router['id'], self.subnet['id'])
self.fip = self._create_floatingip(self.ext_net['id'])
# We choose an IP address in the middle of the subnet so
# tests that update the IP in the port,
# like test_concurrent_create_port_forwarding_update_port(),
# don't accidentally choose an invalid IP address in
# the subnet, like the broadcast address.
self.port = self._create_port(
self.fmt, self.net['id'],
fixed_ips=[{'subnet_id': self.subnet['id'],
'ip_address': '10.0.0.100'}]).json['port']
fixed_ips=[{'subnet_id': self.subnet['id']}]).json['port']
self.port_forwarding = {
apidef.RESOURCE_NAME:
{apidef.EXTERNAL_PORT: 2225,
@ -449,11 +442,7 @@ class PortForwardingTestCase(PortForwardingTestCaseBase):
funcs, args_list)
def test_concurrent_create_port_forwarding_update_port(self):
# The initial IP address of the port is in the middle of the
# subnet range, so adding 2 to it should also produce a
# valid IP address.
new_ip = str(
netaddr.IPAddress(self.port['fixed_ips'][0]['ip_address']) + 2)
new_ip = self._find_ip_address(self.subnet)
funcs = [self.pf_plugin.create_floatingip_port_forwarding,
self._update_port]
args_list = [(self.context, self.fip['id'], self.port_forwarding),

View File

@ -289,6 +289,21 @@ class NeutronDbPluginV2TestCase(testlib_api.WebTestCase):
data = self._deserializers[ctype].deserialize(response.body)['body']
return data
def _find_ip_address(self, subnet):
network_ports = self._list_ports(
"json", 200, subnet['network_id']).json['ports']
used_ips = set()
for port in network_ports:
for ip in port['fixed_ips']:
if ip['subnet_id'] == subnet['id']:
used_ips.add(ip['ip_address'])
for pool in subnet['allocation_pools']:
ips_range = netaddr.IPRange(pool['start'], pool['end'])
for ip in ips_range:
if ip not in used_ips:
return str(ip)
def _create_bulk_from_list(self, fmt, resource, objects, **kwargs):
"""Creates a bulk request from a list of objects."""
collection = "%ss" % resource