x/vmware-nsx-tempest-plugin
Code Issues Proposed changes

P0 testcase Automation of NAT and Firewall order feature

Browse Source 
Change-Id: I521f447f698e93b11612da2ce98236d80580c1e6
This commit is contained in:
shubhamk 2020-05-28 06:01:42 +00:00 committed by Shubham Kadam
parent 08cba39042
commit 6ecd11d457
6 changed files with 326 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.zuul.yaml
View File

@ -1,6 +1,6 @@
- project:
templates:
- openstack-lower-constraints-jobs
- openstack-python-jobs-neutron
- openstack-lower-constraints-jobs
- openstack-python36-jobs
- check-requirements

3
setup.cfg
View File

@ -6,6 +6,7 @@ description-file =
author = OpenStack
author-email = openstack-dev@lists.openstack.org
home-page = http://www.openstack.org/
python-requires = >=3.6
classifier =
Environment :: OpenStack
Intended Audience :: Information Technology
@ -13,8 +14,6 @@ classifier =
License :: OSI Approved :: Apache Software License
Operating System :: POSIX :: Linux
Programming Language :: Python
Programming Language :: Python :: 2
Programming Language :: Python :: 2.7
Programming Language :: Python :: 3
Programming Language :: Python :: 3.3
Programming Language :: Python :: 3.6

7
tox.ini
View File

@ -1,10 +1,11 @@
[tox]
minversion = 2.0
envlist = py36,py27,pep8,docs
minversion = 3.1
envlist = py36,pep8,docs
skipsdist = True
[testenv]
basepython = python3
usedevelop = True
install_command = pip install {opts} {packages}
setenv =
@ -40,8 +41,6 @@ commands =
flake8
{[testenv:bandit]commands}
[testenv:py27]
setenv = OS_FAIL_ON_MISSING_DEPS=1
[testenv:venv]
basepython = python3

15
vmware_nsx_tempest_plugin/common/waiters.py
View File

@ -167,15 +167,12 @@ def wait_for_recordset_status(client, zone_id, recordset_id, status):
return
if int(time.time()) - start >= client.build_timeout:
message = ('Recordset %(recordset_id)s failed to reach '
'status=%(status) within the required time '
'(%(timeout)s s). Current '
'status: %(status_curr)s' %
{'recordset_id': recordset_id,
'status': status,
'status_curr': status_curr,
'timeout': client.build_timeout})
message = 'Recordset %s s failed to reach ' \
'status=%s within the required time ' \
'(%s s s). Current ' \
'status: %s s' % \
(recordset_id, status,
status_curr, client.build_timeout)
caller = test_utils.find_test_caller()
if caller:

8
vmware_nsx_tempest_plugin/tests/nsxv/scenario/test_admin_policy_basic_ops.py
View File

@ -341,7 +341,7 @@ class TestAdminPolicyBasicOps(dmgr.TopoDeployScenarioManager):
except Exception as ex:
ssh_client = None
msg = (self.exc_msg +
("\n**FAIL to ssh to host[%s=%s]\n%s" %
("\n**FAIL to ssh to host %s \n%s" %
(host_id, str(ex))))
self.assertTrue(ssh_client, msg)
self.log_exc_msg(
@ -423,10 +423,12 @@ class TestAdminPolicyBasicOps(dmgr.TopoDeployScenarioManager):
router_type))
self.create_nasa_topo(router_type)
self.jpl_private_ips = [y['fixed_ip_address']
for x, y in six.iteritems(self.server_ips)
for x, y in six.iteritems(
self.server_ips)
if x > '2']
self.ames_private_ips = [y['fixed_ip_address']
for x, y in six.iteritems(self.server_ips)
for x, y in six.iteritems(
self.server_ips)
if x < '3']
self.run_policy_AA_on_ames_AA_on_jpl()

308
vmware_nsx_tempest_plugin/tests/scenario/test_verify_nat_fw_order.py Normal file
View File

@ -0,0 +1,308 @@
# Copyright 2017 VMware 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.
import os
import time
from tempest import config
from tempest.lib.common.utils import data_utils
from tempest.lib.common.utils import test_utils
from tempest.lib import decorators
from vmware_nsx_tempest_plugin.common import constants
from vmware_nsx_tempest_plugin.lib import feature_manager
from vmware_nsx_tempest_plugin.services import nsxp_client
from vmware_nsx_tempest_plugin.services import nsxv3_client
CONF = config.CONF
LOG = constants.log.getLogger(__name__)
class TestVerifyFwNatOrder(feature_manager.FeatureManager):
"""Test New Cases Scenario
"""
@classmethod
def setup_clients(cls):
super(TestVerifyFwNatOrder, cls).setup_clients()
cls.cmgr_adm = cls.get_client_manager('admin')
cls.cmgr_alt = cls.get_client_manager('alt')
cls.cmgr_adm = cls.get_client_manager('admin')
cls.routers_client = cls.cmgr_adm.routers_client
cls.networks_client = cls.cmgr_adm.networks_client
cls.subnets_client = cls.cmgr_adm.subnets_client
cls.sec_rule_client = cls.cmgr_adm.security_group_rules_client
cls.sec_client = cls.cmgr_adm.security_groups_client
@classmethod
def resource_setup(cls):
super(TestVerifyFwNatOrder, cls).resource_setup()
cls.nsx = nsxv3_client.NSXV3Client(CONF.nsxv3.nsx_manager,
CONF.nsxv3.nsx_user,
CONF.nsxv3.nsx_password)
cls.nsxp = nsxp_client.NSXPClient(CONF.nsxv3.nsx_manager,
CONF.nsxv3.nsx_user,
CONF.nsxv3.nsx_password)
def _test_ping_from_external_network(self, fip_ip):
out = os.popen('ping -c 5 %s' % fip_ip).read().strip()
return out
@decorators.idempotent_id('2317449c-03ac-9106-c317-09956daa46c3')
def test_verfiy_nat_fw_order_external_fw(self):
"""
Create NAT and Firewall rules on router.
Verify order of NAT and Firewall.
"""
rtr_name = data_utils.rand_name(name='tempest-router')
network_name = data_utils.rand_name(name='tempest-net')
subnet_name = data_utils.rand_name(name='tempest-subnet')
router_state = self.create_topology_router(
rtr_name, set_gateway=True,
routers_client=self.cmgr_adm.routers_client)
network_state = self.create_topology_network(
network_name, networks_client=self.cmgr_adm.networks_client)
subnet_state = self.create_topology_subnet(
subnet_name, network_state,
subnets_client=self.cmgr_adm.subnets_client)
interface = self.cmgr_adm.routers_client.add_router_interface(
router_state['id'], subnet_id=subnet_state["id"])
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
self.cmgr_adm.routers_client.remove_router_interface,
router_state['id'], subnet_id=subnet_state["id"])
security_group = self._create_security_group(
security_group_rules_client=self.cmgr_adm.
security_group_rules_client,
security_groups_client=self.cmgr_adm.security_groups_client)
image_id = self.get_glance_image_id(["cirros", "esx"])
security_groups = [{'name': security_group['name']}]
server1 = self.create_topology_instance(
"state_vm_1", [network_state],
create_floating_ip=True, image_id=image_id, clients=self.cmgr_adm,
security_groups=security_groups)
server2 = self.create_topology_instance(
"state_vm_2", [network_state],
create_floating_ip=True, image_id=image_id, clients=self.cmgr_adm,
security_groups=security_groups)
if CONF.network.backend == 'nsxp':
time.sleep(constants.NSXP_BACKEND_SMALL_TIME_INTERVAL)
nsx_router = self.nsxp.get_logical_router(router_state['name'],
router_state['id'])
nat_rules = self.nsxp.get_logical_router_nat_rules(nsx_router)
for nat_rule in nat_rules:
if nat_rule['firewall_match'] == 'BYPASS':
continue
self.assertEqual('MATCH_EXTERNAL_ADDRESS',
nat_rule['firewall_match'])
nsx_router = self.nsx.get_logical_router(router_state['name'],
router_state['id'])
nat_rules = self.nsx.get_logical_router_nat_rules(nsx_router)
for nat_rule in nat_rules:
if nat_rule['firewall_match'] == 'BYPASS':
continue
self.assertEqual('MATCH_EXTERNAL_ADDRESS',
nat_rule['firewall_match'])
ext_network = self.cmgr_adm.networks_client.show_network(
CONF.network.public_network_id)['network']
ext_subnet = self.cmgr_adm.subnets_client.show_subnet(
ext_network['subnets'][0])['subnet']
fw_rules = self.create_firewall_rule(
name='test_rule', protocol='icmp',
action="allow", destination_ip_address=ext_subnet["cidr"])
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
self.fwaas_v2_client.delete_firewall_v2_rule,
fw_rules['firewall_rule']['id'])
rules = []
# Check firewall rule
rules.append(fw_rules['firewall_rule']['id'])
policy_name = data_utils.rand_name('fw-policy-')
# Create firewall policy
fw_policy = self.create_firewall_policy(
name=policy_name, firewall_rules=rules,
project_id=router_state['project_id'])
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
self.fwaas_v2_client.delete_firewall_v2_policy,
fw_policy['firewall_policy']['id'])
show_policy = self.show_firewall_policy(
fw_policy['firewall_policy']['id'])
# Check firewall policy
self.assertEqual(
show_policy.get('firewall_policy')['name'],
policy_name)
self.assertEqual(show_policy.get('firewall_policy')
['firewall_rules'], rules)
policy_id = fw_policy['firewall_policy']['id']
group_name = data_utils.rand_name('fw-group-')
# Create firewall group
fw_group = self.create_firewall_group(
name=group_name,
ingress_firewall_policy_id=policy_id,
egress_firewall_policy_id=policy_id,
ports=[interface['port_id']],
project_id=router_state['project_id'])
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
self.fwaas_v2_client.update_firewall_v2_group,
fw_group["firewall_group"]["id"], ports=[])
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
self.fwaas_v2_client.delete_firewall_v2_group,
fw_group["firewall_group"]["id"])
self._wait_firewall_ready(fw_group["firewall_group"]["id"])
# Verify traffic to vm
fip = server1["floating_ips"][0]["floating_ip_address"]
out = self._test_ping_from_external_network(fip)
self.assertIn("64 bytes from ", str(out))
fip = server2["floating_ips"][0]["floating_ip_address"]
out = self._test_ping_from_external_network(fip)
self.assertIn("64 bytes from ", str(out))
# Update destination ip address in firewall rule
fw_rules = self.update_firewall_rule(
fw_rules['firewall_rule']['id'],
destination_ip_address=subnet_state['cidr'])
if CONF.network.backend == 'nsxp':
time.sleep(constants.NSXP_BACKEND_SMALL_TIME_INTERVAL)
# Verify traffic to vm
fip = server1["floating_ips"][0]["floating_ip_address"]
out = self._test_ping_from_external_network(fip)
self.assertNotIn("64 bytes from ", str(out))
fip = server2["floating_ips"][0]["floating_ip_address"]
out = self._test_ping_from_external_network(fip)
self.assertNotIn("64 bytes from ", str(out))
self.fwaas_v2_client.update_firewall_v2_group(
fw_group["firewall_group"]["id"], ports=[])
@decorators.idempotent_id('2317449c-14bd-0317-c317-09956daa46c3')
def test_verfiy_nat_fw_order_internal_fw(self):
"""
Create NAT and Firewall rules on router.
Verify order of NAT and Firewall.
"""
rtr_name = data_utils.rand_name(name='tempest-router')
network_name = data_utils.rand_name(name='tempest-net')
subnet_name = data_utils.rand_name(name='tempest-subnet')
router_state = self.create_topology_router(
rtr_name, set_gateway=True,
routers_client=self.cmgr_adm.routers_client)
network_state = self.create_topology_network(
network_name, networks_client=self.cmgr_adm.networks_client)
subnet_state = self.create_topology_subnet(
subnet_name, network_state,
subnets_client=self.cmgr_adm.subnets_client)
interface = self.cmgr_adm.routers_client.add_router_interface(
router_state['id'], subnet_id=subnet_state["id"])
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
self.cmgr_adm.routers_client.remove_router_interface,
router_state['id'], subnet_id=subnet_state["id"])
security_group = self._create_security_group(
security_group_rules_client=self.cmgr_adm.
security_group_rules_client,
security_groups_client=self.cmgr_adm.security_groups_client)
image_id = self.get_glance_image_id(["cirros", "esx"])
security_groups = [{'name': security_group['name']}]
server1 = self.create_topology_instance(
"state_vm_1", [network_state],
create_floating_ip=True, image_id=image_id, clients=self.cmgr_adm,
security_groups=security_groups)
server2 = self.create_topology_instance(
"state_vm_2", [network_state],
create_floating_ip=True, image_id=image_id, clients=self.cmgr_adm,
security_groups=security_groups)
if CONF.network.backend == 'nsxp':
time.sleep(constants.NSXP_BACKEND_SMALL_TIME_INTERVAL)
nsx_router = self.nsxp.get_logical_router(router_state['name'],
router_state['id'])
nat_rules = self.nsxp.get_logical_router_nat_rules(nsx_router)
for nat_rule in nat_rules:
if nat_rule['firewall_match'] == 'BYPASS':
continue
self.assertEqual('MATCH_INTERNAL_ADDRESS',
nat_rule['firewall_match'])
nsx_router = self.nsx.get_logical_router(router_state['name'],
router_state['id'])
nat_rules = self.nsx.get_logical_router_nat_rules(nsx_router)
for nat_rule in nat_rules:
if nat_rule['firewall_match'] == 'BYPASS':
continue
self.assertEqual('MATCH_INTERNAL_ADDRESS',
nat_rule['firewall_match'])
ext_network = self.cmgr_adm.networks_client.show_network(
CONF.network.public_network_id)['network']
ext_subnet = self.cmgr_adm.subnets_client.show_subnet(
ext_network['subnets'][0])['subnet']
fw_rules = self.create_firewall_rule(
name='test_rule', protocol='icmp',
action="allow",
destination_ip_address=subnet_state['cidr'])
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
self.fwaas_v2_client.delete_firewall_v2_rule,
fw_rules['firewall_rule']['id'])
rules = []
# Check firewall rule
rules.append(fw_rules['firewall_rule']['id'])
policy_name = data_utils.rand_name('fw-policy-')
# Create firewall policy
fw_policy = self.create_firewall_policy(
name=policy_name, firewall_rules=rules,
project_id=router_state['project_id'])
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
self.fwaas_v2_client.delete_firewall_v2_policy,
fw_policy['firewall_policy']['id'])
show_policy = self.show_firewall_policy(
fw_policy['firewall_policy']['id'])
# Check firewall policy
self.assertEqual(
show_policy.get('firewall_policy')['name'],
policy_name)
self.assertEqual(show_policy.get('firewall_policy')
['firewall_rules'], rules)
policy_id = fw_policy['firewall_policy']['id']
group_name = data_utils.rand_name('fw-group-')
# Create firewall group
fw_group = self.create_firewall_group(
name=group_name,
ingress_firewall_policy_id=policy_id,
egress_firewall_policy_id=policy_id,
ports=[interface['port_id']],
project_id=router_state['project_id'])
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
self.fwaas_v2_client.update_firewall_v2_group,
fw_group["firewall_group"]["id"], ports=[])
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
self.fwaas_v2_client.delete_firewall_v2_group,
fw_group["firewall_group"]["id"])
self._wait_firewall_ready(fw_group["firewall_group"]["id"])
# Verify traffic to vm
fip = server1["floating_ips"][0]["floating_ip_address"]
out = self._test_ping_from_external_network(fip)
self.assertIn("64 bytes from ", str(out))
fip = server2["floating_ips"][0]["floating_ip_address"]
out = self._test_ping_from_external_network(fip)
self.assertIn("64 bytes from ", str(out))
# Update destination ip address in firewall rule
fw_rules = self.update_firewall_rule(
fw_rules['firewall_rule']['id'],
destination_ip_address=ext_subnet['cidr'])
if CONF.network.backend == 'nsxp':
time.sleep(constants.NSXP_BACKEND_SMALL_TIME_INTERVAL)
# Verify traffic to vm
fip = server1["floating_ips"][0]["floating_ip_address"]
out = self._test_ping_from_external_network(fip)
self.assertNotIn("64 bytes from ", str(out))
fip = server2["floating_ips"][0]["floating_ip_address"]
out = self._test_ping_from_external_network(fip)
self.assertNotIn("64 bytes from ", str(out))
self.fwaas_v2_client.update_firewall_v2_group(
fw_group["firewall_group"]["id"], ports=[])