Add more port_forwarding tests
Extend set of tests for the port_forwarding feature to automate coverage
of specific cases:
- Port forwaring on neutron ports with multiple fixed ips
- Out of range values for port
- Forward communication to multiple fixed IPs of a particular Neutron port
- Editing and Deleting UDP port forwarding rule
Related-Bug: #1897753
Change-Id: I0fbf0a12c050a5a7184c96b62eee32139bc820b4
diff --git a/neutron_tempest_plugin/api/base.py b/neutron_tempest_plugin/api/base.py
index 7cf8dd4..d63dec8 100644
--- a/neutron_tempest_plugin/api/base.py
+++ b/neutron_tempest_plugin/api/base.py
@@ -702,6 +702,12 @@
return pf
@classmethod
+ def update_port_forwarding(cls, fip_id, pf_id, client=None, **kwargs):
+ """Wrapper utility for update_port_forwarding."""
+ client = client or cls.client
+ return client.update_port_forwarding(fip_id, pf_id, **kwargs)
+
+ @classmethod
def delete_port_forwarding(cls, pf, client=None):
"""Delete port forwarding
diff --git a/neutron_tempest_plugin/api/test_port_forwarding_negative.py b/neutron_tempest_plugin/api/test_port_forwarding_negative.py
index 76dd6ee..386cbed 100644
--- a/neutron_tempest_plugin/api/test_port_forwarding_negative.py
+++ b/neutron_tempest_plugin/api/test_port_forwarding_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.
+
from tempest.lib.common.utils import data_utils
from tempest.lib import decorators
from tempest.lib import exceptions
@@ -81,3 +82,43 @@
internal_ip_address=port['fixed_ips'][0]['ip_address'],
internal_port=1111, external_port=5555,
protocol="tcp")
+
+ @decorators.attr(type='negative')
+ @decorators.idempotent_id('e9d3ffb6-e5bf-421d-acaa-ee6010dfbf14')
+ def test_out_of_range_ports(self):
+ port = self.create_port(self.network)
+ fip_for_pf = self.create_floatingip()
+
+ pf_params = {
+ 'internal_port_id': port['id'],
+ 'internal_ip_address': port['fixed_ips'][0]['ip_address'],
+ 'internal_port': 1111,
+ 'external_port': 3333,
+ 'protocol': "tcp"}
+ pf = self.create_port_forwarding(fip_for_pf['id'], **pf_params)
+
+ # Check: Invalid input for external_port update
+ self.assertRaises(
+ exceptions.BadRequest,
+ self.update_port_forwarding,
+ fip_for_pf['id'], pf['id'], external_port=108343)
+
+ # Check: Invalid input for internal_port update
+ self.assertRaises(
+ exceptions.BadRequest,
+ self.update_port_forwarding,
+ fip_for_pf['id'], pf['id'], internal_port=108343)
+
+ # Check: Invalid input for external_port create
+ pf_params['internal_port'] = 4444
+ pf_params['external_port'] = 333333
+ self.assertRaises(
+ exceptions.BadRequest,
+ self.create_port_forwarding, fip_for_pf['id'], **pf_params)
+
+ # Check: Invalid input for internal_port create
+ pf_params['internal_port'] = 444444
+ pf_params['external_port'] = 3333
+ self.assertRaises(
+ exceptions.BadRequest,
+ self.create_port_forwarding, fip_for_pf['id'], **pf_params)
diff --git a/neutron_tempest_plugin/api/test_port_forwardings.py b/neutron_tempest_plugin/api/test_port_forwardings.py
index 82c3a34..79cce39 100644
--- a/neutron_tempest_plugin/api/test_port_forwardings.py
+++ b/neutron_tempest_plugin/api/test_port_forwardings.py
@@ -124,31 +124,10 @@
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)
-
+ 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(forwardings_data), len(fip['port_forwardings']))
+ fip = self.client.show_floatingip(fip_id)['floatingip']
+ self.assertEqual(len(created_pfs), len(fip['port_forwardings']))
for pf in created_pfs:
expected_pf = {
'external_port': pf['external_port'],
@@ -160,13 +139,73 @@
# Test list of port forwardings
port_forwardings = self.client.list_port_forwardings(
fip['id'])['port_forwardings']
- self.assertEqual(len(forwardings_data), len(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)