tempest:
enabled: true
description: Neutron BGPVPN tests in tempest
+ succes_criteria: 100 # all the subtests must pass
+ testname_db: functest_tempest # declared name in the test api
+ # http://testresults.opnfv.org/test/api/v1/projects/sdnvpn/cases
testcase_1:
enabled: true
description: VPN provides connectivity between subnets
+ succes_criteria: 75 # we let fail 25% of the subtests
+ testname_db: functest_testcase_1
ping_timeout: 200
flavor: m1.tiny # adapt to your environment
instance_1_name: sdnvpn-1-1
router_2_name: sdnvpn-1-2-router
sdnvpn_sg_name: sdnvpn-sg
sdnvpn_sg_descr: Security group for SDNVPN test cases
+ targets1: '88:88'
+ targets2: '55:55'
testcase_2:
enabled: true
description: Tenant separation
+ succes_criteria: 100 # all the subtests must pass
+ testname_db: functest_testcase_2
ping_timeout: 200
flavor: m1.tiny # adapt to your environment
instance_1_name: sdnvpn-2-1
router_2_name: sdnvpn-2-2-router
sdnvpn_sg_name: sdnvpn-sg
sdnvpn_sg_descr: Security group for SDNVPN test cases
+ targets1: '88:88'
+ targets2: '55:55'
testcase_3:
enabled: false
description: Data center gateway integration
+ succes_criteria: 100 # all the subtests must pass
+ testname_db: functest_testcase_3
#
# http://www.apache.org/licenses/LICENSE-2.0
#
-
+import argparse
import importlib
+import os
+import time
import functest.utils.functest_logger as ft_logger
import functest.utils.functest_utils as ft_utils
import yaml
+""" tests configuration """
+parser = argparse.ArgumentParser()
+parser.add_argument("-r", "--report",
+ help="Create json result file",
+ action="store_true")
+args = parser.parse_args()
+
+TEST_DB_URL = ft_utils.get_parameter_from_yaml('results.test_db_url')
+logger = ft_logger.Logger("sdnvpn-run-tests").getLogger()
+REPO_PATH = os.environ['repos_dir'] + '/sdnvpn/'
+
+
+def push_results(testname, start_time, end_time, criteria, details):
+ logger.info("Push testcase '%s' results into the DB...\n" % testname)
+ ft_utils.push_results_to_db("sdnvpn",
+ testname,
+ logger,
+ start_time,
+ end_time,
+ criteria,
+ details)
+
+
+def main():
+
+ with open(REPO_PATH + 'test/functest/config.yaml') as f:
+ config_yaml = yaml.safe_load(f)
+
+ testcases = config_yaml.get("testcases")
+ for testcase in testcases:
+ if testcases[testcase]['enabled']:
+ test_name = testcase
+ test_descr = testcases[testcase]['description']
+ test_name_db = testcases[testcase]['testname_db']
+ title = ("Running '%s - %s'" %
+ (test_name, test_descr))
+ logger.info(title)
+ logger.info("%s\n" % ("=" * len(title)))
+ t = importlib.import_module(testcase, package=None)
+ start_time = time.time()
+ result = t.main()
+ if result < 0:
+ result = {"status": "FAILED", "details": "execution error."}
+ logger.info("Results of test case '%s - %s':\n%s\n" %
+ (test_name, test_descr, result))
+ end_time = time.time()
+ # duration = end_time - start_time
+ criteria = result.get("status")
+ details = result.get("details")
+ if args.report:
+ push_results(
+ test_name_db, start_time, end_time, criteria, details)
-logger = ft_logger.Logger("sdnvpn").getLogger()
+ exit(0)
-with open('config.yaml') as f:
- config_yaml = yaml.safe_load(f)
-testcases = config_yaml.get("testcases")
-for testcase in testcases:
- title = ("Running '%s - %s'" %
- (testcase, testcases[testcase]['description']))
- print(title)
- print("%s\n" % ("=" * len(title)))
- if testcases[testcase]['type'] == 'python':
- t = importlib.import_module(testcase, package=None)
- t.main()
- else:
- cmd = "bash " + testcase + ".sh"
- result = ft_utils.execute_command(cmd, logger, exit_on_error=False)
+if __name__ == '__main__':
+ main()
import functest.utils.functest_logger as ft_logger
import functest.utils.functest_utils as ft_utils
-logger = ft_logger.Logger("bgpvpn").getLogger()
+
+logger = ft_logger.Logger("sdnvpn-tempest").getLogger()
+
+REPO_PATH = os.environ['repos_dir'] + '/sdnvpn/'
+config_file = REPO_PATH + 'test/functest/config.yaml'
+
+SUCCESS_CRITERIA = ft_utils.get_parameter_from_yaml(
+ "testcases.testcase_1.succes_criteria", config_file)
def main():
# Look for name of the tests
testcases = re.findall("\{0\} (.*)", output)
- results = {"test_name": "tempest", "duration": duration,
+ results = {"duration": duration,
"num_tests": num_tests, "failed": failed,
"tests": testcases}
- logger.info("Results: %s" % results)
- return results
+ status = "PASS"
+ if 100 - (100 * int(failed) / int(num_tests)) < int(SUCCESS_CRITERIA):
+ status = "FAILED"
+ return {"status": status, "details": results}
except:
logger.error("Problem when parsing the results.")
args = parser.parse_args()
-""" logging configuration """
logger = ft_logger.Logger("sdnvpn-testcase-1").getLogger()
REPO_PATH = os.environ['repos_dir'] + '/sdnvpn/'
-HOME = os.environ['HOME'] + "/"
VM_BOOT_TIMEOUT = 180
"testcases.testcase_1.sdnvpn_sg_name", config_file)
SECGROUP_DESCR = ft_utils.get_parameter_from_yaml(
"testcases.testcase_1.sdnvpn_sg_descr", config_file)
-
+TARGETS_1 = ft_utils.get_parameter_from_yaml(
+ "testcases.testcase_1.targets1", config_file)
+TARGETS_2 = ft_utils.get_parameter_from_yaml(
+ "testcases.testcase_1.targets2", config_file)
+SUCCESS_CRITERIA = ft_utils.get_parameter_from_yaml(
+ "testcases.testcase_1.succes_criteria", config_file)
TEST_DB = ft_utils.get_parameter_from_yaml("results.test_db_url")
TEST_RESULT = "PASS"
SUMMARY = ""
LINE_LENGTH = 60 # length for the summary table
+DETAILS = []
+NUM_TESTS = 0
+NUM_TESTS_FAILED = 0
def create_network(neutron_client, net, subnet, router, cidr):
expected_result = 'can ping' if expected == 'PASS' else 'cannot ping'
test_case_name = ("'%s' %s '%s'" %
(vm_source.name, expected_result, vm_target.name))
- logger.debug("\n%sPing\n%sfrom '%s' (%s)\n%sto '%s' (%s).\n"
+ logger.debug("%sPing\n%sfrom '%s' (%s)\n%sto '%s' (%s).\n"
"%s-->Expected result: %s.\n"
% (tab, tab, vm_source.name, ip_source,
tab, vm_target.name, ip_target,
def add_to_summary(num_cols, col1, col2=""):
- global SUMMARY, LINE_LENGTH
+ global SUMMARY, LINE_LENGTH, DETAILS, NUM_TESTS, NUM_TESTS_FAILED
if num_cols == 0:
SUMMARY += ("+%s+\n" % (col1 * (LINE_LENGTH - 2)))
elif num_cols == 1:
elif num_cols == 2:
SUMMARY += ("| %s" % col1.ljust(7) + "| ")
SUMMARY += (col2.ljust(LINE_LENGTH - 12) + "|\n")
+ DETAILS.append({col2: col1})
+ if col1 in ("FAIL", "PASS"):
+ NUM_TESTS += 1
+ if col1 == "FAIL":
+ NUM_TESTS_FAILED += 1
def main():
logger.debug("Instance '%s' booted successfully. IP='%s'." %
(INSTANCE_1_NAME, vm_1_ip))
msg = ("Create VPN with eRT<>iRT")
- logger.info("\n\n--> %s ..." % msg)
+ logger.info(msg)
add_to_summary(1, msg)
vpn_name = "sdnvpn-" + str(randint(100000, 999999))
- kwargs = {"import_targets": "88:88",
- "export_targets": "55:55",
+ kwargs = {"import_targets": TARGETS_1,
+ "export_targets": TARGETS_2,
"name": vpn_name}
bgpvpn = os_utils.create_bgpvpn(neutron_client, **kwargs)
bgpvpn_id = bgpvpn['bgpvpn']['id']
logger.debug("VPN created details: %s" % bgpvpn)
msg = ("Associate network '%s' to the VPN." % NET_1_NAME)
- logger.info("\n\n--> %s..." % msg)
+ logger.info(msg)
add_to_summary(1, msg)
add_to_summary(0, "-")
get_ping_status(vm_1, vm_1_ip, vm_4, vm_4_ip, expected="FAIL", timeout=30)
msg = ("Associate network '%s' to the VPN." % NET_2_NAME)
- logger.info("\n\n--> %s..." % msg)
+ logger.info(msg)
add_to_summary(0, "-")
add_to_summary(1, msg)
add_to_summary(0, "-")
neutron_client, bgpvpn_id, network_2_id)
# Wait a bit for this to take effect
- time.sleep(10)
+ time.sleep(30)
# Ping from VM4 to VM5 should work
get_ping_status(vm_4, vm_4_ip, vm_5, vm_5_ip, expected="PASS", timeout=30)
get_ping_status(vm_1, vm_1_ip, vm_5, vm_5_ip, expected="FAIL", timeout=30)
msg = ("Update VPN with eRT=iRT ...")
- logger.info("\n\n--> %s..." % msg)
+ logger.info(msg)
add_to_summary(0, "-")
add_to_summary(1, msg)
add_to_summary(0, "-")
- kwargs = {"import_targets": "88:88",
- "export_targets": "88:88",
+ kwargs = {"import_targets": TARGETS_1,
+ "export_targets": TARGETS_1,
"name": vpn_name}
bgpvpn = os_utils.update_bgpvpn(neutron_client, bgpvpn_id, **kwargs)
# Wait a bit for this to take effect
- time.sleep(10)
+ time.sleep(30)
# Ping from VM1 to VM4 should work
get_ping_status(vm_1, vm_1_ip, vm_4, vm_4_ip, expected="PASS", timeout=30)
else:
logger.info("One or more ping tests have failed.")
- sys.exit(0)
+ status = "PASS"
+ success = 100 - (100 * int(NUM_TESTS_FAILED) / int(NUM_TESTS))
+ if success < int(SUCCESS_CRITERIA):
+ status = "FAILED"
+
+ return {"status": status, "details": DETAILS}
if __name__ == '__main__':
args = parser.parse_args()
-""" logging configuration """
-logger = ft_logger.Logger("sdnvpn-testcase-1").getLogger()
+logger = ft_logger.Logger("sdnvpn-testcase-2").getLogger()
REPO_PATH = os.environ['repos_dir'] + '/sdnvpn/'
-HOME = os.environ['HOME'] + "/"
VM_BOOT_TIMEOUT = 180
"testcases.testcase_2.sdnvpn_sg_name", config_file)
SECGROUP_DESCR = ft_utils.get_parameter_from_yaml(
"testcases.testcase_2.sdnvpn_sg_descr", config_file)
-
+TARGETS_1 = ft_utils.get_parameter_from_yaml(
+ "testcases.testcase_2.targets1", config_file)
+TARGETS_2 = ft_utils.get_parameter_from_yaml(
+ "testcases.testcase_2.targets2", config_file)
+SUCCESS_CRITERIA = ft_utils.get_parameter_from_yaml(
+ "testcases.testcase_1.succes_criteria", config_file)
TEST_DB = ft_utils.get_parameter_from_yaml("results.test_db_url")
TEST_RESULT = "PASS"
SUMMARY = ""
LINE_LENGTH = 90 # length for the summary table
+DETAILS = []
+NUM_TESTS = 0
+NUM_TESTS_FAILED = 0
def create_network(neutron_client, net, subnet1, cidr1,
test_case_name = ("[%s] returns 'I am %s' to '%s'[%s]" %
(ip_target, expected,
vm_source.name, ip_source))
- logger.debug("\n%sSSH\n%sfrom '%s' (%s)\n%sto '%s' (%s).\n"
+ logger.debug("%sSSH\n%sfrom '%s' (%s)\n%sto '%s' (%s).\n"
"%s-->Expected result: %s.\n"
% (tab, tab, vm_source.name, ip_source,
tab, vm_target.name, ip_target,
def add_to_summary(num_cols, col1, col2=""):
- global SUMMARY, LINE_LENGTH
+ global SUMMARY, LINE_LENGTH, NUM_TESTS, NUM_TESTS_FAILED
if num_cols == 0:
SUMMARY += ("+%s+\n" % (col1 * (LINE_LENGTH - 2)))
elif num_cols == 1:
elif num_cols == 2:
SUMMARY += ("| %s" % col1.ljust(7) + "| ")
SUMMARY += (col2.ljust(LINE_LENGTH - 12) + "|\n")
+ DETAILS.append({col2: col1})
+ if col1 in ("FAIL", "PASS"):
+ NUM_TESTS += 1
+ if col1 == "FAIL":
+ NUM_TESTS_FAILED += 1
def main():
(INSTANCE_1_NAME, vm_1_ip))
msg = ("Create VPN1 with eRT=iRT")
- logger.info("\n\n--> %s ..." % msg)
+ logger.info(msg)
add_to_summary(1, msg)
vpn1_name = "sdnvpn-1-" + str(randint(100000, 999999))
- kwargs = {"import_targets": "55:55",
- "export_targets": "55:55",
- "route_targets": "55:55",
+ kwargs = {"import_targets": TARGETS_2,
+ "export_targets": TARGETS_2,
+ "route_targets": TARGETS_2,
"name": vpn1_name}
bgpvpn1 = os_utils.create_bgpvpn(neutron_client, **kwargs)
bgpvpn1_id = bgpvpn1['bgpvpn']['id']
logger.debug("VPN1 created details: %s" % bgpvpn1)
msg = ("Associate network '%s' to the VPN." % NET_1_NAME)
- logger.info("\n\n--> %s..." % msg)
+ logger.info(msg)
add_to_summary(1, msg)
add_to_summary(0, "-")
add_to_summary(0, "-")
msg = ("Create VPN2 with eRT=iRT")
- logger.info("\n\n--> %s ..." % msg)
+ logger.info(msg)
add_to_summary(1, msg)
vpn2_name = "sdnvpn-2-" + str(randint(100000, 999999))
- kwargs = {"import_targets": "88:88",
- "export_targets": "88:88",
- "route_targets": "88:88",
+ kwargs = {"import_targets": TARGETS_1,
+ "export_targets": TARGETS_1,
+ "route_targets": TARGETS_1,
"name": vpn2_name}
bgpvpn2 = os_utils.create_bgpvpn(neutron_client, **kwargs)
bgpvpn2_id = bgpvpn2['bgpvpn']['id']
logger.debug("VPN created details: %s" % bgpvpn2)
msg = ("Associate network '%s' to the VPN2." % NET_2_NAME)
- logger.info("\n\n--> %s..." % msg)
+ logger.info(msg)
add_to_summary(1, msg)
add_to_summary(0, "-")
else:
logger.info("One or more sub tests have failed.")
- sys.exit(0)
+ status = "PASS"
+ success = 100 - (100 * int(NUM_TESTS_FAILED) / int(NUM_TESTS))
+ if success < int(SUCCESS_CRITERIA):
+ status = "FAILED"
+
+ return {"status": status, "details": DETAILS}
if __name__ == '__main__':
Code Review / sdnvpn.git / commitdiff