bdacaa20ef
Neutron has added a new extension `floating-ip-port-forwarding-detail`, which defines that when listing floating ip, it returns the port forwardng's id and 'internal_port_id'. This patch updates the list floating ip verification content. [1] https://review.opendev.org/c/openstack/neutron/+/840565 Change-Id: I5a82d745a9bd9d098ddf8902cdd58d1e81bd80ea
241 lines
11 KiB
Python
241 lines
11 KiB
Python
# Copyright 2019 Red Hat, Inc.
|
|
# All Rights Reserved.
|
|
#
|
|
# 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.
|
|
|
|
from tempest.common import utils
|
|
from tempest.lib.common.utils import data_utils
|
|
from tempest.lib import decorators
|
|
from tempest.lib import exceptions
|
|
|
|
from neutron_tempest_plugin.api import base
|
|
from neutron_tempest_plugin import config
|
|
|
|
CONF = config.CONF
|
|
|
|
|
|
class PortForwardingTestJSON(base.BaseNetworkTest):
|
|
|
|
required_extensions = ['router', 'floating-ip-port-forwarding']
|
|
|
|
@classmethod
|
|
def resource_setup(cls):
|
|
super(PortForwardingTestJSON, cls).resource_setup()
|
|
cls.ext_net_id = CONF.network.public_network_id
|
|
|
|
# Create network, subnet, router and add interface
|
|
cls.network = cls.create_network()
|
|
cls.subnet = cls.create_subnet(cls.network)
|
|
cls.router = cls.create_router(data_utils.rand_name('router'),
|
|
external_network_id=cls.ext_net_id)
|
|
cls.create_router_interface(cls.router['id'], cls.subnet['id'])
|
|
|
|
@decorators.idempotent_id('829a446e-46bc-41ce-b442-6e428aeb3c19')
|
|
def test_port_forwarding_life_cycle(self):
|
|
fip = self.create_floatingip()
|
|
port = self.create_port(self.network)
|
|
# Create port forwarding for one TCP port
|
|
created_pf = self.create_port_forwarding(
|
|
fip['id'],
|
|
internal_port_id=port['id'],
|
|
internal_ip_address=port['fixed_ips'][0]['ip_address'],
|
|
internal_port=1111, external_port=2222, protocol="tcp")
|
|
self.assertEqual(1111, created_pf['internal_port'])
|
|
self.assertEqual(2222, created_pf['external_port'])
|
|
self.assertEqual('tcp', created_pf['protocol'])
|
|
self.assertEqual(port['fixed_ips'][0]['ip_address'],
|
|
created_pf['internal_ip_address'])
|
|
|
|
# Show created port forwarding
|
|
body = self.client.get_port_forwarding(
|
|
fip['id'], created_pf['id'])
|
|
pf = body['port_forwarding']
|
|
self.assertEqual(1111, pf['internal_port'])
|
|
self.assertEqual(2222, pf['external_port'])
|
|
self.assertEqual('tcp', pf['protocol'])
|
|
self.assertEqual(port['fixed_ips'][0]['ip_address'],
|
|
pf['internal_ip_address'])
|
|
|
|
# Update port forwarding
|
|
body = self.client.update_port_forwarding(
|
|
fip['id'], pf['id'], internal_port=3333)
|
|
pf = body['port_forwarding']
|
|
self.assertEqual(3333, pf['internal_port'])
|
|
self.assertEqual(2222, pf['external_port'])
|
|
self.assertEqual('tcp', pf['protocol'])
|
|
self.assertEqual(port['fixed_ips'][0]['ip_address'],
|
|
pf['internal_ip_address'])
|
|
|
|
# Now lets try to remove Floating IP with existing port forwarding,
|
|
# this should fails
|
|
self.assertRaises(exceptions.Conflict,
|
|
self.delete_floatingip, fip)
|
|
|
|
# Delete port forwarding
|
|
self.client.delete_port_forwarding(fip['id'], pf['id'])
|
|
self.assertRaises(exceptions.NotFound,
|
|
self.client.get_port_forwarding,
|
|
fip['id'], pf['id'])
|
|
|
|
# Now Floating IP should be deleted properly
|
|
self.delete_floatingip(fip)
|
|
self.assertRaises(exceptions.NotFound,
|
|
self.client.get_floatingip, fip['id'])
|
|
|
|
@decorators.idempotent_id('aa842070-39ef-4b09-9df9-e723934f96f8')
|
|
@utils.requires_ext(extension="expose-port-forwarding-in-fip",
|
|
service="network")
|
|
def test_port_forwarding_info_in_fip_details(self):
|
|
fip = self.create_floatingip()
|
|
port = self.create_port(self.network)
|
|
|
|
# Ensure that FIP don't have information about any port forwarding yet
|
|
fip = self.client.show_floatingip(fip['id'])['floatingip']
|
|
self.assertEqual(0, len(fip['port_forwardings']))
|
|
|
|
# Now create port forwarding and ensure that it is visible in FIP's
|
|
# details
|
|
pf = self.create_port_forwarding(
|
|
fip['id'],
|
|
internal_port_id=port['id'],
|
|
internal_ip_address=port['fixed_ips'][0]['ip_address'],
|
|
internal_port=1111, external_port=2222, protocol="tcp")
|
|
fip = self.client.show_floatingip(fip['id'])['floatingip']
|
|
self.assertEqual(1, len(fip['port_forwardings']))
|
|
self.assertEqual(1111, fip['port_forwardings'][0]['internal_port'])
|
|
self.assertEqual(2222, fip['port_forwardings'][0]['external_port'])
|
|
self.assertEqual('tcp', fip['port_forwardings'][0]['protocol'])
|
|
self.assertEqual(port['fixed_ips'][0]['ip_address'],
|
|
fip['port_forwardings'][0]['internal_ip_address'])
|
|
|
|
# Delete port forwarding and ensure that it's not in FIP's details
|
|
# anymore
|
|
self.client.delete_port_forwarding(fip['id'], pf['id'])
|
|
fip = self.client.show_floatingip(fip['id'])['floatingip']
|
|
self.assertEqual(0, len(fip['port_forwardings']))
|
|
|
|
def _verify_created_pfs(self, fip_id, created_pfs):
|
|
# Check that all PFs are visible in Floating IP details
|
|
fip = self.client.show_floatingip(fip_id)['floatingip']
|
|
self.assertEqual(len(created_pfs), len(fip['port_forwardings']))
|
|
detail_enabled = utils.is_extension_enabled(
|
|
'floating-ip-port-forwarding-detail', service='network')
|
|
for pf in created_pfs:
|
|
expected_pf = {
|
|
'external_port': pf['external_port'],
|
|
'internal_port': pf['internal_port'],
|
|
'protocol': pf['protocol'],
|
|
'internal_ip_address': pf['internal_ip_address']}
|
|
if detail_enabled:
|
|
expected_pf['id'] = pf['id']
|
|
expected_pf['internal_port_id'] = pf['internal_port_id']
|
|
self.assertIn(expected_pf, fip['port_forwardings'])
|
|
|
|
# Test list of port forwardings
|
|
port_forwardings = self.client.list_port_forwardings(
|
|
fip['id'])['port_forwardings']
|
|
self.assertEqual(len(created_pfs), len(port_forwardings))
|
|
for pf in created_pfs:
|
|
expected_pf = pf.copy()
|
|
expected_pf.pop('client')
|
|
expected_pf.pop('floatingip_id')
|
|
self.assertIn(expected_pf, port_forwardings)
|
|
|
|
def _create_and_validate_pf(self, fip_id, internal_port_id,
|
|
internal_ip_address, internal_port,
|
|
external_port, protocol):
|
|
created_pf = self.create_port_forwarding(
|
|
fip_id,
|
|
internal_port_id=internal_port_id,
|
|
internal_ip_address=internal_ip_address,
|
|
internal_port=internal_port,
|
|
external_port=external_port,
|
|
protocol=protocol)
|
|
self.assertEqual(internal_port, created_pf['internal_port'])
|
|
self.assertEqual(external_port, created_pf['external_port'])
|
|
self.assertEqual(protocol, created_pf['protocol'])
|
|
self.assertEqual(internal_ip_address,
|
|
created_pf['internal_ip_address'])
|
|
return created_pf
|
|
|
|
@decorators.idempotent_id('8202cded-7e82-4420-9585-c091105404f6')
|
|
def test_associate_2_port_forwardings_to_floating_ip(self):
|
|
fip = self.create_floatingip()
|
|
forwardings_data = [(1111, 2222), (3333, 4444)]
|
|
created_pfs = []
|
|
for data in forwardings_data:
|
|
internal_port = data[0]
|
|
external_port = data[1]
|
|
port = self.create_port(self.network)
|
|
created_pf = self._create_and_validate_pf(
|
|
fip_id=fip['id'],
|
|
internal_port_id=port['id'],
|
|
internal_ip_address=port['fixed_ips'][0]['ip_address'],
|
|
internal_port=internal_port, external_port=external_port,
|
|
protocol="tcp")
|
|
created_pfs.append(created_pf)
|
|
self._verify_created_pfs(fip['id'], created_pfs)
|
|
|
|
@decorators.idempotent_id('a7e6cc48-8a9b-49be-82fb-cef6f5c29381')
|
|
def test_associate_port_forwarding_to_2_fixed_ips(self):
|
|
fip = self.create_floatingip()
|
|
port = self.create_port(self.network)
|
|
internal_subnet_id = port['fixed_ips'][0]['subnet_id']
|
|
# Add a second fixed_ip address to port (same subnet)
|
|
port['fixed_ips'].append({'subnet_id': internal_subnet_id})
|
|
port = self.update_port(port, fixed_ips=port['fixed_ips'])
|
|
internal_ip_address1 = port['fixed_ips'][0]['ip_address']
|
|
internal_ip_address2 = port['fixed_ips'][1]['ip_address']
|
|
forwardings_data = [(4001, internal_ip_address1),
|
|
(4002, internal_ip_address2)]
|
|
created_pfs = []
|
|
for data in forwardings_data:
|
|
external_port = data[0]
|
|
internal_ip_address = data[1]
|
|
created_pf = self._create_and_validate_pf(
|
|
fip_id=fip['id'],
|
|
internal_port_id=port['id'],
|
|
internal_ip_address=internal_ip_address,
|
|
internal_port=123, external_port=external_port,
|
|
protocol="tcp")
|
|
created_pfs.append(created_pf)
|
|
self._verify_created_pfs(fip['id'], created_pfs)
|
|
|
|
@decorators.idempotent_id('6a34e811-66d1-4f63-aa4d-9013f15deb62')
|
|
def test_associate_port_forwarding_to_used_floating_ip(self):
|
|
port_for_fip = self.create_port(self.network)
|
|
fip = self.create_floatingip(port=port_for_fip)
|
|
port = self.create_port(self.network)
|
|
self.assertRaises(
|
|
exceptions.Conflict,
|
|
self.create_port_forwarding,
|
|
fip['id'],
|
|
internal_port_id=port['id'],
|
|
internal_ip_address=port['fixed_ips'][0]['ip_address'],
|
|
internal_port=1111, external_port=2222,
|
|
protocol="tcp")
|
|
|
|
@decorators.idempotent_id('4ca72d40-93e4-485f-a876-76caf33c1fe6')
|
|
def test_associate_port_forwarding_to_port_with_fip(self):
|
|
port = self.create_port(self.network)
|
|
self.create_floatingip(port=port)
|
|
fip_for_pf = self.create_floatingip()
|
|
self.assertRaises(
|
|
exceptions.Conflict,
|
|
self.create_port_forwarding,
|
|
fip_for_pf['id'],
|
|
internal_port_id=port['id'],
|
|
internal_ip_address=port['fixed_ips'][0]['ip_address'],
|
|
internal_port=1111, external_port=2222,
|
|
protocol="tcp")
|