aboutsummaryrefslogtreecommitdiffstats
path: root/sfc
diff options
context:
space:
mode:
authorPanagiotis Karalis <pkaralis@intracom-telecom.com>2018-02-14 15:28:02 +0200
committerPanagiotis Karalis <pkaralis@intracom-telecom.com>2018-05-15 11:42:44 +0300
commit23c9ae8fa5db3ea02077d97452ea503097cf985e (patch)
tree1e076dcabac20963465c7380eab4ba16d0b50f6a /sfc
parent5f8eb6bf01ca3ef240cfcdee297327b21b33f35a (diff)
Clean up Test Case code
Too much of duplicated code has been obeserved in test cases. With this patch a small refactoring is necessery to our test cases code. So they can be more generic and more flexible. To achieve that the addition of new parent class is essential, hence all the rest test case classes will inherit from the aforementioned parent class. Change-Id: I6a50d8330a06321742b3bb03cc999682d4a4e6a1 Signed-off-by: Panagiotis Karalis <pkaralis@intracom-telecom.com>
Diffstat (limited to 'sfc')
-rw-r--r--sfc/tests/functest/config.yaml35
-rw-r--r--sfc/tests/functest/run_sfc_tests.py12
-rw-r--r--sfc/tests/functest/sfc_chain_deletion.py288
-rw-r--r--sfc/tests/functest/sfc_one_chain_two_service_functions.py301
-rw-r--r--sfc/tests/functest/sfc_parent_function.py519
-rw-r--r--sfc/tests/functest/sfc_symmetric_chain.py397
-rw-r--r--sfc/tests/functest/sfc_two_chains_SSH_and_HTTP.py337
7 files changed, 870 insertions, 1019 deletions
diff --git a/sfc/tests/functest/config.yaml b/sfc/tests/functest/config.yaml
index cad3cf72..d18a0033 100644
--- a/sfc/tests/functest/config.yaml
+++ b/sfc/tests/functest/config.yaml
@@ -28,6 +28,7 @@ defaults:
testcases:
sfc_one_chain_two_service_functions:
+ class_name: "SfcOneChainTwoServiceTC"
enabled: true
order: 0
description: "ODL-SFC Testing SFs when they are located on the same chain"
@@ -37,11 +38,20 @@ testcases:
subnet_cidr: "11.0.0.0/24"
secgroup_name: "example-sg"
secgroup_descr: "Example Security group"
+ vnf_names:
+ - 'testVNF1'
+ - 'testVNF2'
+ supported_installers:
+ - 'fuel'
+ - 'apex'
+ - 'osa'
+ - 'compass'
test_vnfd_red: "test-one-chain-vnfd1.yaml"
test_vnfd_blue: "test-one-chain-vnfd2.yaml"
test_vnffgd_red: "test-one-chain-vnffgd.yaml"
sfc_two_chains_SSH_and_HTTP:
+ class_name: "SfcTwoChainsSSHandHTTP"
enabled: false
order: 1
description: "ODL-SFC tests with two chains and one SF per chain"
@@ -51,12 +61,21 @@ testcases:
subnet_cidr: "11.0.0.0/24"
secgroup_name: "example-sg"
secgroup_descr: "Example Security group"
+ vnf_names:
+ - 'testVNF1'
+ - 'testVNF2'
+ supported_installers:
+ - 'fuel'
+ - 'apex'
+ - 'osa'
+ - 'compass'
test_vnfd_red: "test-two-chains-vnfd1.yaml"
test_vnfd_blue: "test-two-chains-vnfd2.yaml"
test_vnffgd_red: "test-two-chains-vnffgd1.yaml"
test_vnffgd_blue: "test-two-chains-vnffgd2.yaml"
sfc_symmetric_chain:
+ class_name: "SfcSymmetricChain"
enabled: false
order: 2
description: "Verify the behavior of a symmetric service chain"
@@ -66,11 +85,19 @@ testcases:
subnet_cidr: "11.0.0.0/24"
secgroup_name: "example-sg"
secgroup_descr: "Example Security group"
+ vnf_names:
+ - 'testVNF1'
+ supported_installers:
+ - 'fuel'
+ - 'apex'
+ - 'osa'
+ - 'compass'
test_vnfd: "test-symmetric-vnfd.yaml"
test_vnffgd: "test-symmetric-vnffgd.yaml"
source_port: 22222
sfc_chain_deletion:
+ class_name: "SfcChainDeletion"
enabled: false
order: 3
description: "Verify if chains work correctly after deleting one"
@@ -80,5 +107,13 @@ testcases:
subnet_cidr: "11.0.0.0/24"
secgroup_name: "example-sg"
secgroup_descr: "Example Security group"
+ vnf_names:
+ - 'testVNF1'
+ - 'testVNF2'
+ supported_installers:
+ - 'fuel'
+ - 'apex'
+ - 'osa'
+ - 'compass'
test_vnfd_red: "test-one-chain-vnfd1.yaml"
test_vnffgd_red: "test-deletion-vnffgd.yaml"
diff --git a/sfc/tests/functest/run_sfc_tests.py b/sfc/tests/functest/run_sfc_tests.py
index dba73b98..015255a6 100644
--- a/sfc/tests/functest/run_sfc_tests.py
+++ b/sfc/tests/functest/run_sfc_tests.py
@@ -149,12 +149,20 @@ class SfcFunctest(testcase.TestCase):
(test_name, test_descr))
logger.info(title)
logger.info("%s\n" % ("=" * len(title)))
- t = importlib.import_module(
+ module = importlib.import_module(
"sfc.tests.functest.{0}".format(test_name),
package=None)
+
+ testcase_config = sfc_config.TestcaseConfig(test_name)
+ supported_installers = test_cfg['supported_installers']
+ vnf_names = test_cfg['vnf_names']
+
+ tc_class = getattr(module, test_cfg['class_name'])
+ tc_instance = tc_class(testcase_config, supported_installers,
+ vnf_names)
start_time = time.time()
try:
- result, creators = t.main()
+ result, creators = tc_instance.run()
except Exception as e:
logger.error("Exception when executing: %s" % test_name)
logger.error(e)
diff --git a/sfc/tests/functest/sfc_chain_deletion.py b/sfc/tests/functest/sfc_chain_deletion.py
index 700b39a3..381a5167 100644
--- a/sfc/tests/functest/sfc_chain_deletion.py
+++ b/sfc/tests/functest/sfc_chain_deletion.py
@@ -8,258 +8,98 @@
# http://www.apache.org/licenses/LICENSE-2.0
#
-import os
-import sys
import threading
-import sfc.lib.openstack_utils as os_sfc_utils
import sfc.lib.odl_utils as odl_utils
-import opnfv.utils.ovs_logger as ovs_log
-import sfc.lib.topology_shuffler as topo_shuffler
import sfc.lib.config as sfc_config
import sfc.lib.test_utils as test_utils
-
+from sfc.tests.functest import sfc_parent_function
from opnfv.utils import opnfv_logger as logger
-from sfc.lib.results import Results
-from opnfv.deployment.factory import Factory as DeploymentFactory
logger = logger.Logger(__name__).getLogger()
-CLIENT = "client"
-SERVER = "server"
-COMMON_CONFIG = sfc_config.CommonConfig()
-TESTCASE_CONFIG = sfc_config.TestcaseConfig('sfc_chain_deletion')
-
-
-def main():
- deploymentHandler = DeploymentFactory.get_handler(
- COMMON_CONFIG.installer_type,
- COMMON_CONFIG.installer_ip,
- COMMON_CONFIG.installer_user,
- COMMON_CONFIG.installer_password,
- COMMON_CONFIG.installer_key_file)
-
- installer_type = os.environ.get("INSTALLER_TYPE")
-
- supported_installers = ['fuel', 'apex', 'osa', 'compass']
-
- if installer_type not in supported_installers:
- logger.error(
- '\033[91mYour installer is not supported yet\033[0m')
- sys.exit(1)
-
- installer_ip = os.environ.get("INSTALLER_IP")
- if not installer_ip:
- logger.error(
- '\033[91minstaller ip is not set\033[0m')
- logger.error(
- '\033[91mexport INSTALLER_IP=<ip>\033[0m')
- sys.exit(1)
-
- cluster = COMMON_CONFIG.installer_cluster
- openstack_nodes = (deploymentHandler.get_nodes({'cluster': cluster})
- if cluster is not None
- else deploymentHandler.get_nodes())
- controller_nodes = [node for node in openstack_nodes
- if node.is_controller()]
- compute_nodes = [node for node in openstack_nodes
- if node.is_compute()]
+class SfcChainDeletion(sfc_parent_function.SfcCommonTestCase):
+ """We create one client and one server using nova.
+ Then, a SF is created using tacker.
+ A service chain routing the traffic
+ throught this SF will be created as well.
+ After that the chain is deleted and re-created.
+ Finally, the vxlan tool is used in order to check a single
+ HTTP traffic scenario.
+ """
+ def run(self):
- odl_ip, odl_port = odl_utils.get_odl_ip_port(openstack_nodes)
+ logger.info("The test scenario %s is starting", __name__)
+ self.create_custom_vnfd(self.testcase_config.test_vnfd_red,
+ 'test-vnfd1')
+ self.create_custom_av(self.vnfs[0], 'test-vnfd1', 'test-vim')
- for compute in compute_nodes:
- logger.info("This is a compute: %s" % compute.ip)
+ self.create_chain(self.testcase_config)
- results = Results(COMMON_CONFIG.line_length)
- results.add_to_summary(0, "=")
- results.add_to_summary(2, "STATUS", "SUBTEST")
- results.add_to_summary(0, "=")
-
- openstack_sfc = os_sfc_utils.OpenStackSFC()
-
- custom_flv = openstack_sfc.create_flavor(
- COMMON_CONFIG.flavor,
- COMMON_CONFIG.ram_size_in_mb,
- COMMON_CONFIG.disk_size_in_gb,
- COMMON_CONFIG.vcpu_count)
- if not custom_flv:
- logger.error("Failed to create custom flavor")
- sys.exit(1)
-
- logger.info("Fetching the ssh clients...")
- tacker_client = os_sfc_utils.get_tacker_client()
- controller_clients = test_utils.get_ssh_clients(controller_nodes)
- compute_clients = test_utils.get_ssh_clients(compute_nodes)
-
- ovs_logger = ovs_log.OVSLogger(
- os.path.join(COMMON_CONFIG.sfc_test_dir, 'ovs-logs'),
- COMMON_CONFIG.functest_results_dir)
-
- image_creator = openstack_sfc.register_glance_image(
- COMMON_CONFIG.image_name,
- COMMON_CONFIG.image_url,
- COMMON_CONFIG.image_format,
- 'public')
-
- network, router = openstack_sfc.create_network_infrastructure(
- TESTCASE_CONFIG.net_name,
- TESTCASE_CONFIG.subnet_name,
- TESTCASE_CONFIG.subnet_cidr,
- TESTCASE_CONFIG.router_name)
-
- sg = openstack_sfc.create_security_group(TESTCASE_CONFIG.secgroup_name)
-
- vnf_names = ['testVNF1', 'testVNF2']
+ t1 = threading.Thread(target=odl_utils.wait_for_classification_rules,
+ args=(self.ovs_logger, self.compute_nodes,
+ self.odl_ip, self.odl_port,
+ self.client_instance.compute_host,
+ [self.neutron_port],))
- topo_seed = topo_shuffler.get_seed() # change to None for nova av zone
- testTopology = topo_shuffler.topology(vnf_names, openstack_sfc,
- seed=topo_seed)
+ try:
+ t1.start()
+ except Exception as e:
+ logger.error("Unable to start the thread that counts time %s" % e)
- logger.info('This test is run with the topology {0}'
- .format(testTopology['id']))
- logger.info('Topology description: {0}'
- .format(testTopology['description']))
+ logger.info("Assigning floating IPs to instances")
- client_instance, client_creator = openstack_sfc.create_instance(
- CLIENT, COMMON_CONFIG.flavor, image_creator, network, sg,
- av_zone=testTopology['client'])
+ self.assign_floating_ip_client_server()
+ self.assign_floating_ip_sfs()
- server_instance, server_creator = openstack_sfc.create_instance(
- SERVER, COMMON_CONFIG.flavor, image_creator, network, sg,
- av_zone=testTopology['server'])
+ self.check_floating_ips()
- server_ip = server_instance.ports[0].ips[0]['ip_address']
+ self.start_services_in_vm()
+ logger.info("Wait for ODL to update the classification rules in OVS")
+ t1.join()
- os_sfc_utils.register_vim(tacker_client, vim_file=COMMON_CONFIG.vim_file)
+ self.remove_vnffg('red_http', 'red')
+ self.check_deletion()
- tosca_red = os.path.join(COMMON_CONFIG.sfc_test_dir,
- COMMON_CONFIG.vnfd_dir,
- TESTCASE_CONFIG.test_vnfd_red)
- os_sfc_utils.create_vnfd(tacker_client,
- tosca_file=tosca_red,
- vnfd_name='test-vnfd1')
+ self.create_chain(self.testcase_config)
- default_param_file = os.path.join(
- COMMON_CONFIG.sfc_test_dir,
- COMMON_CONFIG.vnfd_dir,
- COMMON_CONFIG.vnfd_default_params_file)
+ t2 = threading.Thread(target=odl_utils.wait_for_classification_rules,
+ args=(self.ovs_logger, self.compute_nodes,
+ self.odl_ip, self.odl_port,
+ self.client_instance.compute_host,
+ [self.neutron_port],))
+ try:
+ t2.start()
+ except Exception as e:
+ logger.error("Unable to start the thread that counts time %s" % e)
- os_sfc_utils.create_vnf_in_av_zone(
- tacker_client, vnf_names[0], 'test-vnfd1', 'test-vim',
- default_param_file, testTopology[vnf_names[0]])
+ logger.info("Starting SSH firewall on %s" % self.fips_sfs[0])
+ test_utils.start_vxlan_tool(self.fips_sfs[0])
- vnf1_id = os_sfc_utils.wait_for_vnf(tacker_client, vnf_name=vnf_names[0])
- if vnf1_id is None:
- logger.error('ERROR while booting vnfs')
- sys.exit(1)
+ logger.info("Wait for ODL to update the classification rules in OVS")
+ t2.join()
- neutron_port = openstack_sfc.get_client_port(client_instance,
- client_creator)
- odl_utils.create_chain(tacker_client, default_param_file, neutron_port,
- COMMON_CONFIG, TESTCASE_CONFIG)
+ logger.info("Test HTTP")
+ results = self.present_results_allowed_http()
- # Start measuring the time it takes to implement the classification rules
- t1 = threading.Thread(target=odl_utils.wait_for_classification_rules,
- args=(ovs_logger, compute_nodes, odl_ip,
- odl_port, openstack_sfc.get_compute_client(),
- [neutron_port],))
+ self.vxlan_blocking_start(self.fips_sfs[0], "80")
- try:
- t1.start()
- except Exception as e:
- logger.error("Unable to start the thread that counts time %s" % e)
+ logger.info("Test HTTP again")
+ results = self.present_results_http()
- logger.info("Assigning floating IPs to instances")
- client_floating_ip = openstack_sfc.assign_floating_ip(router,
- client_instance,
- client_creator)
- server_floating_ip = openstack_sfc.assign_floating_ip(router,
- server_instance,
- server_creator)
- fips_sfs = openstack_sfc.assign_floating_ip_vnfs(router)
- sf1_floating_ip = fips_sfs[0]
+ if __name__ == '__main__':
+ return results.compile_summary(), self.creators
- fips = [client_floating_ip, server_floating_ip, sf1_floating_ip]
+ if __name__ == 'sfc.tests.functest.sfc_chain_deletion':
+ return results.compile_summary(), self.creators
- for ip in fips:
- logger.info("Checking connectivity towards floating IP [%s]" % ip)
- if not test_utils.ping(ip, retries=50, retry_timeout=3):
- logger.error("Cannot ping floating IP [%s]" % ip)
- os_sfc_utils.get_tacker_items()
- odl_utils.get_odl_items(odl_ip, odl_port)
- sys.exit(1)
- logger.info("Successful ping to floating IP [%s]" % ip)
- if not test_utils.check_ssh([sf1_floating_ip]):
- logger.error("Cannot establish SSH connection to the SFs")
- sys.exit(1)
-
- logger.info("Starting HTTP server on %s" % server_floating_ip)
- if not test_utils.start_http_server(server_floating_ip):
- logger.error('\033[91mFailed to start HTTP server on %s\033[0m'
- % server_floating_ip)
- sys.exit(1)
-
- logger.info("Wait for ODL to update the classification rules in OVS")
- t1.join()
-
- os_sfc_utils.delete_vnffg(tacker_client, vnffg_name='red_http')
-
- os_sfc_utils.delete_vnffgd(tacker_client, vnffgd_name='red')
-
- if not odl_utils.check_vnffg_deletion(odl_ip, odl_port, ovs_logger,
- [neutron_port],
- openstack_sfc.get_compute_client(),
- compute_nodes):
- logger.debug("The chains were not correctly removed")
- raise Exception("Chains not correctly removed, test failed")
-
- odl_utils.create_chain(tacker_client, default_param_file, neutron_port,
- COMMON_CONFIG, TESTCASE_CONFIG)
-
- # Start measuring the time it takes to implement the classification rules
- t2 = threading.Thread(target=odl_utils.wait_for_classification_rules,
- args=(ovs_logger, compute_nodes, odl_ip,
- odl_port, openstack_sfc.get_compute_client(),
- [neutron_port],))
- try:
- t2.start()
- except Exception as e:
- logger.error("Unable to start the thread that counts time %s" % e)
-
- logger.info("Starting SSH firewall on %s" % sf1_floating_ip)
- test_utils.start_vxlan_tool(sf1_floating_ip)
-
- logger.info("Wait for ODL to update the classification rules in OVS")
- t2.join()
-
- logger.info("Test HTTP")
- if not test_utils.is_http_blocked(client_floating_ip, server_ip):
- results.add_to_summary(2, "PASS", "HTTP works")
- else:
- error = ('\033[91mTEST 1 [FAILED] ==> HTTP BLOCKED\033[0m')
- logger.error(error)
- test_utils.capture_ovs_logs(
- ovs_logger, controller_clients, compute_clients, error)
- results.add_to_summary(2, "FAIL", "HTTP Blocked")
-
- logger.info("Stopping HTTP firewall on %s" % sf1_floating_ip)
- test_utils.stop_vxlan_tool(sf1_floating_ip)
- logger.info("Starting HTTP firewall on %s" % sf1_floating_ip)
- test_utils.start_vxlan_tool(sf1_floating_ip, block="80")
-
- logger.info("Test HTTP again")
- if test_utils.is_http_blocked(client_floating_ip, server_ip):
- results.add_to_summary(2, "PASS", "HTTP Blocked")
- else:
- error = ('\033[91mTEST 2 [FAILED] ==> HTTP works\033[0m')
- logger.error(error)
- test_utils.capture_ovs_logs(
- ovs_logger, controller_clients, compute_clients, error)
- results.add_to_summary(2, "FAIL", "HTTP works")
-
- return results.compile_summary(), openstack_sfc.creators
+if __name__ == '__main__':
+ TESTCASE_CONFIG = sfc_config.TestcaseConfig('sfc_chain_deletion')
+ supported_installers = ['fuel', 'apex', 'osa', 'compass']
+ vnf_names = ['testVNF1', 'testVNF2']
-if __name__ == '__main__':
- main()
+ test_run = SfcChainDeletion(TESTCASE_CONFIG, supported_installers,
+ vnf_names)
+ test_run.run()
diff --git a/sfc/tests/functest/sfc_one_chain_two_service_functions.py b/sfc/tests/functest/sfc_one_chain_two_service_functions.py
index e2f423ef..7b24d575 100644
--- a/sfc/tests/functest/sfc_one_chain_two_service_functions.py
+++ b/sfc/tests/functest/sfc_one_chain_two_service_functions.py
@@ -7,278 +7,83 @@
#
# http://www.apache.org/licenses/LICENSE-2.0
#
-import os
-import sys
import threading
-import sfc.lib.openstack_utils as os_sfc_utils
import sfc.lib.odl_utils as odl_utils
-import opnfv.utils.ovs_logger as ovs_log
import sfc.lib.config as sfc_config
-import sfc.lib.test_utils as test_utils
-import sfc.lib.topology_shuffler as topo_shuffler
-
from opnfv.utils import opnfv_logger as logger
-from sfc.lib.results import Results
-from opnfv.deployment.factory import Factory as DeploymentFactory
+from sfc.tests.functest import sfc_parent_function
""" logging configuration """
logger = logger.Logger(__name__).getLogger()
-CLIENT = "client"
-SERVER = "server"
-COMMON_CONFIG = sfc_config.CommonConfig()
-TESTCASE_CONFIG = sfc_config.TestcaseConfig(
- 'sfc_one_chain_two_service'
- '_functions')
-
-
-def main():
- deploymentHandler = DeploymentFactory.get_handler(
- COMMON_CONFIG.installer_type,
- COMMON_CONFIG.installer_ip,
- COMMON_CONFIG.installer_user,
- COMMON_CONFIG.installer_password,
- COMMON_CONFIG.installer_key_file)
- installer_type = os.environ.get("INSTALLER_TYPE")
-
- supported_installers = ['fuel', 'apex', 'osa', 'compass']
+class SfcOneChainTwoServiceTC(sfc_parent_function.SfcCommonTestCase):
+ """We create one client and one server using nova.
+ Then, 2 SFs are created using tacker.
+ A chain is created where both SFs are included.
+ The vxlan tool is used on both SFs. The purpose is to
+ check different HTTP traffic combinations.
+ """
+ def run(self):
- if installer_type not in supported_installers:
- logger.error(
- '\033[91mYour installer is not supported yet\033[0m')
- sys.exit(1)
+ logger.info("The test scenario %s is starting", __name__)
+ self.create_custom_vnfd(self.testcase_config.test_vnfd_red,
+ 'test-vnfd1')
+ self.create_custom_vnfd(self.testcase_config.test_vnfd_blue,
+ 'test-vnfd2')
- installer_ip = os.environ.get("INSTALLER_IP")
- if not installer_ip:
- logger.error(
- '\033[91minstaller ip is not set\033[0m')
- logger.error(
- '\033[91mexport INSTALLER_IP=<ip>\033[0m')
- sys.exit(1)
+ self.create_custom_av(self.vnfs[0], 'test-vnfd1', 'test-vim')
+ self.create_custom_av(self.vnfs[1], 'test-vnfd2', 'test-vim')
- cluster = COMMON_CONFIG.installer_cluster
+ self.create_vnffg(self.testcase_config.test_vnffgd_red, 'red',
+ 'red_http')
+ # Start measuring the time it takes to implement the
+ # classification rules
+ t1 = threading.Thread(target=odl_utils.wait_for_classification_rules,
+ args=(self.ovs_logger, self.compute_nodes,
+ self.odl_ip, self.odl_port,
+ self.client_instance.compute_host,
+ [self.neutron_port],))
+ try:
+ t1.start()
+ except Exception as e:
+ logger.error("Unable to start the thread that counts time %s" % e)
- openstack_nodes = (deploymentHandler.get_nodes({'cluster': cluster})
- if cluster is not None
- else deploymentHandler.get_nodes())
+ self.assign_floating_ip_client_server()
- controller_nodes = [node for node in openstack_nodes
- if node.is_controller()]
- compute_nodes = [node for node in openstack_nodes
- if node.is_compute()]
+ self.assign_floating_ip_sfs()
- odl_ip, odl_port = odl_utils.get_odl_ip_port(openstack_nodes)
+ self.check_floating_ips()
+ self.start_services_in_vm()
- for compute in compute_nodes:
- logger.info("This is a compute: %s" % compute.ip)
+ t1.join()
- results = Results(COMMON_CONFIG.line_length)
- results.add_to_summary(0, "=")
- results.add_to_summary(2, "STATUS", "SUBTEST")
- results.add_to_summary(0, "=")
+ logger.info("Allowed HTTP scenario")
+ results = self.present_results_allowed_http()
- openstack_sfc = os_sfc_utils.OpenStackSFC()
+ self.vxlan_blocking_start(self.fips_sfs[0], "80")
+ results = self.present_results_http()
- custom_flv = openstack_sfc.create_flavor(
- COMMON_CONFIG.flavor,
- COMMON_CONFIG.ram_size_in_mb,
- COMMON_CONFIG.disk_size_in_gb,
- COMMON_CONFIG.vcpu_count)
- if not custom_flv:
- logger.error("Failed to create custom flavor")
- sys.exit(1)
+ self.vxlan_blocking_start(self.fips_sfs[1], "80")
+ self.vxlan_blocking_stop(self.fips_sfs[0])
- logger.info("Fetching the ssh clients...")
- tacker_client = os_sfc_utils.get_tacker_client()
- controller_clients = test_utils.get_ssh_clients(controller_nodes)
- compute_clients = test_utils.get_ssh_clients(compute_nodes)
+ results = self.present_results_http()
- ovs_logger = ovs_log.OVSLogger(
- os.path.join(COMMON_CONFIG.sfc_test_dir, 'ovs-logs'),
- COMMON_CONFIG.functest_results_dir)
+ if __name__ == '__main__':
+ return results.compile_summary(), self.creators
- image_creator = openstack_sfc.register_glance_image(
- COMMON_CONFIG.image_name,
- COMMON_CONFIG.image_url,
- COMMON_CONFIG.image_format,
- 'public')
+ if __name__ == \
+ 'sfc.tests.functest.sfc_one_chain_two_service_functions':
+ return results.compile_summary(), self.creators
- network, router = openstack_sfc.create_network_infrastructure(
- TESTCASE_CONFIG.net_name,
- TESTCASE_CONFIG.subnet_name,
- TESTCASE_CONFIG.subnet_cidr,
- TESTCASE_CONFIG.router_name)
- sg = openstack_sfc.create_security_group(TESTCASE_CONFIG.secgroup_name)
+if __name__ == '__main__':
+ TESTCASE_CONFIG = sfc_config.TestcaseConfig('sfc_one_chain_two_service'
+ '_functions')
+ supported_installers = ['fuel', 'apex', 'osa', 'compass']
vnfs = ['testVNF1', 'testVNF2']
- topo_seed = topo_shuffler.get_seed()
- testTopology = topo_shuffler.topology(vnfs, openstack_sfc, seed=topo_seed)
-
- logger.info('This test is run with the topology {0}'
- .format(testTopology['id']))
- logger.info('Topology description: {0}'
- .format(testTopology['description']))
-
- client_instance, client_creator = openstack_sfc.create_instance(
- CLIENT, COMMON_CONFIG.flavor, image_creator, network, sg,
- av_zone=testTopology['client'])
-
- server_instance, server_creator = openstack_sfc.create_instance(
- SERVER, COMMON_CONFIG.flavor, image_creator, network, sg,
- av_zone=testTopology['server'])
-
- server_ip = server_instance.ports[0].ips[0]['ip_address']
- logger.info("Server instance received private ip [{}]".format(server_ip))
-
- os_sfc_utils.register_vim(tacker_client, vim_file=COMMON_CONFIG.vim_file)
-
- tosca_file = os.path.join(COMMON_CONFIG.sfc_test_dir,
- COMMON_CONFIG.vnfd_dir,
- TESTCASE_CONFIG.test_vnfd_red)
-
- os_sfc_utils.create_vnfd(
- tacker_client,
- tosca_file=tosca_file, vnfd_name='test-vnfd1')
-
- tosca_file = os.path.join(COMMON_CONFIG.sfc_test_dir,
- COMMON_CONFIG.vnfd_dir,
- TESTCASE_CONFIG.test_vnfd_blue)
- os_sfc_utils.create_vnfd(
- tacker_client,
- tosca_file=tosca_file, vnfd_name='test-vnfd2')
-
- default_param_file = os.path.join(
- COMMON_CONFIG.sfc_test_dir,
- COMMON_CONFIG.vnfd_dir,
- COMMON_CONFIG.vnfd_default_params_file)
-
- os_sfc_utils.create_vnf_in_av_zone(
- tacker_client, vnfs[0], 'test-vnfd1', 'test-vim',
- default_param_file, testTopology[vnfs[0]])
- os_sfc_utils.create_vnf_in_av_zone(
- tacker_client, vnfs[1], 'test-vnfd2', 'test-vim',
- default_param_file, testTopology[vnfs[1]])
-
- vnf1_id = os_sfc_utils.wait_for_vnf(tacker_client, vnf_name=vnfs[0])
- vnf2_id = os_sfc_utils.wait_for_vnf(tacker_client, vnf_name=vnfs[1])
- if vnf1_id is None or vnf2_id is None:
- logger.error('ERROR while booting vnfs')
- sys.exit(1)
-
- tosca_file = os.path.join(COMMON_CONFIG.sfc_test_dir,
- COMMON_CONFIG.vnffgd_dir,
- TESTCASE_CONFIG.test_vnffgd_red)
-
- os_sfc_utils.create_vnffgd(tacker_client,
- tosca_file=tosca_file,
- vnffgd_name='red')
-
- neutron_port = openstack_sfc.get_client_port(client_instance,
- client_creator)
- os_sfc_utils.create_vnffg_with_param_file(tacker_client, 'red',
- 'red_http',
- default_param_file,
- neutron_port.id)
-
- # Start measuring the time it takes to implement the classification rules
- t1 = threading.Thread(target=odl_utils.wait_for_classification_rules,
- args=(ovs_logger, compute_nodes, odl_ip,
- odl_port, openstack_sfc.get_compute_client(),
- [neutron_port],))
- try:
- t1.start()
- except Exception as e:
- logger.error("Unable to start the thread that counts time %s" % e)
-
- logger.info("Assigning floating IPs to instances")
- client_floating_ip = openstack_sfc.assign_floating_ip(router,
- client_instance,
- client_creator)
- server_floating_ip = openstack_sfc.assign_floating_ip(router,
- server_instance,
- server_creator)
- fips_sfs = openstack_sfc.assign_floating_ip_vnfs(router)
- sf1_floating_ip = fips_sfs[0]
- sf2_floating_ip = fips_sfs[1]
-
- fips = [client_floating_ip, server_floating_ip, sf1_floating_ip,
- sf2_floating_ip]
-
- for ip in fips:
- logger.info("Checking connectivity towards floating IP [%s]" % ip)
- if not test_utils.ping(ip, retries=50, retry_timeout=3):
- logger.error("Cannot ping floating IP [%s]" % ip)
- os_sfc_utils.get_tacker_items()
- odl_utils.get_odl_items(odl_ip, odl_port)
- sys.exit(1)
- logger.info("Successful ping to floating IP [%s]" % ip)
-
- if not test_utils.check_ssh([sf1_floating_ip, sf2_floating_ip]):
- logger.error("Cannot establish SSH connection to the SFs")
- sys.exit(1)
-
- logger.info("Starting HTTP server on %s" % server_floating_ip)
- if not test_utils.start_http_server(server_floating_ip):
- logger.error(
- 'Failed to start HTTP server on %s' % server_floating_ip)
- sys.exit(1)
-
- for sf_floating_ip in (sf1_floating_ip, sf2_floating_ip):
- logger.info("Starting vxlan_tool on %s" % sf_floating_ip)
- test_utils.start_vxlan_tool(sf_floating_ip)
-
- logger.info("Wait for ODL to update the classification rules in OVS")
- t1.join()
-
- logger.info("Test HTTP")
- if not test_utils.is_http_blocked(client_floating_ip, server_ip):
- results.add_to_summary(2, "PASS", "HTTP works")
- else:
- error = ('\033[91mTEST 1 [FAILED] ==> HTTP BLOCKED\033[0m')
- logger.error(error)
- test_utils.capture_ovs_logs(
- ovs_logger, controller_clients, compute_clients, error)
- results.add_to_summary(2, "FAIL", "HTTP blocked")
-
- logger.info("Changing the vxlan_tool to block HTTP traffic")
-
- # Make SF1 block http traffic
- test_utils.stop_vxlan_tool(sf1_floating_ip)
- logger.info("Starting HTTP firewall on %s" % sf1_floating_ip)
- test_utils.start_vxlan_tool(sf1_floating_ip, block="80")
-
- logger.info("Test HTTP again blocking SF1")
- if test_utils.is_http_blocked(client_floating_ip, server_ip):
- results.add_to_summary(2, "PASS", "HTTP Blocked")
- else:
- error = ('\033[91mTEST 2 [FAILED] ==> HTTP WORKS\033[0m')
- logger.error(error)
- test_utils.capture_ovs_logs(
- ovs_logger, controller_clients, compute_clients, error)
- results.add_to_summary(2, "FAIL", "HTTP not blocked")
-
- # Make SF2 block http traffic
- test_utils.stop_vxlan_tool(sf2_floating_ip)
- logger.info("Starting HTTP firewall on %s" % sf2_floating_ip)
- test_utils.start_vxlan_tool(sf2_floating_ip, block="80")
- logger.info("Stopping HTTP firewall on %s" % sf1_floating_ip)
- test_utils.stop_vxlan_tool(sf1_floating_ip)
-
- logger.info("Test HTTP again blocking SF2")
- if test_utils.is_http_blocked(client_floating_ip, server_ip):
- results.add_to_summary(2, "PASS", "HTTP Blocked")
- else:
- error = ('\033[91mTEST 3 [FAILED] ==> HTTP WORKS\033[0m')
- logger.error(error)
- test_utils.capture_ovs_logs(
- ovs_logger, controller_clients, compute_clients, error)
- results.add_to_summary(2, "FAIL", "HTTP not blocked")
-
- return results.compile_summary(), openstack_sfc.creators
-
-
-if __name__ == '__main__':
- main()
+ test_run = SfcOneChainTwoServiceTC(TESTCASE_CONFIG, supported_installers,
+ vnfs)
+ test_run.run()
diff --git a/sfc/tests/functest/sfc_parent_function.py b/sfc/tests/functest/sfc_parent_function.py
new file mode 100644
index 00000000..24ddd2e8
--- /dev/null
+++ b/sfc/tests/functest/sfc_parent_function.py
@@ -0,0 +1,519 @@
+import os
+import sys
+import logging
+import sfc.lib.test_utils as test_utils
+import sfc.lib.openstack_utils as os_sfc_utils
+import sfc.lib.topology_shuffler as topo_shuffler
+
+from opnfv.utils import ovs_logger as ovs_log
+from opnfv.deployment.factory import Factory as DeploymentFactory
+from sfc.lib import config as sfc_config
+from sfc.lib import odl_utils as odl_utils
+from sfc.lib.results import Results
+
+
+logger = logging.getLogger(__name__)
+CLIENT = "client"
+SERVER = "server"
+openstack_sfc = os_sfc_utils.OpenStackSFC()
+COMMON_CONFIG = sfc_config.CommonConfig()
+results = Results(COMMON_CONFIG.line_length)
+
+
+class SfcCommonTestCase(object):
+
+ def __init__(self, testcase_config, supported_installers, vnfs):
+
+ self.compute_nodes = None
+ self.controller_clients = None
+ self.compute_clients = None
+ self.tacker_client = None
+ self.ovs_logger = None
+ self.network = None
+ self.router = None
+ self.sg = None
+ self.image_creator = None
+ self.creators = None
+ self.odl_ip = None
+ self.odl_port = None
+ self.default_param_file = None
+ self.topo_seed = None
+ self.test_topology = None
+ self.server_instance = None
+ self.server_creator = None
+ self.client_instance = None
+ self.client_creator = None
+ self.server_ip = None
+ self.vnf_id = None
+ self.client_floating_ip = None
+ self.server_floating_ip = None
+ self.fips_sfs = None
+ self.neutron_port = None
+ self.testcase_config = testcase_config
+ self.vnfs = vnfs
+
+ self.prepare_env(testcase_config, supported_installers, vnfs)
+
+ def prepare_env(self, testcase_config, supported_installers, vnfs):
+ """Prepare the testcase environment and the components
+ that the test scenario is going to use later on.
+
+ :param testcase_config: the input test config file
+ :param supported_installers: the supported installers for this tc
+ :param vnfs: the names of vnfs
+ :return: Environment preparation
+ """
+
+ deployment_handler = DeploymentFactory.get_handler(
+ COMMON_CONFIG.installer_type,
+ COMMON_CONFIG.installer_ip,
+ COMMON_CONFIG.installer_user,
+ COMMON_CONFIG.installer_password,
+ COMMON_CONFIG.installer_key_file)
+
+ installer_type = os.environ.get("INSTALLER_TYPE")
+
+ if installer_type not in supported_installers:
+ logger.error(
+ '\033[91mYour installer is not supported yet\033[0m')
+ sys.exit(1)
+
+ installer_ip = os.environ.get("INSTALLER_IP")
+ if not installer_ip:
+ logger.error(
+ '\033[91minstaller ip is not set\033[0m')
+ logger.error(
+ '\033[91mexport INSTALLER_IP=<ip>\033[0m')
+ sys.exit(1)
+
+ cluster = COMMON_CONFIG.installer_cluster
+ openstack_nodes = (deployment_handler.get_nodes({'cluster': cluster})
+ if cluster is not None
+ else deployment_handler.get_nodes())
+
+ self.compute_nodes = [node for node in openstack_nodes
+ if node.is_compute()]
+
+ for compute in self.compute_nodes:
+ logger.info("This is a compute: %s" % compute.ip)
+
+ results.add_to_summary(0, "=")
+ results.add_to_summary(2, "STATUS", "SUBTEST")
+ results.add_to_summary(0, "=")
+
+ custom_flv = openstack_sfc.create_flavor(
+ COMMON_CONFIG.flavor,
+ COMMON_CONFIG.ram_size_in_mb,
+ COMMON_CONFIG.disk_size_in_gb,
+ COMMON_CONFIG.vcpu_count)
+ if not custom_flv:
+ logger.error("Failed to create custom flavor")
+ sys.exit(1)
+
+ controller_nodes = [node for node in openstack_nodes
+ if node.is_controller()]
+
+ self.controller_clients = test_utils.get_ssh_clients(controller_nodes)
+ self.compute_clients = test_utils.get_ssh_clients(self.compute_nodes)
+
+ self.tacker_client = os_sfc_utils.get_tacker_client()
+ os_sfc_utils.register_vim(self.tacker_client,
+ vim_file=COMMON_CONFIG.vim_file)
+
+ self.ovs_logger = ovs_log.OVSLogger(
+ os.path.join(COMMON_CONFIG.sfc_test_dir, 'ovs-logs'),
+ COMMON_CONFIG.functest_results_dir)
+
+ self.network, self.router = openstack_sfc.\
+ create_network_infrastructure(testcase_config.net_name,
+ testcase_config.subnet_name,
+ testcase_config.subnet_cidr,
+ testcase_config.router_name)
+
+ self.sg = openstack_sfc.create_security_group(
+ testcase_config.secgroup_name)
+
+ self.image_creator = openstack_sfc.register_glance_image(
+ COMMON_CONFIG.image_name,
+ COMMON_CONFIG.image_url,
+ COMMON_CONFIG.image_format,
+ 'public')
+
+ self.creators = openstack_sfc.creators
+
+ self.odl_ip, self.odl_port = odl_utils.get_odl_ip_port(openstack_nodes)
+
+ self.default_param_file = os.path.join(
+ COMMON_CONFIG.sfc_test_dir,
+ COMMON_CONFIG.vnfd_dir,
+ COMMON_CONFIG.vnfd_default_params_file)
+
+ self.topo_seed = topo_shuffler.get_seed()
+ self.test_topology = topo_shuffler.topology(vnfs, openstack_sfc,
+ seed=self.topo_seed)
+
+ logger.info('This test is run with the topology {0}'
+ .format(self.test_topology['id']))
+ logger.info('Topology description: {0}'
+ .format(self.test_topology['description']))
+
+ self.server_instance, self.server_creator = \
+ openstack_sfc.create_instance(SERVER, COMMON_CONFIG.flavor,
+ self.image_creator, self.network,
+ self.sg,
+ av_zone=self.test_topology['server'])
+
+ self.client_instance, self.client_creator = \
+ openstack_sfc.create_instance(CLIENT, COMMON_CONFIG.flavor,
+ self.image_creator, self.network,
+ self.sg,
+ av_zone=self.test_topology['client'])
+ logger.info('This test is run with the topology {0}'.format(
+ self.test_topology['id']))
+ logger.info('Topology description: {0}'.format(
+ self.test_topology['description']))
+
+ self.server_ip = self.server_instance.ports[0].ips[0]['ip_address']
+ logger.info("Server instance received private ip [{}]".format(
+ self.server_ip))
+
+ def create_custom_vnfd(self, test_case_name, vnfd_name):
+ """Create VNF Descriptor (VNFD)
+
+ :param test_case_name: the name of test case
+ :param vnfd_name: the name of vnfd
+ :return: vnfd
+ """
+
+ tosca_file = os.path.join(COMMON_CONFIG.sfc_test_dir,
+ COMMON_CONFIG.vnfd_dir, test_case_name)
+
+ os_sfc_utils.create_vnfd(self.tacker_client,
+ tosca_file=tosca_file,
+ vnfd_name=vnfd_name)
+
+ def create_custom_av(self, vnf_names, av_member1, av_member2):
+ """Create custom 'av'
+
+ :param vnf_names: names of available vnf(s)
+ :param av_member1: the first member of av zone
+ :param av_member2: the second member of av zone
+ :return: av zone
+ """
+
+ logger.info('This test is run with the topology {0}'.
+ format(self.test_topology['id']))
+ logger.info('Topology description: {0}'
+ .format(self.test_topology['description']))
+
+ os_sfc_utils.create_vnf_in_av_zone(
+ self.tacker_client, vnf_names, av_member1, av_member2,
+ self.default_param_file, self.test_topology[vnf_names])
+
+ self.vnf_id = os_sfc_utils.wait_for_vnf(self.tacker_client,
+ vnf_name=vnf_names)
+ if self.vnf_id is None:
+ logger.error('ERROR while booting vnfs')
+ sys.exit(1)
+
+ def assign_floating_ip_client_server(self):
+ """Assign floating IPs on the router about server and the client
+ instances
+
+ :return: Floating IPs for client and server
+ """
+
+ logger.info("Assigning floating IPs to client amd server instances")
+
+ self.client_floating_ip = openstack_sfc.assign_floating_ip(
+ self.router, self.client_instance, self.client_creator)
+ self.server_floating_ip = openstack_sfc.assign_floating_ip(
+ self.router, self.server_instance, self.server_creator)
+
+ def assign_floating_ip_sfs(self, vnf_ip=None):
+ """Assign floating IPs to service function
+
+ :param vnf_ip: IP of vnf - optional
+ :return: The list fips_sfs consist of the available IPs for service
+ functions
+ """
+
+ logger.info("Assigning floating IPs to service functions")
+
+ self.fips_sfs = openstack_sfc.assign_floating_ip_vnfs(self.router,
+ vnf_ip)
+
+ def check_floating_ips(self):
+ """Check the responsivness of the floating IPs
+
+ :return: The responsivness of IPs in the fips_sfs list is checked
+ """
+
+ fips = [self.client_floating_ip, self.server_floating_ip]
+
+ for sf in self.fips_sfs:
+ fips.append(sf)
+
+ for ip in fips:
+ logger.info("Checking connectivity towards floating IP [%s]" % ip)
+ if not test_utils.ping(ip, retries=50, retry_timeout=3):
+ logger.error("Cannot ping floating IP [%s]" % ip)
+ os_sfc_utils.get_tacker_items()
+ odl_utils.get_odl_items(self.odl_ip, self.odl_port)
+ sys.exit(1)
+ logger.info("Successful ping to floating IP [%s]" % ip)
+
+ if not test_utils.check_ssh(self.fips_sfs):
+ logger.error("Cannot establish SSH connection to the SFs")
+ sys.exit(1)
+
+ def start_services_in_vm(self):
+ """Start the HTTP server in the server VM as well as the vxlan tool for
+ the SFs IPs included in the fips_sfs list
+
+ :return: HTTP server and vxlan tools are started
+ """
+
+ logger.info("Starting HTTP server on %s" % self.server_floating_ip)
+ if not test_utils.start_http_server(self.server_floating_ip):
+ logger.error('\033[91mFailed to start HTTP server on %s\033[0m'
+ % self.server_floating_ip)
+ sys.exit(1)
+
+ for sf_floating_ip in self.fips_sfs:
+ logger.info("Starting vxlan_tool on %s" % sf_floating_ip)
+ test_utils.start_vxlan_tool(sf_floating_ip)
+
+ def present_results_ssh(self):
+ """Check whether the connection between server and client using
+ SSH protocol is blocked or not.
+
+ :return: The results for the specific action of the scenario
+ """
+
+ logger.info("Test SSH")
+ if test_utils.is_ssh_blocked(self.client_floating_ip, self.server_ip):
+ results.add_to_summary(2, "PASS", "SSH Blocked")
+ else:
+ error = ('\033[91mTEST [FAILED] ==> SSH NOT BLOCKED\033[0m')
+ logger.error(error)
+ test_utils.capture_ovs_logs(
+ self.ovs_logger, self.controller_clients, self.compute_clients,
+ error)
+ results.add_to_summary(2, "FAIL", "SSH Works")
+
+ return results
+
+ def present_results_allowed_ssh(self):
+ """Check whether the connection between server and client using
+ SSH protocol is available or not.
+
+ :return: The results for the specific action of the scenario
+ """
+
+ logger.info("Test SSH")
+ if not test_utils.is_ssh_blocked(self.client_floating_ip,
+ self.server_ip):
+ results.add_to_summary(2, "PASS", "SSH works")
+ else:
+ error = ('\033[91mTEST [FAILED] ==> SSH BLOCKED\033[0m')
+ logger.error(error)
+ test_utils.capture_ovs_logs(
+ self.ovs_logger, self.controller_clients, self.compute_clients,
+ error)
+ results.add_to_summary(2, "FAIL", "SSH is blocked")
+
+ return results
+
+ def remove_vnffg(self, par_vnffg_name, par_vnffgd_name):
+ """Delete the vnffg and the vnffgd items that have been created
+ during the test scenario.
+
+ :param par_vnffg_name: The vnffg name of network components
+ :param par_vnffgd_name: The vnffgd name of network components
+ :return: Remove the vnffg and vnffgd components
+ """
+
+ os_sfc_utils.delete_vnffg(self.tacker_client,
+ vnffg_name=par_vnffg_name)
+
+ os_sfc_utils.delete_vnffgd(self.tacker_client,
+ vnffgd_name=par_vnffgd_name)
+
+ def create_vnffg(self, testcase_config_name, vnf_name, conn_name):
+ """Create the vnffg components following the instructions from
+ relevant templates.
+
+ :param testcase_config_name: The config input of the test case
+ :param vnf_name: The name of the vnf
+ :param conn_name: Protocol type / name of the component
+ :return: Create the vnffg component
+ """
+
+ tosca_file = os.path.join(COMMON_CONFIG.sfc_test_dir,
+ COMMON_CONFIG.vnffgd_dir,
+ testcase_config_name)
+
+ os_sfc_utils.create_vnffgd(self.tacker_client,
+ tosca_file=tosca_file,
+ vnffgd_name=vnf_name)
+
+ self.neutron_port = openstack_sfc.get_client_port(self.client_instance,
+ self.client_creator)
+ os_sfc_utils.create_vnffg_with_param_file(self.tacker_client, vnf_name,
+ conn_name,
+ self.default_param_file,
+ self.neutron_port.id)
+
+ def present_results_http(self):
+ """Check whether the connection between server and client using
+ HTTP protocol is blocked or not.
+
+ :return: The results for the specific action of the scenario
+ """
+
+ logger.info("Test HTTP")
+ if test_utils.is_http_blocked(self.client_floating_ip, self.server_ip):
+ results.add_to_summary(2, "PASS", "HTTP Blocked")
+ else:
+ error = ('\033[91mTEST [FAILED] ==> HTTP WORKS\033[0m')
+ logger.error(error)
+ test_utils.capture_ovs_logs(
+ self.ovs_logger, self.controller_clients, self.compute_clients,
+ error)
+ results.add_to_summary(2, "FAIL", "HTTP works")
+
+ return results
+
+ def present_results_allowed_port_http(self, testcase_config):
+ """Check whether the connection between server and client using
+ HTTP protocol and for a specific port is available or not.
+
+ :param testcase_config: The config input of the test case
+ :return: The results for the specific action of the scenario
+ """
+
+ allowed_port = testcase_config.source_port
+ logger.info("Test if HTTP from port %s works" % allowed_port)
+ if not test_utils.is_http_blocked(
+ self.client_floating_ip, self.server_ip, allowed_port):
+ results.add_to_summary(2, "PASS", "HTTP works")
+ else:
+ error = ('\033[91mTEST [FAILED] ==> HTTP BLOCKED\033[0m')
+ logger.error(error)
+ test_utils.capture_ovs_logs(
+ self.ovs_logger, self.controller_clients, self.compute_clients,
+ error)
+ results.add_to_summary(2, "FAIL", "HTTP is blocked")
+
+ return results
+
+ def present_results_blocked_port_http(self, testcase_config):
+ """Check whether the connection between server and client using
+ HTTP protocol and for a specific port is blocked or not.
+
+ :param testcase_config: The config input of the test case
+ :return: The results for the specific action of the scenario
+ """
+
+ allowed_port = testcase_config.source_port
+ logger.info("Test if HTTP from port %s doesn't work" % allowed_port)
+ if test_utils.is_http_blocked(
+ self.client_floating_ip, self.server_ip, allowed_port):
+ results.add_to_summary(2, "PASS", "HTTP uplink blocked")
+ else:
+ error = ('\033[91mTEST [FAILED] ==> HTTP BLOCKED\033[0m')
+ logger.error(error)
+ test_utils.capture_ovs_logs(
+ self.ovs_logger, self.controller_clients, self.compute_clients,
+ error)
+ results.add_to_summary(2, "FAIL", "HTTP works")
+
+ return results
+
+ def create_chain(self, testcase_config):
+ """Create a connection chain for the test scenario purposes
+
+ :param testcase_config: The config input of the test case
+ :return: Create the proper chain for the specific test scenario
+ """
+
+ self.neutron_port = openstack_sfc.get_client_port(self.client_instance,
+ self.client_creator)
+ odl_utils.create_chain(self.tacker_client, self.default_param_file,
+ self.neutron_port, COMMON_CONFIG,
+ testcase_config)
+
+ def check_deletion(self):
+ """Check that the deletion of the chain has been completed sucessfully.
+
+ :return: Check that the chain has been completed deleted without
+ leftovers.
+ """
+
+ if not odl_utils.\
+ check_vnffg_deletion(self.odl_ip, self.odl_port,
+ self.ovs_logger,
+ [self.neutron_port],
+ self.client_instance.compute_host,
+ self.compute_nodes):
+ logger.debug("The chains were not correctly removed")
+ raise Exception("Chains not correctly removed, test failed")
+
+ def present_results_allowed_http(self):
+ """Check whether the connection between server and client using
+ HTTP protocol is available or not.
+
+ :return: The results for the specific action of the scenario
+ """
+
+ if not test_utils.is_http_blocked(self.client_floating_ip,
+ self.server_ip):
+ results.add_to_summary(2, "PASS", "HTTP works")
+ else:
+ error = ('\033[91mTEST [FAILED] ==> HTTP BLOCKED\033[0m')
+ logger.error(error)
+ test_utils.capture_ovs_logs(
+ self.ovs_logger, self.controller_clients, self.compute_clients,
+ error)
+ results.add_to_summary(2, "FAIL", "HTTP is blocked")
+
+ return results
+
+ def vxlan_blocking_start(self, floating_ip, port_blocked):
+ """Start the vxlan tool for one floating IP and blocking
+ a specific port.
+
+ :param floating_ip: Floating IP
+ :param port_blocked: Port
+ :return: The port for the floating IP is blocked
+ """
+
+ test_utils.stop_vxlan_tool(floating_ip)
+ logger.info("Starting HTTP firewall on %s" % floating_ip)
+ test_utils.start_vxlan_tool(floating_ip, block=port_blocked)
+
+ def vxlan_blocking_stop(self, floating_ip):
+ """Stop the vxlan tool for a specific IP
+
+ :param floating_ip: Floating IP
+ :return: The vxlan tool for the specific floating IP is stopped
+ """
+
+ logger.info("Starting HTTP firewall on %s" % floating_ip)
+ test_utils.stop_vxlan_tool(floating_ip)
+
+ def vxlan_start_interface(self, floating_ip, interface, output, block):
+ """Start the vxlan tool for one floating IP and blocking
+ a specific interface.
+
+ :param floating_ip: Floating IP
+ :param interface: Interface
+ :param output: output interface
+ :param block: port
+ :return: The interface or/and port for specific floating IP is blocked
+ """
+
+ logger.info("Starting vxlan_tool on %s" % floating_ip)
+ test_utils.start_vxlan_tool(floating_ip, interface=interface,
+ output=output, block=block)
diff --git a/sfc/tests/functest/sfc_symmetric_chain.py b/sfc/tests/functest/sfc_symmetric_chain.py
index 4ee45047..309d0dbc 100644
--- a/sfc/tests/functest/sfc_symmetric_chain.py
+++ b/sfc/tests/functest/sfc_symmetric_chain.py
@@ -8,295 +8,133 @@
#
# http://www.apache.org/licenses/LICENSE-2.0
#
-
import os
import sys
import threading
import sfc.lib.openstack_utils as os_sfc_utils
import sfc.lib.odl_utils as odl_utils
-import opnfv.utils.ovs_logger as ovs_log
import sfc.lib.config as sfc_config
-import sfc.lib.test_utils as test_utils
-import sfc.lib.topology_shuffler as topo_shuffler
-
+from sfc.tests.functest import sfc_parent_function
from opnfv.utils import opnfv_logger as logger
-from sfc.lib.results import Results
-from opnfv.deployment.factory import Factory as DeploymentFactory
""" logging configuration """
logger = logger.Logger(__name__).getLogger()
-
+COMMON_CONFIG = sfc_config.CommonConfig()
CLIENT = "client"
SERVER = "server"
-COMMON_CONFIG = sfc_config.CommonConfig()
-TESTCASE_CONFIG = sfc_config.TestcaseConfig('sfc_symmetric_chain')
-
-
-def main():
- deploymentHandler = DeploymentFactory.get_handler(
- COMMON_CONFIG.installer_type,
- COMMON_CONFIG.installer_ip,
- COMMON_CONFIG.installer_user,
- COMMON_CONFIG.installer_password,
- COMMON_CONFIG.installer_key_file)
-
- cluster = COMMON_CONFIG.installer_cluster
- all_nodes = (deploymentHandler.get_nodes({'cluster': cluster})
- if cluster is not None
- else deploymentHandler.get_nodes())
-
- controller_nodes = [node for node in all_nodes if node.is_controller()]
- compute_nodes = [node for node in all_nodes if node.is_compute()]
-
- odl_ip, odl_port = odl_utils.get_odl_ip_port(all_nodes)
-
- results = Results(COMMON_CONFIG.line_length)
- results.add_to_summary(0, "=")
- results.add_to_summary(2, "STATUS", "SUBTEST")
- results.add_to_summary(0, "=")
-
- openstack_sfc = os_sfc_utils.OpenStackSFC()
-
- custom_flavor = openstack_sfc.create_flavor(
- COMMON_CONFIG.flavor,
- COMMON_CONFIG.ram_size_in_mb,
- COMMON_CONFIG.disk_size_in_gb,
- COMMON_CONFIG.vcpu_count)
- if custom_flavor is None:
- logger.error("Failed to create custom flavor")
- sys.exit(1)
-
- logger.info("Fetching the ssh clients...")
- tacker_client = os_sfc_utils.get_tacker_client()
- controller_clients = test_utils.get_ssh_clients(controller_nodes)
- compute_clients = test_utils.get_ssh_clients(compute_nodes)
-
- ovs_logger = ovs_log.OVSLogger(
- os.path.join(COMMON_CONFIG.sfc_test_dir, 'ovs-logs'),
- COMMON_CONFIG.functest_results_dir)
-
- image_creator = openstack_sfc.register_glance_image(
- COMMON_CONFIG.image_name,
- COMMON_CONFIG.image_url,
- COMMON_CONFIG.image_format,
- 'public')
-
- network, router = openstack_sfc.create_network_infrastructure(
- TESTCASE_CONFIG.net_name,
- TESTCASE_CONFIG.subnet_name,
- TESTCASE_CONFIG.subnet_cidr,
- TESTCASE_CONFIG.router_name)
-
- sg = openstack_sfc.create_security_group(TESTCASE_CONFIG.secgroup_name)
-
- vnf_name = 'testVNF1'
- topo_seed = topo_shuffler.get_seed()
- testTopology = topo_shuffler.topology([vnf_name], openstack_sfc,
- seed=topo_seed)
- logger.info('This test is run with the topology {0}'
- .format(testTopology['id']))
- logger.info('Topology description: {0}'
- .format(testTopology['description']))
-
- client_instance, client_creator = openstack_sfc.create_instance(
- CLIENT, COMMON_CONFIG.flavor, image_creator, network, sg,
- av_zone=testTopology[CLIENT])
-
- server_instance, server_creator = openstack_sfc.create_instance(
- SERVER, COMMON_CONFIG.flavor, image_creator, network, sg,
- av_zone=testTopology[SERVER])
-
- server_ip = server_instance.ports[0].ips[0]['ip_address']
- logger.info("Server instance received private ip [{}]".format(server_ip))
-
- os_sfc_utils.register_vim(tacker_client, vim_file=COMMON_CONFIG.vim_file)
-
- tosca_file = os.path.join(
- COMMON_CONFIG.sfc_test_dir,
- COMMON_CONFIG.vnfd_dir,
- TESTCASE_CONFIG.test_vnfd)
-
- default_param_file = os.path.join(
- COMMON_CONFIG.sfc_test_dir,
- COMMON_CONFIG.vnfd_dir,
- COMMON_CONFIG.vnfd_default_params_file)
-
- os_sfc_utils.create_vnfd(
- tacker_client,
- tosca_file=tosca_file,
- vnfd_name='test-vnfd1')
- os_sfc_utils.create_vnf_in_av_zone(
- tacker_client,
- vnf_name,
- 'test-vnfd1',
- 'test-vim',
- default_param_file,
- testTopology[vnf_name])
-
- vnf_id = os_sfc_utils.wait_for_vnf(tacker_client, vnf_name=vnf_name)
- if vnf_id is None:
- logger.error('ERROR while booting VNF')
- sys.exit(1)
-
- tosca_file = os.path.join(
- COMMON_CONFIG.sfc_test_dir,
- COMMON_CONFIG.vnffgd_dir,
- TESTCASE_CONFIG.test_vnffgd)
- os_sfc_utils.create_vnffgd(
- tacker_client,
- tosca_file=tosca_file,
- vnffgd_name='test-vnffgd')
-
- client_port = openstack_sfc.get_client_port(
- client_instance,
- client_creator)
- server_port = openstack_sfc.get_client_port(
- server_instance,
- server_creator)
-
- server_ip_prefix = server_ip + '/32'
-
- os_sfc_utils.create_vnffg_with_param_file(
- tacker_client,
- 'test-vnffgd',
- 'test-vnffg',
- default_param_file,
- client_port.id,
- server_port.id,
- server_ip_prefix)
-
- # Start measuring the time it takes to implement the classification rules
- t1 = threading.Thread(
- target=wait_for_classification_rules,
- args=(ovs_logger, compute_nodes,
- openstack_sfc.get_compute_server(), server_port,
- openstack_sfc.get_compute_client(), client_port,
- odl_ip, odl_port,))
-
- try:
- t1.start()
- except Exception as e:
- logger.error("Unable to start the thread that counts time %s" % e)
-
- logger.info("Assigning floating IPs to instances")
- client_floating_ip = openstack_sfc.assign_floating_ip(router,
- client_instance,
- client_creator)
- server_floating_ip = openstack_sfc.assign_floating_ip(router,
- server_instance,
- server_creator)
-
- vnf_ip = os_sfc_utils.get_vnf_ip(tacker_client, vnf_id=vnf_id)
- fips_sfs = openstack_sfc.assign_floating_ip_vnfs(router, [vnf_ip])
- sf_floating_ip = fips_sfs[0]
-
- fips = [client_floating_ip, server_floating_ip, sf_floating_ip]
-
- for ip in fips:
- logger.info("Checking connectivity towards floating IP [%s]" % ip)
- if not test_utils.ping(ip, retries=50, retry_timeout=3):
- logger.error("Cannot ping floating IP [%s]" % ip)
- os_sfc_utils.get_tacker_items()
- odl_utils.get_odl_items(odl_ip, odl_port)
+openstack_sfc = os_sfc_utils.OpenStackSFC()
+
+
+class SfcSymmetricChain(sfc_parent_function.SfcCommonTestCase):
+ """One client and one server are created using nova.
+ The server will be running a web server on port 80.
+ Then one Service Function (SF) is created using Tacker.
+ This service function will be running a firewall that
+ blocks the traffic in a specific port.
+ A symmetric service chain routing the traffic throught
+ this SF will be created as well.
+ The purpose is to check different HTTP traffic
+ combinations using firewall.
+ """
+
+ def run(self):
+
+ logger.info("The test scenario %s is starting", __name__)
+ self.create_custom_vnfd(self.testcase_config.test_vnfd, 'test-vnfd1')
+ self.create_custom_av(self.vnfs[0], 'test-vnfd1', 'test-vim')
+
+ if self.vnf_id is None:
+ logger.error('ERROR while booting VNF')
sys.exit(1)
- logger.info("Successful ping to floating IP [%s]" % ip)
-
- if not test_utils.check_ssh([sf_floating_ip]):
- logger.error("Cannot establish SSH connection to the SFs")
- sys.exit(1)
-
- logger.info("Starting HTTP server on %s" % server_floating_ip)
- if not test_utils.start_http_server(server_floating_ip):
- logger.error('\033[91mFailed to start the HTTP server\033[0m')
- sys.exit(1)
-
- logger.info("Starting vxlan_tool on %s" % sf_floating_ip)
- test_utils.start_vxlan_tool(sf_floating_ip, interface='eth0',
- output='eth1')
- test_utils.start_vxlan_tool(sf_floating_ip, interface='eth1',
- output='eth0')
-
- logger.info("Wait for ODL to update the classification rules in OVS")
- t1.join()
-
- logger.info("Test HTTP")
- if not test_utils.is_http_blocked(client_floating_ip,
- server_ip,
- TESTCASE_CONFIG.source_port):
- results.add_to_summary(2, "PASS", "HTTP works")
- else:
- error = ('\033[91mTEST 1 [FAILED] ==> HTTP BLOCKED\033[0m')
- logger.error(error)
- test_utils.capture_ovs_logs(
- ovs_logger, controller_clients, compute_clients, error)
- results.add_to_summary(2, "FAIL", "HTTP blocked")
-
- logger.info("Changing the vxlan_tool to block HTTP request traffic")
-
- # Make SF1 block http request traffic
- test_utils.stop_vxlan_tool(sf_floating_ip)
- logger.info("Starting HTTP firewall on %s" % sf_floating_ip)
- test_utils.start_vxlan_tool(sf_floating_ip, interface='eth0',
- output='eth1', block="80")
- test_utils.start_vxlan_tool(sf_floating_ip, interface='eth1',
- output='eth0')
-
- logger.info("Test HTTP again blocking request on SF1")
- if test_utils.is_http_blocked(client_floating_ip,
- server_ip,
- TESTCASE_CONFIG.source_port):
- results.add_to_summary(2, "PASS", "HTTP uplink blocked")
- else:
- error = ('\033[91mTEST 2 [FAILED] ==> HTTP WORKS\033[0m')
- logger.error(error)
- test_utils.capture_ovs_logs(
- ovs_logger, controller_clients, compute_clients, error)
- results.add_to_summary(2, "FAIL", "HTTP works")
-
- logger.info("Changing the vxlan_tool to block HTTP response traffic")
-
- # Make SF1 block response http traffic
- test_utils.stop_vxlan_tool(sf_floating_ip)
- logger.info("Starting HTTP firewall on %s" % sf_floating_ip)
- test_utils.start_vxlan_tool(sf_floating_ip, interface='eth0',
- output='eth1')
- test_utils.start_vxlan_tool(sf_floating_ip, interface='eth1',
- output='eth0',
- block=TESTCASE_CONFIG.source_port)
-
- logger.info("Test HTTP again blocking response on SF1")
- if test_utils.is_http_blocked(client_floating_ip,
- server_ip,
- TESTCASE_CONFIG.source_port):
- results.add_to_summary(2, "PASS", "HTTP downlink blocked")
- else:
- error = ('\033[91mTEST 3 [FAILED] ==> HTTP WORKS\033[0m')
- logger.error(error)
- test_utils.capture_ovs_logs(
- ovs_logger, controller_clients, compute_clients, error)
- results.add_to_summary(2, "FAIL", "HTTP works")
-
- logger.info("Changing the vxlan_tool to allow HTTP traffic")
-
- # Make SF1 allow http traffic
- test_utils.stop_vxlan_tool(sf_floating_ip)
- logger.info("Starting HTTP firewall on %s" % sf_floating_ip)
- test_utils.start_vxlan_tool(sf_floating_ip, interface='eth0',
- output='eth1')
- test_utils.start_vxlan_tool(sf_floating_ip, interface='eth1',
- output='eth0')
-
- logger.info("Test HTTP")
- if not test_utils.is_http_blocked(client_floating_ip, server_ip):
- results.add_to_summary(2, "PASS", "HTTP restored")
- else:
- error = ('\033[91mTEST 4 [FAILED] ==> HTTP BLOCKED\033[0m')
- logger.error(error)
- test_utils.capture_ovs_logs(
- ovs_logger, controller_clients, compute_clients, error)
- results.add_to_summary(2, "FAIL", "HTTP blocked")
- return results.compile_summary(), openstack_sfc.creators
+ tosca_file = os.path.join(
+ COMMON_CONFIG.sfc_test_dir,
+ COMMON_CONFIG.vnffgd_dir,
+ self.testcase_config.test_vnffgd)
+ os_sfc_utils.create_vnffgd(
+ self.tacker_client,
+ tosca_file=tosca_file,
+ vnffgd_name='test-vnffgd')
+
+ client_port = openstack_sfc.get_client_port(
+ self.client_instance,
+ self.client_creator)
+ server_port = openstack_sfc.get_client_port(
+ self.server_instance,
+ self.server_creator)
+
+ server_ip_prefix = self.server_ip + '/32'
+
+ default_param_file = os.path.join(
+ COMMON_CONFIG.sfc_test_dir,
+ COMMON_CONFIG.vnfd_dir,
+ COMMON_CONFIG.vnfd_default_params_file)
+
+ os_sfc_utils.create_vnffg_with_param_file(
+ self.tacker_client,
+ 'test-vnffgd',
+ 'test-vnffg',
+ default_param_file,
+ client_port.id,
+ server_port.id,
+ server_ip_prefix)
+ # Start measuring the time it takes to implement the classification
+ # rules
+ t1 = threading.Thread(target=wait_for_classification_rules,
+ args=(self.ovs_logger, self.compute_nodes,
+ self.server_instance.compute_host,
+ server_port,
+ self.client_instance.compute_host,
+ client_port, self.odl_ip,
+ self.odl_port,))
+
+ try:
+ t1.start()
+ except Exception as e:
+ logger.error("Unable to start the thread that counts time %s" % e)
+
+ logger.info("Assigning floating IPs to instances")
+ self.assign_floating_ip_client_server()
+
+ vnf_ip = os_sfc_utils.get_vnf_ip(self.tacker_client,
+ vnf_id=self.vnf_id)
+ self.assign_floating_ip_sfs(vnf_ip)
+
+ self.check_floating_ips()
+
+ self.start_services_in_vm()
+
+ self.vxlan_start_interface(self.fips_sfs[0], 'eth0', 'eth1', None)
+ self.vxlan_start_interface(self.fips_sfs[0], 'eth1', 'eth0', None)
+
+ logger.info("Wait for ODL to update the classification rules in OVS")
+ t1.join()
+
+ results = self.present_results_allowed_port_http(self.testcase_config)
+
+ self.vxlan_blocking_stop(self.fips_sfs[0])
+ self.vxlan_start_interface(self.fips_sfs[0], 'eth0', 'eth1', "80")
+ self.vxlan_start_interface(self.fips_sfs[0], 'eth1', 'eth0', None)
+
+ results = self.present_results_blocked_port_http(self.testcase_config)
+
+ self.vxlan_blocking_stop(self.fips_sfs[0])
+ self.vxlan_start_interface(self.fips_sfs[0], 'eth0', 'eth1', None)
+ self.vxlan_start_interface(self.fips_sfs[0], 'eth1', 'eth0',
+ self.testcase_config.source_port)
+
+ results = self.present_results_blocked_port_http(self.testcase_config)
+
+ self.vxlan_blocking_stop(self.fips_sfs[0])
+ self.vxlan_start_interface(self.fips_sfs[0], 'eth0', 'eth1', None)
+ self.vxlan_start_interface(self.fips_sfs[0], 'eth1', 'eth0', None)
+ results = self.present_results_allowed_http()
+
+ if __name__ == '__main__':
+ return results.compile_summary(), self.creators
+
+ if __name__ == 'sfc.tests.functest.sfc_symmetric_chain':
+ return results.compile_summary(), self.creators
def wait_for_classification_rules(ovs_logger, compute_nodes,
@@ -329,4 +167,11 @@ def wait_for_classification_rules(ovs_logger, compute_nodes,
if __name__ == '__main__':
- main()
+
+ TESTCASE_CONFIG = sfc_config.TestcaseConfig('sfc_symmetric_chain')
+ supported_installers = ['fuel', 'apex', 'osa', 'compass']
+ vnf_names = ['testVNF1']
+
+ test_run = SfcSymmetricChain(TESTCASE_CONFIG, supported_installers,
+ vnf_names)
+ test_run.run()
diff --git a/sfc/tests/functest/sfc_two_chains_SSH_and_HTTP.py b/sfc/tests/functest/sfc_two_chains_SSH_and_HTTP.py
index a57fa745..a8b9d84a 100644
--- a/sfc/tests/functest/sfc_two_chains_SSH_and_HTTP.py
+++ b/sfc/tests/functest/sfc_two_chains_SSH_and_HTTP.py
@@ -8,305 +8,104 @@
# http://www.apache.org/licenses/LICENSE-2.0
#
-import os
-import sys
import threading
-import sfc.lib.openstack_utils as os_sfc_utils
import sfc.lib.odl_utils as odl_utils
-import opnfv.utils.ovs_logger as ovs_log
import sfc.lib.config as sfc_config
-import sfc.lib.test_utils as test_utils
-import sfc.lib.topology_shuffler as topo_shuffler
-
+from sfc.tests.functest import sfc_parent_function
from opnfv.utils import opnfv_logger as logger
-from sfc.lib.results import Results
-from opnfv.deployment.factory import Factory as DeploymentFactory
""" logging configuration """
logger = logger.Logger(__name__).getLogger()
-CLIENT = "client"
-SERVER = "server"
-COMMON_CONFIG = sfc_config.CommonConfig()
-TESTCASE_CONFIG = sfc_config.TestcaseConfig('sfc_two_chains_SSH_and_HTTP')
-
-
-def main():
- deploymentHandler = DeploymentFactory.get_handler(
- COMMON_CONFIG.installer_type,
- COMMON_CONFIG.installer_ip,
- COMMON_CONFIG.installer_user,
- COMMON_CONFIG.installer_password,
- COMMON_CONFIG.installer_key_file)
-
- installer_type = os.environ.get("INSTALLER_TYPE")
-
- supported_installers = ['fuel', 'apex', 'osa', 'compass']
-
- if installer_type not in supported_installers:
- logger.error(
- '\033[91mYour installer is not supported yet\033[0m')
- sys.exit(1)
-
- installer_ip = os.environ.get("INSTALLER_IP")
- if not installer_ip:
- logger.error(
- '\033[91minstaller ip is not set\033[0m')
- logger.error(
- '\033[91mexport INSTALLER_IP=<ip>\033[0m')
- sys.exit(1)
-
- cluster = COMMON_CONFIG.installer_cluster
- openstack_nodes = (deploymentHandler.get_nodes({'cluster': cluster})
- if cluster is not None
- else deploymentHandler.get_nodes())
-
- controller_nodes = [node for node in openstack_nodes
- if node.is_controller()]
- compute_nodes = [node for node in openstack_nodes
- if node.is_compute()]
-
- odl_ip, odl_port = odl_utils.get_odl_ip_port(openstack_nodes)
-
- for compute in compute_nodes:
- logger.info("This is a compute: %s" % compute.ip)
-
- results = Results(COMMON_CONFIG.line_length)
- results.add_to_summary(0, "=")
- results.add_to_summary(2, "STATUS", "SUBTEST")
- results.add_to_summary(0, "=")
-
- openstack_sfc = os_sfc_utils.OpenStackSFC()
-
- custom_flv = openstack_sfc.create_flavor(
- COMMON_CONFIG.flavor,
- COMMON_CONFIG.ram_size_in_mb,
- COMMON_CONFIG.disk_size_in_gb,
- COMMON_CONFIG.vcpu_count)
- if not custom_flv:
- logger.error("Failed to create custom flavor")
- sys.exit(1)
-
- logger.info("Fetching the ssh clients...")
- tacker_client = os_sfc_utils.get_tacker_client()
- controller_clients = test_utils.get_ssh_clients(controller_nodes)
- compute_clients = test_utils.get_ssh_clients(compute_nodes)
-
- ovs_logger = ovs_log.OVSLogger(
- os.path.join(COMMON_CONFIG.sfc_test_dir, 'ovs-logs'),
- COMMON_CONFIG.functest_results_dir)
-
- image_creator = openstack_sfc.register_glance_image(
- COMMON_CONFIG.image_name,
- COMMON_CONFIG.image_url,
- COMMON_CONFIG.image_format,
- 'public')
-
- network, router = openstack_sfc.create_network_infrastructure(
- TESTCASE_CONFIG.net_name,
- TESTCASE_CONFIG.subnet_name,
- TESTCASE_CONFIG.subnet_cidr,
- TESTCASE_CONFIG.router_name)
-
- sg = openstack_sfc.create_security_group(TESTCASE_CONFIG.secgroup_name)
- vnf_names = ['testVNF1', 'testVNF2']
-
- topo_seed = topo_shuffler.get_seed() # change to None for nova av zone
- testTopology = topo_shuffler.topology(vnf_names, openstack_sfc,
- seed=topo_seed)
-
- logger.info('This test is run with the topology {0}'
- .format(testTopology['id']))
- logger.info('Topology description: {0}'
- .format(testTopology['description']))
-
- client_instance, client_creator = openstack_sfc.create_instance(
- CLIENT, COMMON_CONFIG.flavor, image_creator, network, sg,
- av_zone=testTopology['client'])
-
- server_instance, server_creator = openstack_sfc.create_instance(
- SERVER, COMMON_CONFIG.flavor, image_creator, network, sg,
- av_zone=testTopology['server'])
-
- server_ip = server_instance.ports[0].ips[0]['ip_address']
-
- os_sfc_utils.register_vim(tacker_client, vim_file=COMMON_CONFIG.vim_file)
-
- tosca_red = os.path.join(COMMON_CONFIG.sfc_test_dir,
- COMMON_CONFIG.vnfd_dir,
- TESTCASE_CONFIG.test_vnfd_red)
- os_sfc_utils.create_vnfd(tacker_client,
- tosca_file=tosca_red,
- vnfd_name='test-vnfd1')
-
- tosca_blue = os.path.join(COMMON_CONFIG.sfc_test_dir,
- COMMON_CONFIG.vnfd_dir,
- TESTCASE_CONFIG.test_vnfd_blue)
-
- os_sfc_utils.create_vnfd(tacker_client,
- tosca_file=tosca_blue,
- vnfd_name='test-vnfd2')
+class SfcTwoChainsSSHandHTTP(sfc_parent_function.SfcCommonTestCase):
+ """We create one client and one server using nova.
+ Then, 2 SFs are created using tacker.
+ Two chains are created, having one SF each.
+ The vxlan tool is used on both SFs. The purpose is to
+ check different HTTP and SSH traffic combinations.
+ """
- default_param_file = os.path.join(
- COMMON_CONFIG.sfc_test_dir,
- COMMON_CONFIG.vnfd_dir,
- COMMON_CONFIG.vnfd_default_params_file)
+ def run(self):
- os_sfc_utils.create_vnf_in_av_zone(
- tacker_client, vnf_names[0], 'test-vnfd1', 'test-vim',
- default_param_file, testTopology[vnf_names[0]])
- os_sfc_utils.create_vnf_in_av_zone(
- tacker_client, vnf_names[1], 'test-vnfd2', 'test-vim',
- default_param_file, testTopology[vnf_names[1]])
+ logger.info("The test scenario %s is starting", __name__)
- vnf1_id = os_sfc_utils.wait_for_vnf(tacker_client, vnf_name=vnf_names[0])
- vnf2_id = os_sfc_utils.wait_for_vnf(tacker_client, vnf_name=vnf_names[1])
- if vnf1_id is None or vnf2_id is None:
- logger.error('ERROR while booting vnfs')
- sys.exit(1)
+ self.create_custom_vnfd(self.testcase_config.test_vnfd_red,
+ 'test-vnfd1')
+ self.create_custom_vnfd(self.testcase_config.test_vnfd_blue,
+ 'test-vnfd2')
- tosca_file = os.path.join(COMMON_CONFIG.sfc_test_dir,
- COMMON_CONFIG.vnffgd_dir,
- TESTCASE_CONFIG.test_vnffgd_red)
+ self.create_custom_av(self.vnfs[0], 'test-vnfd1', 'test-vim')
+ self.create_custom_av(self.vnfs[1], 'test-vnfd2', 'test-vim')
- os_sfc_utils.create_vnffgd(tacker_client,
- tosca_file=tosca_file,
- vnffgd_name='red')
+ self.create_vnffg(self.testcase_config.test_vnffgd_red, 'red',
+ 'red_http')
- neutron_port = openstack_sfc.get_client_port(client_instance,
- client_creator)
- os_sfc_utils.create_vnffg_with_param_file(tacker_client, 'red',
- 'red_http',
- default_param_file,
- neutron_port.id)
+ t1 = threading.Thread(target=odl_utils.wait_for_classification_rules,
+ args=(self.ovs_logger, self.compute_nodes,
+ self.odl_ip, self.odl_port,
+ self.client_instance.compute_host,
+ [self.neutron_port],))
- # Start measuring the time it takes to implement the classification rules
- t1 = threading.Thread(target=odl_utils.wait_for_classification_rules,
- args=(ovs_logger, compute_nodes, odl_ip,
- odl_port, openstack_sfc.get_compute_client(),
- [neutron_port],))
+ try:
+ t1.start()
+ except Exception as e:
+ logger.error("Unable to start the thread that counts time %s" % e)
- try:
- t1.start()
- except Exception as e:
- logger.error("Unable to start the thread that counts time %s" % e)
+ logger.info("Assigning floating IPs to instances")
+ self.assign_floating_ip_client_server()
- logger.info("Assigning floating IPs to instances")
- client_floating_ip = openstack_sfc.assign_floating_ip(router,
- client_instance,
- client_creator)
- server_floating_ip = openstack_sfc.assign_floating_ip(router,
- server_instance,
- server_creator)
- fips_sfs = openstack_sfc.assign_floating_ip_vnfs(router)
- sf1_floating_ip = fips_sfs[0]
- sf2_floating_ip = fips_sfs[1]
+ self.assign_floating_ip_sfs()
- fips = [client_floating_ip, server_floating_ip, sf1_floating_ip,
- sf2_floating_ip]
+ self.check_floating_ips()
+ self.start_services_in_vm()
+ self.vxlan_blocking_start(self.fips_sfs[0], "22")
+ self.vxlan_blocking_start(self.fips_sfs[1], "80")
- for ip in fips:
- logger.info("Checking connectivity towards floating IP [%s]" % ip)
- if not test_utils.ping(ip, retries=50, retry_timeout=3):
- logger.error("Cannot ping floating IP [%s]" % ip)
- os_sfc_utils.get_tacker_items()
- odl_utils.get_odl_items(odl_ip, odl_port)
- sys.exit(1)
- logger.info("Successful ping to floating IP [%s]" % ip)
+ logger.info("Wait for ODL to update the classification rules in OVS")
+ t1.join()
- if not test_utils.check_ssh([sf1_floating_ip, sf2_floating_ip]):
- logger.error("Cannot establish SSH connection to the SFs")
- sys.exit(1)
+ results = self.present_results_ssh()
+ results = self.present_results_allowed_http()
- logger.info("Starting HTTP server on %s" % server_floating_ip)
- if not test_utils.start_http_server(server_floating_ip):
- logger.error('\033[91mFailed to start HTTP server on %s\033[0m'
- % server_floating_ip)
- sys.exit(1)
+ logger.info("Changing the classification")
- logger.info("Starting SSH firewall on %s" % sf1_floating_ip)
- test_utils.start_vxlan_tool(sf1_floating_ip, block="22")
- logger.info("Starting HTTP firewall on %s" % sf2_floating_ip)
- test_utils.start_vxlan_tool(sf2_floating_ip, block="80")
+ self.remove_vnffg('red_http', 'red')
- logger.info("Wait for ODL to update the classification rules in OVS")
- t1.join()
+ self.create_vnffg(self.testcase_config.test_vnffgd_blue, 'blue',
+ 'blue_ssh')
- logger.info("Test SSH")
- if test_utils.is_ssh_blocked(client_floating_ip, server_ip):
- results.add_to_summary(2, "PASS", "SSH Blocked")
- else:
- error = ('\033[91mTEST 1 [FAILED] ==> SSH NOT BLOCKED\033[0m')
- logger.error(error)
- test_utils.capture_ovs_logs(
- ovs_logger, controller_clients, compute_clients, error)
- results.add_to_summary(2, "FAIL", "SSH Blocked")
+ # Start measuring the time it takes to implement the classification
+ # rules
+ t2 = threading.Thread(target=odl_utils.wait_for_classification_rules,
+ args=(self.ovs_logger, self.compute_nodes,
+ self.odl_ip, self.odl_port,
+ self.client_instance.compute_host,
+ self.neutron_port,))
+ try:
+ t2.start()
+ except Exception as e:
+ logger.error("Unable to start the thread that counts time %s" % e)
- logger.info("Test HTTP")
- if not test_utils.is_http_blocked(client_floating_ip, server_ip):
- results.add_to_summary(2, "PASS", "HTTP works")
- else:
- error = ('\033[91mTEST 2 [FAILED] ==> HTTP BLOCKED\033[0m')
- logger.error(error)
- test_utils.capture_ovs_logs(
- ovs_logger, controller_clients, compute_clients, error)
- results.add_to_summary(2, "FAIL", "HTTP works")
+ logger.info("Wait for ODL to update the classification rules in OVS")
+ t2.join()
- logger.info("Changing the classification")
+ results = self.present_results_http()
+ results = self.present_results_allowed_ssh()
- os_sfc_utils.delete_vnffg(tacker_client, vnffg_name='red_http_works')
+ if __name__ == '__main__':
+ return results.compile_summary(), self.creators
- os_sfc_utils.delete_vnffgd(tacker_client, vnffgd_name='red')
+ if __name__ == 'sfc.tests.functest.sfc_two_chains_SSH_and_HTTP':
+ return results.compile_summary(), self.creators
- tosca_file = os.path.join(COMMON_CONFIG.sfc_test_dir,
- COMMON_CONFIG.vnffgd_dir,
- TESTCASE_CONFIG.test_vnffgd_blue)
- os_sfc_utils.create_vnffgd(tacker_client,
- tosca_file=tosca_file,
- vnffgd_name='blue')
-
- os_sfc_utils.create_vnffg_with_param_file(tacker_client, 'blue',
- 'blue_ssh',
- default_param_file,
- neutron_port)
-
- # Start measuring the time it takes to implement the classification rules
- t2 = threading.Thread(target=odl_utils.wait_for_classification_rules,
- args=(ovs_logger, compute_nodes, odl_ip,
- odl_port, openstack_sfc.get_compute_client(),
- [neutron_port],))
- try:
- t2.start()
- except Exception as e:
- logger.error("Unable to start the thread that counts time %s" % e)
-
- logger.info("Wait for ODL to update the classification rules in OVS")
- t2.join()
-
- logger.info("Test HTTP")
- if test_utils.is_http_blocked(client_floating_ip, server_ip):
- results.add_to_summary(2, "PASS", "HTTP Blocked")
- else:
- error = ('\033[91mTEST 3 [FAILED] ==> HTTP WORKS\033[0m')
- logger.error(error)
- test_utils.capture_ovs_logs(
- ovs_logger, controller_clients, compute_clients, error)
- results.add_to_summary(2, "FAIL", "HTTP Blocked")
-
- logger.info("Test SSH")
- if not test_utils.is_ssh_blocked(client_floating_ip, server_ip):
- results.add_to_summary(2, "PASS", "SSH works")
- else:
- error = ('\033[91mTEST 4 [FAILED] ==> SSH BLOCKED\033[0m')
- logger.error(error)
- test_utils.capture_ovs_logs(
- ovs_logger, controller_clients, compute_clients, error)
- results.add_to_summary(2, "FAIL", "SSH works")
-
- return results.compile_summary(), openstack_sfc.creators
+if __name__ == '__main__':
+ TESTCASE_CONFIG = sfc_config.TestcaseConfig('sfc_two_chains_SSH_and_HTTP')
+ supported_installers = ['fuel', 'apex', 'osa', 'compass']
+ vnf_names = ['testVNF1', 'testVNF2']
-if __name__ == '__main__':
- main()
+ test_run = SfcTwoChainsSSHandHTTP(TESTCASE_CONFIG, supported_installers,
+ vnf_names)
+ test_run.run()