neutron-tempest-plugin/neutron_tempest_plugin/api/test_port_forwardings.py
zhouhenglc bdacaa20ef list floating ip api contains port forwarding id
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
2022-09-09 09:30:29 +08:00

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")