Merge "Replace old requires-python with python-requires"
diff --git a/.zuul.yaml b/.zuul.yaml
index 2e99198..3a6b925 100644
--- a/.zuul.yaml
+++ b/.zuul.yaml
@@ -63,8 +63,10 @@
- qos-fip
- quotas
- quota_details
+ - rbac-address-scope
- rbac-policies
- rbac-security-groups
+ - rbac-subnetpool
- router
- router-admin-state-down-before-update
- router_availability_zone
@@ -512,6 +514,10 @@
description: |
Perform setup common to all tempest scenario test jobs.
vars:
+ # NOTE(slaweq): in case of some tests, which requires advanced image,
+ # default test timeout set to 1200 seconds may be not enough if job is
+ # run on slow node
+ tempest_test_timeout: 2400
tempest_test_regex: ^neutron_tempest_plugin\.scenario
devstack_localrc:
PHYSICAL_NETWORK: default
@@ -869,6 +875,10 @@
tempest_concurrency: 4
tox_envlist: all
tempest_test_regex: ^neutron_tempest_plugin\.scenario
+ # NOTE(slaweq): in case of some tests, which requires advanced image,
+ # default test timeout set to 1200 seconds may be not enough if job is
+ # run on slow node
+ tempest_test_timeout: 2400
network_api_extensions_common: *api_extensions_master
network_api_extensions_dvr:
- dvr
diff --git a/neutron_tempest_plugin/api/base.py b/neutron_tempest_plugin/api/base.py
index f8cb339..1b02211 100644
--- a/neutron_tempest_plugin/api/base.py
+++ b/neutron_tempest_plugin/api/base.py
@@ -789,12 +789,15 @@
return body['address_scope']
@classmethod
- def create_subnetpool(cls, name, is_admin=False, **kwargs):
+ def create_subnetpool(cls, name, is_admin=False, client=None, **kwargs):
+ if client is None:
+ client = cls.admin_client if is_admin else cls.client
+
if is_admin:
- body = cls.admin_client.create_subnetpool(name, **kwargs)
+ body = client.create_subnetpool(name, **kwargs)
cls.admin_subnetpools.append(body['subnetpool'])
else:
- body = cls.client.create_subnetpool(name, **kwargs)
+ body = client.create_subnetpool(name, **kwargs)
cls.subnetpools.append(body['subnetpool'])
return body['subnetpool']
diff --git a/neutron_tempest_plugin/api/test_address_scopes.py b/neutron_tempest_plugin/api/test_address_scopes.py
index 6cf0885..b8c143a 100644
--- a/neutron_tempest_plugin/api/test_address_scopes.py
+++ b/neutron_tempest_plugin/api/test_address_scopes.py
@@ -11,6 +11,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.common import utils
from tempest.lib.common.utils import data_utils
@@ -115,3 +116,141 @@
address_scope = self._test_update_address_scope_helper(is_admin=True,
shared=True)
self.assertTrue(address_scope['shared'])
+
+
+class RbacAddressScopeTest(AddressScopeTestBase):
+
+ force_tenant_isolation = True
+ credentials = ['primary', 'alt', 'admin']
+ required_extensions = ['address-scope', 'rbac-address-scope']
+
+ @classmethod
+ def resource_setup(cls):
+ super(RbacAddressScopeTest, cls).resource_setup()
+ cls.client2 = cls.os_alt.network_client
+
+ def _make_admin_as_shared_to_project_id(self, project_id):
+ a_s = self._create_address_scope(ip_version=4, is_admin=True)
+ rbac_policy = self.admin_client.create_rbac_policy(
+ object_type='address_scope',
+ object_id=a_s['id'],
+ action='access_as_shared',
+ target_tenant=project_id,
+ )['rbac_policy']
+ return {'address_scope': a_s, 'rbac_policy': rbac_policy}
+
+ @decorators.idempotent_id('038e999b-cd4b-4021-a9ff-ebb734f6e056')
+ def test_policy_target_update(self):
+ res = self._make_admin_as_shared_to_project_id(
+ self.client.tenant_id)
+ # change to client2
+ update_res = self.admin_client.update_rbac_policy(
+ res['rbac_policy']['id'], target_tenant=self.client2.tenant_id)
+ self.assertEqual(self.client2.tenant_id,
+ update_res['rbac_policy']['target_tenant'])
+ # make sure everything else stayed the same
+ res['rbac_policy'].pop('target_tenant')
+ update_res['rbac_policy'].pop('target_tenant')
+ self.assertEqual(res['rbac_policy'], update_res['rbac_policy'])
+
+ @decorators.idempotent_id('798ac6c6-96cc-49ce-ba5c-c6eced7a09d3')
+ def test_subnet_pool_presence_prevents_rbac_policy_deletion(self):
+ res = self._make_admin_as_shared_to_project_id(
+ self.client2.tenant_id)
+ snp = self.create_subnetpool(
+ data_utils.rand_name("rbac-address-scope"),
+ default_prefixlen=24, prefixes=['10.0.0.0/8'],
+ address_scope_id=res['address_scope']['id'],
+ client=self.client2
+ )
+ self.addCleanup(
+ self.admin_client.delete_rbac_policy,
+ res['rbac_policy']['id']
+ )
+ self.addCleanup(self.client2.delete_subnetpool, snp['id'])
+
+ # a port with shared sg should prevent the deletion of an
+ # rbac-policy required for it to be shared
+ with testtools.ExpectedException(lib_exc.Conflict):
+ self.admin_client.delete_rbac_policy(res['rbac_policy']['id'])
+
+ @decorators.idempotent_id('57da6ba2-6329-49c8-974c-9858fe187136')
+ def test_regular_client_shares_to_another_regular_client(self):
+ # owned by self.admin_client
+ a_s = self._create_address_scope(ip_version=4, is_admin=True)
+ with testtools.ExpectedException(lib_exc.NotFound):
+ self.client.show_address_scope(a_s['id'])
+ rbac_policy = self.admin_client.create_rbac_policy(
+ object_type='address_scope', object_id=a_s['id'],
+ action='access_as_shared',
+ target_tenant=self.client.tenant_id)['rbac_policy']
+ self.client.show_address_scope(a_s['id'])
+
+ self.assertIn(rbac_policy,
+ self.admin_client.list_rbac_policies()['rbac_policies'])
+ # ensure that 'client2' can't see the rbac-policy sharing the
+ # as to it because the rbac-policy belongs to 'client'
+ self.assertNotIn(rbac_policy['id'], [p['id'] for p in
+ self.client2.list_rbac_policies()['rbac_policies']])
+
+ @decorators.idempotent_id('051248e7-d66f-4c69-9022-2b73ee5b9e73')
+ def test_filter_fields(self):
+ a_s = self._create_address_scope(ip_version=4)
+ self.admin_client.create_rbac_policy(
+ object_type='address_scope', object_id=a_s['id'],
+ action='access_as_shared', target_tenant=self.client2.tenant_id)
+ field_args = (('id',), ('id', 'action'), ('object_type', 'object_id'),
+ ('project_id', 'target_tenant'))
+ for fields in field_args:
+ res = self.admin_client.list_rbac_policies(fields=fields)
+ self.assertEqual(set(fields), set(res['rbac_policies'][0].keys()))
+
+ @decorators.idempotent_id('19cbd62e-c6c3-4495-98b9-b9c6c6c9c127')
+ def test_rbac_policy_show(self):
+ res = self._make_admin_as_shared_to_project_id(
+ self.client.tenant_id)
+ p1 = res['rbac_policy']
+ p2 = self.admin_client.create_rbac_policy(
+ object_type='address_scope',
+ object_id=res['address_scope']['id'],
+ action='access_as_shared',
+ target_tenant='*')['rbac_policy']
+
+ self.assertEqual(
+ p1, self.admin_client.show_rbac_policy(p1['id'])['rbac_policy'])
+ self.assertEqual(
+ p2, self.admin_client.show_rbac_policy(p2['id'])['rbac_policy'])
+
+ @decorators.idempotent_id('88852ba0-8546-4ce7-8f79-4a9c840c881d')
+ def test_filter_rbac_policies(self):
+ a_s = self._create_address_scope(ip_version=4)
+ rbac_pol1 = self.admin_client.create_rbac_policy(
+ object_type='address_scope', object_id=a_s['id'],
+ action='access_as_shared',
+ target_tenant=self.client2.tenant_id)['rbac_policy']
+ rbac_pol2 = self.admin_client.create_rbac_policy(
+ object_type='address_scope', object_id=a_s['id'],
+ action='access_as_shared',
+ target_tenant=self.admin_client.tenant_id)['rbac_policy']
+ res1 = self.admin_client.list_rbac_policies(id=rbac_pol1['id'])[
+ 'rbac_policies']
+ res2 = self.admin_client.list_rbac_policies(id=rbac_pol2['id'])[
+ 'rbac_policies']
+ self.assertEqual(1, len(res1))
+ self.assertEqual(1, len(res2))
+ self.assertEqual(rbac_pol1['id'], res1[0]['id'])
+ self.assertEqual(rbac_pol2['id'], res2[0]['id'])
+
+ @decorators.idempotent_id('222a638d-819e-41a7-a3fe-550265c06e79')
+ def test_regular_client_blocked_from_sharing_anothers_policy(self):
+ a_s = self._make_admin_as_shared_to_project_id(
+ self.client.tenant_id)['address_scope']
+ with testtools.ExpectedException(lib_exc.BadRequest):
+ self.client.create_rbac_policy(
+ object_type='address_scope', object_id=a_s['id'],
+ action='access_as_shared',
+ target_tenant=self.client2.tenant_id)
+
+ # make sure the rbac-policy is invisible to the tenant for which it's
+ # being shared
+ self.assertFalse(self.client.list_rbac_policies()['rbac_policies'])
diff --git a/neutron_tempest_plugin/api/test_security_groups.py b/neutron_tempest_plugin/api/test_security_groups.py
index dab183f..3c611eb 100644
--- a/neutron_tempest_plugin/api/test_security_groups.py
+++ b/neutron_tempest_plugin/api/test_security_groups.py
@@ -291,6 +291,22 @@
self.assertEqual(quota_set, self._get_sg_rules_amount(),
"Amount of security groups rules doesn't match quota")
+ @decorators.idempotent_id('37508c8d-270b-4b93-8007-72876a1fec38')
+ def test_sg_rules_quota_values(self):
+ """Test security group rules quota values.
+
+ This test checks if it is possible to change the SG rules Quota
+ values, different values.
+ """
+ sg_rules_quota = self._get_sg_rules_quota()
+ project_id = self.client.tenant_id
+ self.addCleanup(self.admin_client.update_quotas,
+ project_id, **{'security_group_rule': sg_rules_quota})
+ values = [-1, 0, 10, 2147483647]
+ for value in values:
+ self._set_sg_rules_quota(value)
+ self.assertEqual(value, self._get_sg_rules_quota())
+
class SecGroupProtocolTest(base.BaseNetworkTest):
diff --git a/neutron_tempest_plugin/api/test_subnetpools.py b/neutron_tempest_plugin/api/test_subnetpools.py
index 9d927cf..38c721f 100644
--- a/neutron_tempest_plugin/api/test_subnetpools.py
+++ b/neutron_tempest_plugin/api/test_subnetpools.py
@@ -12,10 +12,12 @@
# 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.common import utils
from tempest.lib.common.utils import data_utils
from tempest.lib import decorators
+from tempest.lib import exceptions as lib_exc
from neutron_tempest_plugin.api import base
@@ -418,3 +420,171 @@
self._test_list_validation_filters(self.list_kwargs)
self._test_list_validation_filters({
'unknown_filter': 'value'}, filter_is_valid=False)
+
+
+class RbacSubnetPoolTest(SubnetPoolsTestBase):
+
+ force_tenant_isolation = True
+ credentials = ['primary', 'alt', 'admin']
+ required_extensions = ['rbac-subnetpool']
+
+ @classmethod
+ def resource_setup(cls):
+ super(RbacSubnetPoolTest, cls).resource_setup()
+ cls.client2 = cls.os_alt.network_client
+
+ def _make_admin_snp_shared_to_project_id(self, project_id):
+ snp = self._create_subnetpool(is_admin=True)
+ rbac_policy = self.admin_client.create_rbac_policy(
+ object_type='subnetpool',
+ object_id=snp['id'],
+ action='access_as_shared',
+ target_tenant=project_id,
+ )['rbac_policy']
+ return {'subnetpool': snp, 'rbac_policy': rbac_policy}
+
+ @decorators.idempotent_id('71b35ad0-51cd-40da-985d-89a51c95ec6a')
+ def test_policy_target_update(self):
+ res = self._make_admin_snp_shared_to_project_id(
+ self.client.tenant_id)
+ # change to client2
+ update_res = self.admin_client.update_rbac_policy(
+ res['rbac_policy']['id'], target_tenant=self.client2.tenant_id)
+ self.assertEqual(self.client2.tenant_id,
+ update_res['rbac_policy']['target_tenant'])
+ # make sure everything else stayed the same
+ res['rbac_policy'].pop('target_tenant')
+ update_res['rbac_policy'].pop('target_tenant')
+ self.assertEqual(res['rbac_policy'], update_res['rbac_policy'])
+
+ @decorators.idempotent_id('451d9d38-65a0-4916-a805-1460d6a938d1')
+ def test_subnet_presence_prevents_rbac_policy_deletion(self):
+ res = self._make_admin_snp_shared_to_project_id(
+ self.client2.tenant_id)
+ network = self.create_network(client=self.client2)
+ subnet = self.client2.create_subnet(
+ network_id=network['id'],
+ ip_version=4,
+ subnetpool_id=res['subnetpool']['id'],
+ name=data_utils.rand_name("rbac-subnetpool"),
+ )["subnet"]
+ self.addCleanup(self.client2.delete_network, network['id'])
+ self.addCleanup(
+ self.admin_client.delete_subnetpool,
+ res['subnetpool']['id']
+ )
+ self.addCleanup(self.client2.delete_subnet, subnet['id'])
+
+ # a port with shared sg should prevent the deletion of an
+ # rbac-policy required for it to be shared
+ with testtools.ExpectedException(lib_exc.Conflict):
+ self.admin_client.delete_rbac_policy(res['rbac_policy']['id'])
+
+ @decorators.idempotent_id('f74a71de-9abf-49c6-8199-4ac7f53e383b')
+ @utils.requires_ext(extension='rbac-address-scope', service='network')
+ def test_cannot_share_if_no_access_to_address_scope(self):
+ # Create Address Scope shared only to client but not to client2
+ a_s = self.admin_client.create_address_scope(
+ name=data_utils.rand_name("rbac-subnetpool"),
+ ip_version=4
+ )["address_scope"]
+ rbac_policy = self.admin_client.create_rbac_policy(
+ object_type='address_scope', object_id=a_s['id'],
+ action='access_as_shared',
+ target_tenant=self.client.tenant_id)['rbac_policy']
+
+ # Create subnet pool owned by client with shared AS
+ snp = self._create_subnetpool(address_scope_id=a_s["id"])
+
+ with testtools.ExpectedException(lib_exc.BadRequest):
+ self.client.create_rbac_policy(
+ object_type='subnetpool', object_id=snp['id'],
+ action='access_as_shared',
+ target_tenant=self.client2.tenant_id
+ )
+
+ # cleanup
+ self.client.delete_subnetpool(snp["id"])
+ self.admin_client.delete_rbac_policy(rbac_policy['id'])
+ self.admin_client.delete_address_scope(a_s['id'])
+
+ @decorators.idempotent_id('9cf8bba5-0163-4083-9397-678bb9b5f5a2')
+ def test_regular_client_shares_to_another_regular_client(self):
+ # owned by self.admin_client
+ snp = self._create_subnetpool(is_admin=True)
+ with testtools.ExpectedException(lib_exc.NotFound):
+ self.client.show_subnetpool(snp['id'])
+ rbac_policy = self.admin_client.create_rbac_policy(
+ object_type='subnetpool', object_id=snp['id'],
+ action='access_as_shared',
+ target_tenant=self.client.tenant_id)['rbac_policy']
+ self.client.show_subnetpool(snp['id'])
+
+ self.assertIn(rbac_policy,
+ self.admin_client.list_rbac_policies()['rbac_policies'])
+ # ensure that 'client2' can't see the rbac-policy sharing the
+ # as to it because the rbac-policy belongs to 'client'
+ self.assertNotIn(rbac_policy['id'], [p['id'] for p in
+ self.client2.list_rbac_policies()['rbac_policies']])
+
+ @decorators.idempotent_id('17b2b437-a5fa-4340-ad98-912a986d0d7c')
+ def test_filter_fields(self):
+ snp = self._create_subnetpool()
+ self.admin_client.create_rbac_policy(
+ object_type='subnetpool', object_id=snp['id'],
+ action='access_as_shared', target_tenant=self.client2.tenant_id)
+ field_args = (('id',), ('id', 'action'), ('object_type', 'object_id'),
+ ('project_id', 'target_tenant'))
+ for fields in field_args:
+ res = self.admin_client.list_rbac_policies(fields=fields)
+ self.assertEqual(set(fields), set(res['rbac_policies'][0].keys()))
+
+ @decorators.idempotent_id('e59e4502-4e6a-4e49-b446-a5d5642bbd69')
+ def test_rbac_policy_show(self):
+ res = self._make_admin_snp_shared_to_project_id(
+ self.client.tenant_id)
+ p1 = res['rbac_policy']
+ p2 = self.admin_client.create_rbac_policy(
+ object_type='subnetpool',
+ object_id=res['subnetpool']['id'],
+ action='access_as_shared',
+ target_tenant='*')['rbac_policy']
+
+ self.assertEqual(
+ p1, self.admin_client.show_rbac_policy(p1['id'])['rbac_policy'])
+ self.assertEqual(
+ p2, self.admin_client.show_rbac_policy(p2['id'])['rbac_policy'])
+
+ @decorators.idempotent_id('1c24c28c-eb1e-466e-af29-255cf127653a')
+ def test_filter_rbac_policies(self):
+ snp = self._create_subnetpool()
+ rbac_pol1 = self.admin_client.create_rbac_policy(
+ object_type='subnetpool', object_id=snp['id'],
+ action='access_as_shared',
+ target_tenant=self.client2.tenant_id)['rbac_policy']
+ rbac_pol2 = self.admin_client.create_rbac_policy(
+ object_type='subnetpool', object_id=snp['id'],
+ action='access_as_shared',
+ target_tenant=self.admin_client.tenant_id)['rbac_policy']
+ res1 = self.admin_client.list_rbac_policies(id=rbac_pol1['id'])[
+ 'rbac_policies']
+ res2 = self.admin_client.list_rbac_policies(id=rbac_pol2['id'])[
+ 'rbac_policies']
+ self.assertEqual(1, len(res1))
+ self.assertEqual(1, len(res2))
+ self.assertEqual(rbac_pol1['id'], res1[0]['id'])
+ self.assertEqual(rbac_pol2['id'], res2[0]['id'])
+
+ @decorators.idempotent_id('63d9acbe-403c-4e77-9ffd-80e636a4621e')
+ def test_regular_client_blocked_from_sharing_anothers_policy(self):
+ snp = self._make_admin_snp_shared_to_project_id(
+ self.client.tenant_id)['subnetpool']
+ with testtools.ExpectedException(lib_exc.BadRequest):
+ self.client.create_rbac_policy(
+ object_type='subnetpool', object_id=snp['id'],
+ action='access_as_shared',
+ target_tenant=self.client2.tenant_id)
+
+ # make sure the rbac-policy is invisible to the tenant for which it's
+ # being shared
+ self.assertFalse(self.client.list_rbac_policies()['rbac_policies'])
diff --git a/neutron_tempest_plugin/api/test_subnetpools_negative.py b/neutron_tempest_plugin/api/test_subnetpools_negative.py
index 214a012..1e222df 100644
--- a/neutron_tempest_plugin/api/test_subnetpools_negative.py
+++ b/neutron_tempest_plugin/api/test_subnetpools_negative.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
import netaddr
from oslo_utils import uuidutils
@@ -171,6 +172,10 @@
@decorators.attr(type='negative')
@decorators.idempotent_id('3396ec6c-cb80-4ebe-b897-84e904580bdf')
+ @testtools.skipIf(
+ utils.is_extension_enabled('rbac-address-scope', 'network'),
+ reason="Test is outdated starting from Ussuri release."
+ )
@utils.requires_ext(extension='address-scope', service='network')
def test_tenant_create_subnetpool_associate_shared_address_scope(self):
address_scope = self.create_address_scope(
diff --git a/neutron_tempest_plugin/api/test_trunk.py b/neutron_tempest_plugin/api/test_trunk.py
index 823a95d..1f83bd8 100644
--- a/neutron_tempest_plugin/api/test_trunk.py
+++ b/neutron_tempest_plugin/api/test_trunk.py
@@ -235,7 +235,9 @@
'segmentation_id': segmentation_id2}]
# Validate that subport got segmentation details from the network
- self.assertEqual(expected_subports, trunk['sub_ports'])
+ self.assertEqual(
+ sorted(expected_subports, key=lambda subport: subport['port_id']),
+ sorted(trunk['sub_ports'], key=lambda subport: subport['port_id']))
class TrunkTestMtusJSONBase(TrunkTestJSONBase):
diff --git a/neutron_tempest_plugin/scenario/test_floatingip.py b/neutron_tempest_plugin/scenario/test_floatingip.py
index ed5d239..2a137b5 100644
--- a/neutron_tempest_plugin/scenario/test_floatingip.py
+++ b/neutron_tempest_plugin/scenario/test_floatingip.py
@@ -25,6 +25,7 @@
from tempest.lib import exceptions
import testscenarios
from testscenarios.scenarios import multiply_scenarios
+import testtools
from neutron_tempest_plugin.api import base as base_api
from neutron_tempest_plugin.common import ssh
@@ -440,3 +441,124 @@
self.fail(
"Server %s is not accessible via its floating ip %s" % (
servers[-1]['id'], self.fip['id']))
+
+
+class FloatingIpMultipleRoutersTest(base.BaseTempestTestCase):
+ credentials = ['primary', 'admin']
+
+ @classmethod
+ @utils.requires_ext(extension="router", service="network")
+ def skip_checks(cls):
+ super(FloatingIpMultipleRoutersTest, cls).skip_checks()
+
+ def _create_keypair_and_secgroup(self):
+ self.keypair = self.create_keypair()
+ self.secgroup = self.create_security_group()
+ self.create_loginable_secgroup_rule(
+ secgroup_id=self.secgroup['id'])
+ self.create_pingable_secgroup_rule(
+ secgroup_id=self.secgroup['id'])
+
+ def _create_network_and_servers(self, servers_num=1, fip_addresses=None):
+ if fip_addresses:
+ self.assertEqual(servers_num, len(fip_addresses),
+ ('Number of specified fip addresses '
+ 'does not match the number of servers'))
+ network = self.create_network()
+ subnet = self.create_subnet(network)
+ router = self.create_router_by_client()
+ self.create_router_interface(router['id'], subnet['id'])
+
+ fips = []
+ for server in range(servers_num):
+ fip = fip_addresses[server] if fip_addresses else None
+ fips.append(
+ self._create_server_and_fip(
+ network=network, fip_address=fip))
+ return fips
+
+ def _create_server_and_fip(self, network, fip_address=None):
+ server = self.create_server(
+ flavor_ref=CONF.compute.flavor_ref,
+ image_ref=CONF.compute.image_ref,
+ key_name=self.keypair['name'],
+ networks=[{'uuid': network['id']}],
+ security_groups=[{'name': self.secgroup['name']}])
+ waiters.wait_for_server_status(self.os_primary.servers_client,
+ server['server']['id'],
+ constants.SERVER_STATUS_ACTIVE)
+ port = self.client.list_ports(
+ network_id=network['id'],
+ device_id=server['server']['id'])['ports'][0]
+
+ if fip_address:
+ fip = self.create_floatingip(
+ floating_ip_address=fip_address,
+ client=self.os_admin.network_client,
+ port=port)
+ self.addCleanup(
+ self.delete_floatingip, fip, self.os_admin.network_client)
+ else:
+ fip = self.create_floatingip(port=port)
+ return fip
+
+ def _check_fips_connectivity(self, mutable_fip, permanent_fip):
+ for fip in [mutable_fip, permanent_fip]:
+ fip['ssh_client'] = ssh.Client(fip['floating_ip_address'],
+ CONF.validation.image_ssh_user,
+ pkey=self.keypair['private_key'])
+ self.check_remote_connectivity(
+ permanent_fip['ssh_client'], mutable_fip['floating_ip_address'])
+ self.check_remote_connectivity(
+ mutable_fip['ssh_client'], permanent_fip['floating_ip_address'])
+
+ @testtools.skipUnless(CONF.network.public_network_id,
+ 'The public_network_id option must be specified.')
+ @decorators.idempotent_id('b0382ab3-3c86-4415-84e3-649a8b040dab')
+ def test_reuse_ip_address_with_other_fip_on_other_router(self):
+ """Reuse IP address by another floating IP on another router
+
+ Scenario:
+ 1. Create and connect a router to the external network.
+ 2. Create and connect an internal network to the router.
+ 3. Create and connect 2 VMs to the internal network.
+ 4. Create FIPs in the external network for the VMs.
+ 5. Make sure that VM1 can ping VM2 FIP address.
+ 6. Delete VM2 FIP but save IP address that it used.
+ 7. Create and connect one more router to the external network.
+ 8. Create and connect an internal network to the second router.
+ 9. Create and connect a VM (VM3) to the internal network of
+ the second router.
+ 10. Create a FIP for the VM3 in the external network with
+ the same IP address that was used for VM2.
+ 11. Make sure that now VM1 is able to reach VM3 using the FIP.
+
+ Note, the scenario passes only in case corresponding
+ ARP update was sent to the external network when reusing same IP
+ address for another FIP.
+ """
+
+ self._create_keypair_and_secgroup()
+ [mutable_fip, permanent_fip] = (
+ self._create_network_and_servers(servers_num=2))
+ self._check_fips_connectivity(mutable_fip, permanent_fip)
+ ip_address = mutable_fip['floating_ip_address']
+ self.delete_floatingip(mutable_fip)
+
+ def _fip_is_free():
+ fips = self.os_admin.network_client.list_floatingips(
+ )['floatingips']
+ for fip in fips:
+ if ip_address == fip['floating_ip_address']:
+ return False
+ return True
+
+ try:
+ common_utils.wait_until_true(lambda: _fip_is_free(),
+ timeout=30, sleep=5)
+ except common_utils.WaitTimeout:
+ self.fail("Can't reuse IP address because it is not free")
+
+ [mutable_fip] = self._create_network_and_servers(
+ servers_num=1, fip_addresses=[ip_address])
+ self._check_fips_connectivity(mutable_fip, permanent_fip)
diff --git a/neutron_tempest_plugin/services/network/json/network_client.py b/neutron_tempest_plugin/services/network/json/network_client.py
index ddb6f95..f056c7f 100644
--- a/neutron_tempest_plugin/services/network/json/network_client.py
+++ b/neutron_tempest_plugin/services/network/json/network_client.py
@@ -951,6 +951,17 @@
body = jsonutils.loads(body)
return service_client.ResponseBody(resp, body)
+ def list_floatingips(self, **kwargs):
+ post_body = {'floatingips': kwargs}
+ body = jsonutils.dumps(post_body)
+ uri = '%s/floatingips' % self.uri_prefix
+ if kwargs:
+ uri += '?' + urlparse.urlencode(kwargs, doseq=1)
+ resp, body = self.get(uri)
+ self.expected_success(200, resp.status)
+ body = jsonutils.loads(body)
+ return service_client.ResponseBody(resp, body)
+
def create_floatingip(self, floating_network_id, **kwargs):
post_body = {'floatingip': {
'floating_network_id': floating_network_id}}