Merge "Ensure tempest tests don't assume IP address allocation strategy"
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