Merge "Introduce an API test for specified floating ip address"

neutron/tests/api/admin/test_floating_ips_admin_actions.py

@@ -12,6 +12,7 @@
 #    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 testtools
 
 from tempest_lib.common.utils import data_utils
 from tempest_lib import exceptions as lib_exc
@@ -132,3 +133,20 @@ class FloatingIPAdminTestJSON(base.BaseAdminNetworkTest):
         self.assertRaises(lib_exc.BadRequest,
                           self.admin_client.update_floatingip,
                           floating_ip['id'], port_id=port['port']['id'])
+
+    @testtools.skipUnless(
+        CONF.network_feature_enabled.specify_floating_ip_address_available,
+        "Feature for specifying floating IP address is disabled")
+    @test.attr(type='smoke')
+    @test.idempotent_id('332a8ae4-402e-4b98-bb6f-532e5a87b8e0')
+    def test_create_floatingip_with_specified_ip_address(self):
+        fip = self.get_unused_ip(self.ext_net_id)
+        body = self.admin_client.create_floatingip(
+            floating_network_id=self.ext_net_id,
+            floating_ip_address=fip)
+        created_floating_ip = body['floatingip']
+        self.addCleanup(self.admin_client.delete_floatingip,
+                        created_floating_ip['id'])
+        self.assertIsNotNone(created_floating_ip['id'])
+        self.assertIsNotNone(created_floating_ip['tenant_id'])
+        self.assertEqual(created_floating_ip['floating_ip_address'], fip)

neutron/tests/api/base.py

@@ -542,3 +542,41 @@ class BaseAdminNetworkTest(BaseNetworkTest):
         service_profile = body['service_profile']
         cls.service_profiles.append(service_profile)
         return service_profile
+
+    @classmethod
+    def get_unused_ip(cls, net_id):
+        """Get an unused ip address in a allocaion pool of net"""
+        body = cls.admin_client.list_ports(network_id=net_id)
+        ports = body['ports']
+        used_ips = []
+        for port in ports:
+            used_ips.extend(
+                [fixed_ip['ip_address'] for fixed_ip in port['fixed_ips']])
+        body = cls.admin_client.list_subnets(network_id=net_id)
+        subnets = body['subnets']
+
+        for subnet in subnets:
+            cidr = subnet['cidr']
+            allocation_pools = subnet['allocation_pools']
+            iterators = []
+            if allocation_pools:
+                for allocation_pool in allocation_pools:
+                    iterators.append(netaddr.iter_iprange(
+                        allocation_pool['start'], allocation_pool['end']))
+            else:
+                net = netaddr.IPNetwork(cidr)
+
+                def _iterip():
+                    for ip in net:
+                        if ip not in (net.network, net.broadcast):
+                            yield ip
+                iterators.append(_iterip)
+
+            for iterator in iterators:
+                for ip in iterator:
+                    if str(ip) not in used_ips:
+                        return str(ip)
+
+        message = (
+            "net(%s) has no usable IP address in allocation pools" % net_id)
+        raise exceptions.InvalidConfiguration(message)

neutron/tests/tempest/config.py

@@ -479,6 +479,10 @@ NetworkFeaturesGroup = [
                      "the extended IPv6 attributes ipv6_ra_mode "
                      "and ipv6_address_mode"
                 ),
+    cfg.BoolOpt('specify_floating_ip_address_available',
+                default=True,
+                help='Allow passing an IP Address of the floating ip when '
+                     'creating the floating ip'),
 ]
 
 messaging_group = cfg.OptGroup(name='messaging',