From 414bdd4c5997886c97d0e3a0b19c684dafeabb7a Mon Sep 17 00:00:00 2001 From: Slawek Kaplonski Date: Fri, 1 Feb 2019 14:42:23 +0100 Subject: [PATCH] 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 --- .../portforwarding/test_port_forwarding.py | 15 ++------------- neutron/tests/unit/db/test_db_base_plugin_v2.py | 15 +++++++++++++++ 2 files changed, 17 insertions(+), 13 deletions(-) diff --git a/neutron/tests/functional/services/portforwarding/test_port_forwarding.py b/neutron/tests/functional/services/portforwarding/test_port_forwarding.py index ad3c0d9e645..e0fc7120065 100644 --- a/neutron/tests/functional/services/portforwarding/test_port_forwarding.py +++ b/neutron/tests/functional/services/portforwarding/test_port_forwarding.py @@ -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), diff --git a/neutron/tests/unit/db/test_db_base_plugin_v2.py b/neutron/tests/unit/db/test_db_base_plugin_v2.py index 7ef95fe75d2..58bdba48ccb 100644 --- a/neutron/tests/unit/db/test_db_base_plugin_v2.py +++ b/neutron/tests/unit/db/test_db_base_plugin_v2.py @@ -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