openstack/neutron-tempest-plugin
Code Issues Proposed changes

Merge "Add base API tests for port forwarding"

Browse Source
This commit is contained in:
Zuul 2019-05-29 22:04:34 +00:00 committed by Gerrit Code Review
commit c89a00a3ba
4 changed files with 293 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
.zuul.yaml
View File

@ -26,6 +26,7 @@
- dns-domain-ports
- dns-integration
- empty-string-filtering
- expose-port-forwarding-in-fip
- ext-gw-mode
- external-net
- extra_dhcp_opt
@ -33,6 +34,7 @@
- filter-validation
- fip-port-details
- flavors
- floating-ip-port-forwarding
- floatingip-pools
- ip-substring-filtering
- l3-flavors
@ -94,6 +96,7 @@
neutron-trunk: true
neutron-uplink-status-propagation: true
neutron-network-segment-range: true
neutron-port-forwarding: true
devstack_local_conf:
post-config:
$NEUTRON_CONF:
@ -249,6 +252,7 @@
- dns-domain-ports
- dns-integration
- empty-string-filtering
- expose-port-forwarding-in-fip
- ext-gw-mode
- external-net
- extra_dhcp_opt
@ -257,6 +261,7 @@
- fip-port-details
- flavors
- floatingip-pools
- floating-ip-port-forwarding
- ip-substring-filtering
- l3-flavors
- l3-ha
@ -325,12 +330,14 @@
- dns-domain-ports
- dns-integration
- empty-string-filtering
- expose-port-forwarding-in-fip
- ext-gw-mode
- external-net
- extra_dhcp_opt
- extraroute
- fip-port-details
- flavors
- floating-ip-port-forwarding
- ip-substring-filtering
- l3-flavors
- l3-ha

65
neutron_tempest_plugin/api/base.py
View File

@ -117,6 +117,7 @@ class BaseNetworkTest(test.BaseTestCase):
cls.ports = []
cls.routers = []
cls.floating_ips = []
cls.port_forwardings = []
cls.metering_labels = []
cls.service_profiles = []
cls.flavors = []
@ -144,6 +145,10 @@ class BaseNetworkTest(test.BaseTestCase):
for trunk in cls.trunks:
cls._try_delete_resource(cls.delete_trunk, trunk)
# Clean up port forwardings
for pf in cls.port_forwardings:
cls._try_delete_resource(cls.delete_port_forwarding, pf)
# Clean up floating IPs
for floating_ip in cls.floating_ips:
cls._try_delete_resource(cls.delete_floatingip, floating_ip)
@ -651,6 +656,66 @@ class BaseNetworkTest(test.BaseTestCase):
client = client or floating_ip.get('client') or cls.client
client.delete_floatingip(floating_ip['id'])
@classmethod
def create_port_forwarding(cls, fip_id, internal_port_id,
internal_port, external_port,
internal_ip_address=None, protocol="tcp",
client=None):
"""Creates a port forwarding.
Create a port forwarding and schedule it for later deletion.
If a client is passed, then it is used for deleting the PF too.
:param fip_id: The ID of the floating IP address.
:param internal_port_id: The ID of the Neutron port associated to
the floating IP port forwarding.
:param internal_port: The TCP/UDP/other protocol port number of the
Neutron port fixed IP address associated to the floating ip
port forwarding.
:param external_port: The TCP/UDP/other protocol port number of
the port forwarding floating IP address.
:param internal_ip_address: The fixed IPv4 address of the Neutron
port associated to the floating IP port forwarding.
:param protocol: The IP protocol used in the floating IP port
forwarding.
:param client: network client to be used for creating and cleaning up
the floating IP port forwarding.
"""
client = client or cls.client
pf = client.create_port_forwarding(
fip_id, internal_port_id, internal_port, external_port,
internal_ip_address, protocol)['port_forwarding']
# save ID of floating IP associated with port forwarding for final
# cleanup
pf['floatingip_id'] = fip_id
# save client to be used later in cls.delete_port_forwarding
# for final cleanup
pf['client'] = client
cls.port_forwardings.append(pf)
return pf
@classmethod
def delete_port_forwarding(cls, pf, client=None):
"""Delete port forwarding
:param client: Client to be used
If client is not given it will use the client used to create
the port forwarding, or cls.client if unknown.
"""
client = client or pf.get('client') or cls.client
client.delete_port_forwarding(pf['floatingip_id'], pf['id'])
@classmethod
def create_router_interface(cls, router_id, subnet_id):
"""Wrapper utility that returns a router interface."""

172
neutron_tempest_plugin/api/test_port_forwardings.py Normal file
View File

@ -0,0 +1,172 @@
# 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'])
# 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'])
@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']))
@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_port_forwarding(
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")
self.assertEqual(internal_port, created_pf['internal_port'])
self.assertEqual(external_port, created_pf['external_port'])
self.assertEqual('tcp', created_pf['protocol'])
self.assertEqual(port['fixed_ips'][0]['ip_address'],
created_pf['internal_ip_address'])
created_pfs.append(created_pf)
# Check that all PFs are visible in Floating IP details
fip = self.client.show_floatingip(fip['id'])['floatingip']
self.assertEqual(len(forwardings_data), len(fip['port_forwardings']))
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']}
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(forwardings_data), 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)
@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")

49
neutron_tempest_plugin/services/network/json/network_client.py
View File

@ -938,6 +938,55 @@ class NetworkClientJSON(service_client.RestClient):
body = jsonutils.loads(resp_body)
return service_client.ResponseBody(put_resp, body)
def create_port_forwarding(self, fip_id, internal_port_id,
internal_port, external_port,
internal_ip_address=None, protocol='tcp'):
post_body = {'port_forwarding': {
'protocol': protocol,
'internal_port_id': internal_port_id,
'internal_port': int(internal_port),
'external_port': int(external_port)}}
if internal_ip_address:
post_body['port_forwarding']['internal_ip_address'] = (
internal_ip_address)
body = jsonutils.dumps(post_body)
uri = '%s/floatingips/%s/port_forwardings' % (self.uri_prefix, fip_id)
resp, body = self.post(uri, body)
self.expected_success(201, resp.status)
body = jsonutils.loads(body)
return service_client.ResponseBody(resp, body)
def get_port_forwarding(self, fip_id, pf_id):
uri = '%s/floatingips/%s/port_forwardings/%s' % (self.uri_prefix,
fip_id, pf_id)
get_resp, get_resp_body = self.get(uri)
self.expected_success(200, get_resp.status)
body = jsonutils.loads(get_resp_body)
return service_client.ResponseBody(get_resp, body)
def list_port_forwardings(self, fip_id):
uri = '%s/floatingips/%s/port_forwardings' % (self.uri_prefix, fip_id)
resp, body = self.get(uri)
self.expected_success(200, resp.status)
body = jsonutils.loads(body)
return service_client.ResponseBody(resp, body)
def update_port_forwarding(self, fip_id, pf_id, **kwargs):
uri = '%s/floatingips/%s/port_forwardings/%s' % (self.uri_prefix,
fip_id, pf_id)
put_body = jsonutils.dumps({'port_forwarding': kwargs})
put_resp, resp_body = self.put(uri, put_body)
self.expected_success(200, put_resp.status)
body = jsonutils.loads(resp_body)
return service_client.ResponseBody(put_resp, body)
def delete_port_forwarding(self, fip_id, pf_id):
uri = '%s/floatingips/%s/port_forwardings/%s' % (self.uri_prefix,
fip_id, pf_id)
resp, body = self.delete(uri)
self.expected_success(204, resp.status)
service_client.ResponseBody(resp, body)
def create_network_keystone_v3(self, name, project_id, tenant_id=None):
uri = '%s/networks' % self.uri_prefix
post_data = {