#!/usr/bin/env python3
-# Copyright 2015-2016 Intel Corporation.
+# Copyright 2015-2017 Intel Corporation.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
import os
import sys
import argparse
+import re
import time
import datetime
import shutil
import unittest
-import xmlrunner
import locale
-
-sys.dont_write_bytecode = True
-
+import copy
+import glob
+import subprocess
+import ast
+import xmlrunner
from conf import settings
-from conf import get_test_param
+import core.component_factory as component_factory
from core.loader import Loader
-from testcases import TestCase
+from testcases import PerformanceTestCase
+from testcases import IntegrationTestCase
from tools import tasks
+from tools import networkcard
+from tools import functions
from tools.pkt_gen import trafficgen
from tools.opnfvdashboard import opnfvdashboard
+sys.dont_write_bytecode = True
+
VERBOSITY_LEVELS = {
'debug': logging.DEBUG,
'info': logging.INFO,
'critical': logging.CRITICAL
}
+_CURR_DIR = os.path.dirname(os.path.realpath(__file__))
+
+_TEMPLATE_RST = {'head' : os.path.join(_CURR_DIR, 'tools/report/report_head.rst'),
+ 'foot' : os.path.join(_CURR_DIR, 'tools/report/report_foot.rst'),
+ 'final' : 'test_report.rst',
+ 'tmp' : os.path.join(_CURR_DIR, 'tools/report/report_tmp_caption.rst')
+ }
+
+
+_LOGGER = logging.getLogger()
def parse_arguments():
"""
This expects either 'x=y', 'x=y,z' or 'x' (implicit true)
values. For multiple overrides use a ; separated list for
- e.g. --test-params 'x=z; y=a,b'
+ e.g. --test-params 'x=z; y=(a,b)'
"""
def __call__(self, parser, namespace, values, option_string=None):
results = {}
- for value in values.split(';'):
- result = [key.strip() for key in value.split('=')]
- if len(result) == 1:
- results[result[0]] = True
- elif len(result) == 2:
- results[result[0]] = result[1]
- else:
- raise argparse.ArgumentTypeError(
- 'expected \'%s\' to be of format \'key=val\' or'
- ' \'key\'' % result)
+ for param, _, value in re.findall('([^;=]+)(=([^;]+))?', values):
+ param = param.strip()
+ value = value.strip()
+ if len(param):
+ if len(value):
+ # values are passed inside string from CLI, so we must retype them accordingly
+ try:
+ results[param] = ast.literal_eval(value)
+ except ValueError:
+ # for backward compatibility, we have to accept strings without quotes
+ _LOGGER.warning("Adding missing quotes around string value: %s = %s",
+ param, str(value))
+ results[param] = str(value)
+ else:
+ results[param] = True
setattr(namespace, self.dest, results)
help='list all system metrics loggers and exit')
parser.add_argument('--list-vswitches', action='store_true',
help='list all system vswitches and exit')
+ parser.add_argument('--list-fwdapps', action='store_true',
+ help='list all system forwarding applications and exit')
parser.add_argument('--list-vnfs', action='store_true',
help='list all system vnfs and exit')
parser.add_argument('--list-settings', action='store_true',
To run all tests omit both positional args and --tests arg.')
group = parser.add_argument_group('test selection options')
+ group.add_argument('-m', '--mode', help='vsperf mode of operation;\
+ Values: "normal" - execute vSwitch, VNF and traffic generator;\
+ "trafficgen" - execute only traffic generator; "trafficgen-off" \
+ - execute vSwitch and VNF; trafficgen-pause - execute vSwitch \
+ and VNF but pause before traffic transmission ', default='normal')
+
group.add_argument('-f', '--test-spec', help='test specification file')
group.add_argument('-d', '--test-dir', help='directory containing tests')
group.add_argument('-t', '--tests', help='Comma-separated list of terms \
indicating tests to run. e.g. "RFC2544,!p2p" - run all tests whose\
- name contains RFC2544 less those containing "p2p"')
+ name contains RFC2544 less those containing "p2p"; "!back2back" - \
+ run all tests except those containing back2back')
group.add_argument('--verbosity', choices=list_logging_levels(),
help='debug level')
+ group.add_argument('--integration', action='store_true', help='execute integration tests')
group.add_argument('--trafficgen', help='traffic generator to use')
group.add_argument('--vswitch', help='vswitch implementation to use')
+ group.add_argument('--fwdapp', help='packet forwarding application to use')
group.add_argument('--vnf', help='vnf to use')
- group.add_argument('--duration', help='traffic transmit duration')
group.add_argument('--sysmetrics', help='system metrics logger to use')
group = parser.add_argument_group('test behavior options')
group.add_argument('--xunit', action='store_true',
group.add_argument('--conf-file', action=_ValidateFileAction,
help='settings file')
group.add_argument('--test-params', action=_SplitTestParamsAction,
- help='csv list of test parameters: key=val; e.g.'
- 'including pkt_sizes=x,y; duration=x; '
- 'rfc2544_trials=x ...')
+ help='csv list of test parameters: key=val; e.g. '
+ 'TRAFFICGEN_PKT_SIZES=(64,128);TRAFICGEN_DURATION=30; '
+ 'GUEST_LOOPBACK=["l2fwd"] ...')
group.add_argument('--opnfvpod', help='name of POD in opnfv')
args = vars(parser.parse_args())
settings.getValue('LOG_DIR'),
settings.getValue('LOG_FILE_TRAFFIC_GEN'))
- logger = logging.getLogger()
- logger.setLevel(logging.DEBUG)
+ _LOGGER.setLevel(logging.DEBUG)
stream_logger = logging.StreamHandler(sys.stdout)
stream_logger.setLevel(VERBOSITY_LEVELS[level])
stream_logger.setFormatter(logging.Formatter(
- '[%(levelname)s] %(asctime)s : (%(name)s) - %(message)s'))
- logger.addHandler(stream_logger)
+ '[%(levelname)-5s] %(asctime)s : (%(name)s) - %(message)s'))
+ _LOGGER.addHandler(stream_logger)
file_logger = logging.FileHandler(filename=log_file_default)
file_logger.setLevel(logging.DEBUG)
- logger.addHandler(file_logger)
+ _LOGGER.addHandler(file_logger)
class CommandFilter(logging.Filter):
"""Filter out strings beginning with 'cmd :'"""
cmd_logger = logging.FileHandler(filename=log_file_host_cmds)
cmd_logger.setLevel(logging.DEBUG)
cmd_logger.addFilter(CommandFilter())
- logger.addHandler(cmd_logger)
+ _LOGGER.addHandler(cmd_logger)
gen_logger = logging.FileHandler(filename=log_file_traffic_gen)
gen_logger.setLevel(logging.DEBUG)
gen_logger.addFilter(TrafficGenCommandFilter())
- logger.addHandler(gen_logger)
+ _LOGGER.addHandler(gen_logger)
def apply_filter(tests, tc_filter):
e.g. '' - empty string selects all tests.
:return: A list of the selected Tests.
"""
- result = []
+ # if negative filter is first we have to start with full list of tests
+ if tc_filter.strip()[0] == '!':
+ result = tests
+ else:
+ result = []
if tc_filter is None:
tc_filter = ""
if not term or term[0] != '!':
# Add matching tests from 'tests' into results
result.extend([test for test in tests \
- if test.name.lower().find(term) >= 0])
+ if test['Name'].lower().find(term) >= 0])
else:
# Term begins with '!' so we remove matching tests
result = [test for test in result \
- if test.name.lower().find(term[1:]) < 0]
+ if test['Name'].lower().find(term[1:]) < 0]
return result
system_locale = locale.getdefaultlocale()
if None in system_locale:
os.environ['LC_ALL'] = settings.getValue('DEFAULT_LOCALE')
- logging.warning("Locale was not properly configured. Default values were set. Old locale: %s, New locale: %s",
+ _LOGGER.warning("Locale was not properly configured. Default values were set. Old locale: %s, New locale: %s",
system_locale, locale.getdefaultlocale())
+def get_vswitch_names(rst_files):
+ """ Function will return a list of vSwitches detected in given ``rst_files``.
+ """
+ vswitch_names = set()
+ if len(rst_files):
+ try:
+ output = subprocess.check_output(['grep', '-h', '^* vSwitch'] + rst_files).decode().splitlines()
+ for line in output:
+ match = re.search(r'^\* vSwitch: ([^,]+)', str(line))
+ if match:
+ vswitch_names.add(match.group(1))
+
+ if len(vswitch_names):
+ return list(vswitch_names)
+
+ except subprocess.CalledProcessError:
+ _LOGGER.warning('Cannot detect vSwitches used during testing.')
+
+ # fallback to the default value
+ return ['vSwitch']
+
+
+
+def generate_final_report():
+ """ Function will check if partial test results are available
+ and generates final report in rst format.
+ """
+
+ path = settings.getValue('RESULTS_PATH')
+ # check if there are any results in rst format
+ rst_results = glob.glob(os.path.join(path, 'result*rst'))
+ pkt_processors = get_vswitch_names(rst_results)
+ if len(rst_results):
+ try:
+ test_report = os.path.join(path, '{}_{}'.format('_'.join(pkt_processors), _TEMPLATE_RST['final']))
+ # create report caption directly - it is not worth to execute jinja machinery
+ report_caption = '{}\n{} {}\n{}\n\n'.format(
+ '============================================================',
+ 'Performance report for',
+ ', '.join(pkt_processors),
+ '============================================================')
+
+ with open(_TEMPLATE_RST['tmp'], 'w') as file_:
+ file_.write(report_caption)
+
+ retval = subprocess.call('cat {} {} {} {} > {}'.format(_TEMPLATE_RST['tmp'], _TEMPLATE_RST['head'],
+ ' '.join(rst_results), _TEMPLATE_RST['foot'],
+ test_report), shell=True)
+ if retval == 0 and os.path.isfile(test_report):
+ _LOGGER.info('Overall test report written to "%s"', test_report)
+ else:
+ _LOGGER.error('Generation of overall test report has failed.')
+
+ # remove temporary file
+ os.remove(_TEMPLATE_RST['tmp'])
+
+ except subprocess.CalledProcessError:
+ _LOGGER.error('Generatrion of overall test report has failed.')
+
+
+def enable_sriov(nic_list):
+ """ Enable SRIOV for given enhanced PCI IDs
+
+ :param nic_list: A list of enhanced PCI IDs
+ """
+ # detect if sriov is required
+ sriov_nic = {}
+ for nic in nic_list:
+ if networkcard.is_sriov_nic(nic):
+ tmp_nic = nic.split('|')
+ if tmp_nic[0] in sriov_nic:
+ if int(tmp_nic[1][2:]) > sriov_nic[tmp_nic[0]]:
+ sriov_nic[tmp_nic[0]] = int(tmp_nic[1][2:])
+ else:
+ sriov_nic.update({tmp_nic[0] : int(tmp_nic[1][2:])})
+
+ # sriov is required for some NICs
+ if len(sriov_nic):
+ for nic in sriov_nic:
+ # check if SRIOV is supported and enough virt interfaces are available
+ if not networkcard.is_sriov_supported(nic) \
+ or networkcard.get_sriov_numvfs(nic) <= sriov_nic[nic]:
+ # if not, enable and set appropriate number of VFs
+ if not networkcard.set_sriov_numvfs(nic, sriov_nic[nic] + 1):
+ raise RuntimeError('SRIOV cannot be enabled for NIC {}'.format(nic))
+ else:
+ _LOGGER.debug("SRIOV enabled for NIC %s", nic)
+
+ # WORKAROUND: it has been observed with IXGBE(VF) driver,
+ # that NIC doesn't correclty dispatch traffic to VFs based
+ # on their MAC address. Unbind and bind to the same driver
+ # solves this issue.
+ networkcard.reinit_vfs(nic)
+
+ # After SRIOV is enabled it takes some time until network drivers
+ # properly initialize all cards.
+ # Wait also in case, that SRIOV was already configured as it can be
+ # configured automatically just before vsperf execution.
+ time.sleep(2)
+
+ return True
+
+ return False
+
+
+def disable_sriov(nic_list):
+ """ Disable SRIOV for given PCI IDs
+
+ :param nic_list: A list of enhanced PCI IDs
+ """
+ for nic in nic_list:
+ if networkcard.is_sriov_nic(nic):
+ if not networkcard.set_sriov_numvfs(nic.split('|')[0], 0):
+ raise RuntimeError('SRIOV cannot be disabled for NIC {}'.format(nic))
+ else:
+ _LOGGER.debug("SRIOV disabled for NIC %s", nic.split('|')[0])
+
+
+def handle_list_options(args):
+ """ Process --list cli arguments if needed
+
+ :param args: A dictionary with all CLI arguments
+ """
+ if args['list_trafficgens']:
+ print(Loader().get_trafficgens_printable())
+ sys.exit(0)
+
+ if args['list_collectors']:
+ print(Loader().get_collectors_printable())
+ sys.exit(0)
+
+ if args['list_vswitches']:
+ print(Loader().get_vswitches_printable())
+ sys.exit(0)
+
+ if args['list_vnfs']:
+ print(Loader().get_vnfs_printable())
+ sys.exit(0)
+
+ if args['list_fwdapps']:
+ print(Loader().get_pktfwds_printable())
+ sys.exit(0)
+
+ if args['list_settings']:
+ print(str(settings))
+ sys.exit(0)
+
+ if args['list']:
+ # configure tests
+ if args['integration']:
+ testcases = settings.getValue('INTEGRATION_TESTS')
+ else:
+ testcases = settings.getValue('PERFORMANCE_TESTS')
+
+ print("Available Tests:")
+ print("================")
+
+ for test in testcases:
+ print('* %-30s %s' % ('%s:' % test['Name'], test['Description']))
+ sys.exit(0)
+
+
+def vsperf_finalize():
+ """ Clean up before exit
+ """
+ # remove directory if no result files were created
+ try:
+ results_path = settings.getValue('RESULTS_PATH')
+ if os.path.exists(results_path):
+ files_list = os.listdir(results_path)
+ if files_list == []:
+ _LOGGER.info("Removing empty result directory: " + results_path)
+ shutil.rmtree(results_path)
+ except AttributeError:
+ # skip it if parameter doesn't exist
+ pass
+
+ # disable SRIOV if needed
+ try:
+ if settings.getValue('SRIOV_ENABLED'):
+ disable_sriov(settings.getValue('WHITELIST_NICS_ORIG'))
+ except AttributeError:
+ # skip it if parameter doesn't exist
+ pass
+
+
class MockTestCase(unittest.TestCase):
"""Allow use of xmlrunner to generate Jenkins compatible output without
using xmlrunner to actually run tests.
on how self.is_pass was set in the constructor"""
self.assertTrue(self.is_pass, self.msg)
-
+# pylint: disable=too-many-locals, too-many-branches, too-many-statements
def main():
"""Main function.
"""
# configure settings
- settings.load_from_dir('conf')
+ settings.load_from_dir(os.path.join(_CURR_DIR, 'conf'))
+
+ # Load non performance/integration tests
+ if args['integration']:
+ settings.load_from_dir(os.path.join(_CURR_DIR, 'conf/integration'))
# load command line parameters first in case there are settings files
# to be used
# than both a settings file and environment variables
settings.load_from_dict(args)
- # set dpdk and ovs paths accorfing to VNF and VSWITCH
- if settings.getValue('VSWITCH').endswith('Vanilla'):
- # settings paths for Vanilla
- settings.setValue('OVS_DIR', (settings.getValue('OVS_DIR_VANILLA')))
- elif settings.getValue('VSWITCH').endswith('Vhost'):
- if settings.getValue('VNF').endswith('Cuse'):
- # settings paths for Cuse
- settings.setValue('RTE_SDK', (settings.getValue('RTE_SDK_CUSE')))
- settings.setValue('OVS_DIR', (settings.getValue('OVS_DIR_CUSE')))
- else:
- # settings paths for VhostUser
- settings.setValue('RTE_SDK', (settings.getValue('RTE_SDK_USER')))
- settings.setValue('OVS_DIR', (settings.getValue('OVS_DIR_USER')))
- else:
- # default - set to VHOST USER but can be changed during enhancement
- settings.setValue('RTE_SDK', (settings.getValue('RTE_SDK_USER')))
- settings.setValue('OVS_DIR', (settings.getValue('OVS_DIR_USER')))
+ settings.setValue('mode', args['mode'])
+
+ # set dpdk and ovs paths according to VNF and VSWITCH
+ if settings.getValue('mode') != 'trafficgen':
+ functions.settings_update_paths()
+
+ # if required, handle list-* operations
+ handle_list_options(args)
configure_logging(settings.getValue('VERBOSITY'))
- logger = logging.getLogger()
# check and fix locale
check_and_set_locale()
# configure trafficgens
-
if args['trafficgen']:
trafficgens = Loader().get_trafficgens()
if args['trafficgen'] not in trafficgens:
- logging.error('There are no trafficgens matching \'%s\' found in'
+ _LOGGER.error('There are no trafficgens matching \'%s\' found in'
' \'%s\'. Exiting...', args['trafficgen'],
settings.getValue('TRAFFICGEN_DIR'))
sys.exit(1)
- # configure vswitch
+ # configuration validity checks
if args['vswitch']:
- vswitches = Loader().get_vswitches()
- if args['vswitch'] not in vswitches:
- logging.error('There are no vswitches matching \'%s\' found in'
- ' \'%s\'. Exiting...', args['vswitch'],
- settings.getValue('VSWITCH_DIR'))
+ vswitch_none = args['vswitch'].strip().lower() == 'none'
+ if vswitch_none:
+ settings.setValue('VSWITCH', 'none')
+ else:
+ vswitches = Loader().get_vswitches()
+ if args['vswitch'] not in vswitches:
+ _LOGGER.error('There are no vswitches matching \'%s\' found in'
+ ' \'%s\'. Exiting...', args['vswitch'],
+ settings.getValue('VSWITCH_DIR'))
+ sys.exit(1)
+
+ if args['fwdapp']:
+ settings.setValue('PKTFWD', args['fwdapp'])
+ fwdapps = Loader().get_pktfwds()
+ if args['fwdapp'] not in fwdapps:
+ _LOGGER.error('There are no forwarding application'
+ ' matching \'%s\' found in'
+ ' \'%s\'. Exiting...', args['fwdapp'],
+ settings.getValue('PKTFWD_DIR'))
sys.exit(1)
if args['vnf']:
vnfs = Loader().get_vnfs()
if args['vnf'] not in vnfs:
- logging.error('there are no vnfs matching \'%s\' found in'
+ _LOGGER.error('there are no vnfs matching \'%s\' found in'
' \'%s\'. exiting...', args['vnf'],
- settings.getValue('vnf_dir'))
+ settings.getValue('VNF_DIR'))
sys.exit(1)
- if args['duration']:
- if args['duration'].isdigit() and int(args['duration']) > 0:
- settings.setValue('duration', args['duration'])
+ if args['exact_test_name'] and args['tests']:
+ _LOGGER.error("Cannot specify tests with both positional args and --test.")
+ sys.exit(1)
+
+ # modify NIC configuration to decode enhanced PCI IDs
+ wl_nics_orig = list(networkcard.check_pci(pci) for pci in settings.getValue('WHITELIST_NICS'))
+ settings.setValue('WHITELIST_NICS_ORIG', wl_nics_orig)
+
+ # sriov handling is performed on checked/expanded PCI IDs
+ settings.setValue('SRIOV_ENABLED', enable_sriov(wl_nics_orig))
+
+ nic_list = []
+ for nic in wl_nics_orig:
+ tmp_nic = networkcard.get_nic_info(nic)
+ if tmp_nic:
+ nic_list.append({'pci' : tmp_nic,
+ 'type' : 'vf' if networkcard.get_sriov_pf(tmp_nic) else 'pf',
+ 'mac' : networkcard.get_mac(tmp_nic),
+ 'driver' : networkcard.get_driver(tmp_nic),
+ 'device' : networkcard.get_device_name(tmp_nic)})
else:
- logging.error('The selected Duration is not a number')
- sys.exit(1)
+ vsperf_finalize()
+ raise RuntimeError("Invalid network card PCI ID: '{}'".format(nic))
- # update global settings
- guest_loopback = get_test_param('guest_loopback', None)
- if guest_loopback:
- tmp_gl = []
- for i in range(len(settings.getValue('GUEST_LOOPBACK'))):
- tmp_gl.append(guest_loopback)
- settings.setValue('GUEST_LOOPBACK', tmp_gl)
+ settings.setValue('NICS', nic_list)
+ # for backward compatibility
+ settings.setValue('WHITELIST_NICS', list(nic['pci'] for nic in nic_list))
# generate results directory name
date = datetime.datetime.fromtimestamp(time.time())
results_dir = "results_" + date.strftime('%Y-%m-%d_%H-%M-%S')
results_path = os.path.join(settings.getValue('LOG_DIR'), results_dir)
-
- # configure tests
- testcases = settings.getValue('PERFORMANCE_TESTS')
- all_tests = []
- for cfg in testcases:
- try:
- all_tests.append(TestCase(cfg, results_path))
- except (Exception) as _:
- logger.exception("Failed to create test: %s",
- cfg.get('Name', '<Name not set>'))
- raise
-
- # if required, handle list-* operations
-
- if args['list']:
- print("Available Tests:")
- print("======")
- for test in all_tests:
- print('* %-18s%s' % ('%s:' % test.name, test.desc))
- exit()
-
- if args['list_trafficgens']:
- print(Loader().get_trafficgens_printable())
- exit()
-
- if args['list_collectors']:
- print(Loader().get_collectors_printable())
- exit()
-
- if args['list_vswitches']:
- print(Loader().get_vswitches_printable())
- exit()
-
- if args['list_vnfs']:
- print(Loader().get_vnfs_printable())
- exit()
-
- if args['list_settings']:
- print(str(settings))
- exit()
-
- # select requested tests
- if args['exact_test_name'] and args['tests']:
- logger.error("Cannot specify tests with both positional args and --test.")
- sys.exit(1)
-
- if args['exact_test_name']:
- exact_names = args['exact_test_name']
- # positional args => exact matches only
- selected_tests = [test for test in all_tests if test.name in exact_names]
- elif args['tests']:
- # --tests => apply filter to select requested tests
- selected_tests = apply_filter(all_tests, args['tests'])
- else:
- # Default - run all tests
- selected_tests = all_tests
-
- if not selected_tests:
- logger.error("No tests matched --test option or positional args. Done.")
- sys.exit(1)
+ settings.setValue('RESULTS_PATH', results_path)
# create results directory
if not os.path.exists(results_path):
- logger.info("Creating result directory: " + results_path)
+ _LOGGER.info("Creating result directory: " + results_path)
os.makedirs(results_path)
- # run tests
- suite = unittest.TestSuite()
- for test in selected_tests:
- try:
- test.run()
- suite.addTest(MockTestCase('', True, test.name))
- #pylint: disable=broad-except
- except (Exception) as ex:
- logger.exception("Failed to run test: %s", test.name)
- suite.addTest(MockTestCase(str(ex), False, test.name))
- logger.info("Continuing with next test...")
-
- if settings.getValue('XUNIT'):
- xmlrunner.XMLTestRunner(
- output=settings.getValue('XUNIT_DIR'), outsuffix="",
- verbosity=0).run(suite)
-
- if args['opnfvpod']:
- pod_name = args['opnfvpod']
- installer_name = settings.getValue('OPNFV_INSTALLER')
- opnfv_url = settings.getValue('OPNFV_URL')
- pkg_list = settings.getValue('PACKAGE_LIST')
-
- int_data = {'cuse': False,
- 'vanilla': False,
- 'pod': pod_name,
- 'installer': installer_name,
- 'pkg_list': pkg_list,
- 'db_url': opnfv_url}
- if settings.getValue('VSWITCH').endswith('Vanilla'):
- int_data['vanilla'] = True
- if settings.getValue('VNF').endswith('Cuse'):
- int_data['cuse'] = True
- opnfvdashboard.results2opnfv_dashboard(results_path, int_data)
-
- #remove directory if no result files were created.
- if os.path.exists(results_path):
- files_list = os.listdir(results_path)
- if files_list == []:
- shutil.rmtree(results_path)
+ if settings.getValue('mode') == 'trafficgen':
+ # execute only traffic generator
+ _LOGGER.debug("Executing traffic generator:")
+ loader = Loader()
+ # set traffic details, so they can be passed to traffic ctl
+ traffic = copy.deepcopy(settings.getValue('TRAFFIC'))
+
+ traffic = functions.check_traffic(traffic)
+
+ traffic_ctl = component_factory.create_traffic(
+ traffic['traffic_type'],
+ loader.get_trafficgen_class())
+ with traffic_ctl:
+ traffic_ctl.send_traffic(traffic)
+ _LOGGER.debug("Traffic Results:")
+ traffic_ctl.print_results()
+
+ # write results into CSV file
+ result_file = os.path.join(results_path, "result.csv")
+ PerformanceTestCase.write_result_to_file(traffic_ctl.get_results(), result_file)
+ else:
+ # configure tests
+ if args['integration']:
+ testcases = settings.getValue('INTEGRATION_TESTS')
+ else:
+ testcases = settings.getValue('PERFORMANCE_TESTS')
+
+ if args['exact_test_name']:
+ exact_names = args['exact_test_name']
+ # positional args => exact matches only
+ selected_tests = [test for test in testcases if test['Name'] in exact_names]
+ elif args['tests']:
+ # --tests => apply filter to select requested tests
+ selected_tests = apply_filter(testcases, args['tests'])
+ else:
+ # Default - run all tests
+ selected_tests = testcases
-if __name__ == "__main__":
- main()
+ if not len(selected_tests):
+ _LOGGER.error("No tests matched --tests option or positional args. Done.")
+ vsperf_finalize()
+ sys.exit(1)
+ # run tests
+ # Add pylint exception: Redefinition of test type from
+ # testcases.integration.IntegrationTestCase to testcases.performance.PerformanceTestCase
+ # pylint: disable=redefined-variable-type
+ suite = unittest.TestSuite()
+ for cfg in selected_tests:
+ test_name = cfg.get('Name', '<Name not set>')
+ try:
+ if args['integration']:
+ test = IntegrationTestCase(cfg)
+ else:
+ test = PerformanceTestCase(cfg)
+ test.run()
+ suite.addTest(MockTestCase('', True, test.name))
+ # pylint: disable=broad-except
+ except (Exception) as ex:
+ _LOGGER.exception("Failed to run test: %s", test_name)
+ suite.addTest(MockTestCase(str(ex), False, test_name))
+ _LOGGER.info("Continuing with next test...")
+
+ # generate final rst report with results of all executed TCs
+ generate_final_report()
+
+ if settings.getValue('XUNIT'):
+ xmlrunner.XMLTestRunner(
+ output=settings.getValue('XUNIT_DIR'), outsuffix="",
+ verbosity=0).run(suite)
+
+ if args['opnfvpod']:
+ pod_name = args['opnfvpod']
+ installer_name = settings.getValue('OPNFV_INSTALLER')
+ opnfv_url = settings.getValue('OPNFV_URL')
+ pkg_list = settings.getValue('PACKAGE_LIST')
+
+ int_data = {'vanilla': False,
+ 'pod': pod_name,
+ 'installer': installer_name,
+ 'pkg_list': pkg_list,
+ 'db_url': opnfv_url}
+ if settings.getValue('VSWITCH').endswith('Vanilla'):
+ int_data['vanilla'] = True
+ opnfvdashboard.results2opnfv_dashboard(results_path, int_data)
+
+ # cleanup before exit
+ vsperf_finalize()
+if __name__ == "__main__":
+ main()
Code Review / vswitchperf.git / blobdiff