Ensure tempest tests don't assume IP address allocation strategy
With the advent of pluggable IPAM in Neutron, it is not safe to
assume any particular algorithm is used to allocate IP addresses.
This change ensures the fixed IP sent to Neutron when we need a
specific fixed IP doesn't conflict as it is unsafe to simply take
the numerically highest IP address and add 1 to it. This change
introduces a helper method that finds an IP address not in use,
making it so we can safely ask for that specific IP address.
Change-Id: I84195b0eb63b7ca6a4e00becbe09e579ff8b718e
Partial-Bug: #1543094
diff --git a/tempest/api/compute/servers/test_attach_interfaces.py b/tempest/api/compute/servers/test_attach_interfaces.py
index 8201363..fdf55e5 100644
--- a/tempest/api/compute/servers/test_attach_interfaces.py
+++ b/tempest/api/compute/servers/test_attach_interfaces.py
@@ -13,10 +13,10 @@
# License for the specific language governing permissions and limitations
# under the License.
-import netaddr
import time
from tempest.api.compute import base
+from tempest.common.utils import net_utils
from tempest import config
from tempest import exceptions
from tempest.lib import exceptions as lib_exc
@@ -125,18 +125,21 @@
def _test_create_interface_by_fixed_ips(self, server, ifs):
network_id = ifs[0]['net_id']
- ip_list = [
- ifs[n]['fixed_ips'][0]['ip_address'] for n in range(0, len(ifs))]
- ip = str(netaddr.IPAddress(sorted(ip_list)[-1]) + 1)
+ subnet_id = ifs[0]['fixed_ips'][0]['subnet_id']
+ ip_list = net_utils.get_unused_ip_addresses(self.ports_client,
+ self.subnets_client,
+ network_id,
+ subnet_id,
+ 1)
- fixed_ips = [{'ip_address': ip}]
+ fixed_ips = [{'ip_address': ip_list[0]}]
iface = self.client.create_interface(
server['id'], net_id=network_id,
fixed_ips=fixed_ips)['interfaceAttachment']
self.addCleanup(self.ports_client.delete_port, iface['port_id'])
iface = self.wait_for_interface_status(
server['id'], iface['port_id'], 'ACTIVE')
- self._check_interface(iface, fixed_ip=ip)
+ self._check_interface(iface, fixed_ip=ip_list[0])
return iface
def _test_show_interface(self, server, ifs):
diff --git a/tempest/api/network/test_floating_ips.py b/tempest/api/network/test_floating_ips.py
index 2156e64..2abbf93 100644
--- a/tempest/api/network/test_floating_ips.py
+++ b/tempest/api/network/test_floating_ips.py
@@ -13,10 +13,9 @@
# License for the specific language governing permissions and limitations
# under the License.
-import netaddr
-
from tempest.api.network import base
from tempest.common.utils import data_utils
+from tempest.common.utils import net_utils
from tempest import config
from tempest import test
@@ -192,8 +191,12 @@
@test.idempotent_id('45c4c683-ea97-41ef-9c51-5e9802f2f3d7')
def test_create_update_floatingip_with_port_multiple_ip_address(self):
# Find out ips that can be used for tests
- ips = list(netaddr.IPNetwork(self.subnet['cidr']))
- list_ips = [str(ip) for ip in ips[-3:-1]]
+ list_ips = net_utils.get_unused_ip_addresses(
+ self.ports_client,
+ self.subnets_client,
+ self.subnet['network_id'],
+ self.subnet['id'],
+ 2)
fixed_ips = [{'ip_address': list_ips[0]}, {'ip_address': list_ips[1]}]
# Create port
body = self.ports_client.create_port(network_id=self.network['id'],
diff --git a/tempest/common/utils/net_utils.py b/tempest/common/utils/net_utils.py
new file mode 100644
index 0000000..d98fb32
--- /dev/null
+++ b/tempest/common/utils/net_utils.py
@@ -0,0 +1,48 @@
+# Copyright 2016 Hewlett Packard Enterprise Development Company
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+import itertools
+import netaddr
+
+from tempest.lib import exceptions as lib_exc
+
+
+def get_unused_ip_addresses(ports_client, subnets_client,
+ network_id, subnet_id, count):
+
+ """Return a list with the specified number of unused IP addresses
+
+ This method uses the given ports_client to find the specified number of
+ unused IP addresses on the given subnet using the supplied subnets_client
+ """
+
+ ports = ports_client.list_ports(network_id=network_id)['ports']
+ subnet = subnets_client.show_subnet(subnet_id)
+ ip_net = netaddr.IPNetwork(subnet['subnet']['cidr'])
+ subnet_set = netaddr.IPSet(ip_net.iter_hosts())
+ alloc_set = netaddr.IPSet()
+
+ # prune out any addresses already allocated to existing ports
+ for port in ports:
+ for fixed_ip in port.get('fixed_ips'):
+ alloc_set.add(fixed_ip['ip_address'])
+
+ av_set = subnet_set - alloc_set
+ ip_list = [str(ip) for ip in itertools.islice(av_set, count)]
+
+ if len(ip_list) != count:
+ msg = "Insufficient IP addresses available"
+ raise lib_exc.BadRequest(message=msg)
+
+ return ip_list