diff --git a/.zuul.yaml b/.zuul.yaml
index db5e1a5..2420a0c 100644
--- a/.zuul.yaml
+++ b/.zuul.yaml
@@ -1,6 +1,6 @@
- project:
templates:
- - openstack-lower-constraints-jobs
- openstack-python-jobs-neutron
+ - openstack-lower-constraints-jobs
- openstack-python36-jobs
- check-requirements
diff --git a/setup.cfg b/setup.cfg
index 25f0ec6..076afe0 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -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
diff --git a/tox.ini b/tox.ini
index b925e9a..530cddd 100644
--- a/tox.ini
+++ b/tox.ini
@@ -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
diff --git a/vmware_nsx_tempest_plugin/common/waiters.py b/vmware_nsx_tempest_plugin/common/waiters.py
index 6e2ecda..68215fe 100644
--- a/vmware_nsx_tempest_plugin/common/waiters.py
+++ b/vmware_nsx_tempest_plugin/common/waiters.py
@@ -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:
diff --git a/vmware_nsx_tempest_plugin/tests/nsxv/scenario/test_admin_policy_basic_ops.py b/vmware_nsx_tempest_plugin/tests/nsxv/scenario/test_admin_policy_basic_ops.py
index 57ba6c6..7f7a3c2 100644
--- a/vmware_nsx_tempest_plugin/tests/nsxv/scenario/test_admin_policy_basic_ops.py
+++ b/vmware_nsx_tempest_plugin/tests/nsxv/scenario/test_admin_policy_basic_ops.py
@@ -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,11 +423,13 @@ 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)
- if x > '2']
+ 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)
- if x < '3']
+ for x, y in six.iteritems(
+ self.server_ips)
+ if x < '3']
self.run_policy_AA_on_ames_AA_on_jpl()
self.run_policy_AA_on_ames_BB_on_jpl()
diff --git a/vmware_nsx_tempest_plugin/tests/scenario/test_verify_nat_fw_order.py b/vmware_nsx_tempest_plugin/tests/scenario/test_verify_nat_fw_order.py
new file mode 100644
index 0000000..53320bf
--- /dev/null
+++ b/vmware_nsx_tempest_plugin/tests/scenario/test_verify_nat_fw_order.py
@@ -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=[])