#!/usr/bin/env python3 # 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. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """VSPERF main script. """ import logging import os import sys import argparse import re import time import datetime import shutil import unittest import locale import copy import glob import subprocess import ast import xmlrunner from conf import settings import core.component_factory as component_factory from core.loader import Loader 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, 'warning': logging.WARNING, 'error': logging.ERROR, '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(): """ Parse command line arguments. """ class _SplitTestParamsAction(argparse.Action): """ Parse and split the '--test-params' argument. 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)' """ def __call__(self, parser, namespace, values, option_string=None): results = {} 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) class _ValidateFileAction(argparse.Action): """Validate a file can be read from before using it. """ def __call__(self, parser, namespace, values, option_string=None): if not os.path.isfile(values): raise argparse.ArgumentTypeError( 'the path \'%s\' is not a valid path' % values) elif not os.access(values, os.R_OK): raise argparse.ArgumentTypeError( 'the path \'%s\' is not accessible' % values) setattr(namespace, self.dest, values) class _ValidateDirAction(argparse.Action): """Validate a directory can be written to before using it. """ def __call__(self, parser, namespace, values, option_string=None): if not os.path.isdir(values): raise argparse.ArgumentTypeError( 'the path \'%s\' is not a valid path' % values) elif not os.access(values, os.W_OK): raise argparse.ArgumentTypeError( 'the path \'%s\' is not accessible' % values) setattr(namespace, self.dest, values) def list_logging_levels(): """Give a summary of all available logging levels. :return: List of verbosity level names in decreasing order of verbosity """ return sorted(VERBOSITY_LEVELS.keys(), key=lambda x: VERBOSITY_LEVELS[x]) parser = argparse.ArgumentParser(prog=__file__, formatter_class= argparse.ArgumentDefaultsHelpFormatter) parser.add_argument('--version', action='version', version='%(prog)s 0.2') parser.add_argument('--list', '--list-tests', action='store_true', help='list all tests and exit') parser.add_argument('--list-trafficgens', action='store_true', help='list all traffic generators and exit') parser.add_argument('--list-collectors', action='store_true', 
# ==========================================================================
# Installing dtb files
#
# Installs all dtb files listed in $(dtb-y) either in the
# INSTALL_DTBS_PATH directory or the default location:
#
#   $INSTALL_PATH/dtbs/$KERNELRELEASE
#
# Traverse through subdirectories listed in $(dts-dirs).
# ==========================================================================

src := $(obj)

PHONY := __dtbs_install
__dtbs_install:

export dtbinst-root ?= $(obj)

include include/config/auto.conf
include scripts/Kbuild.include
include $(src)/Makefile

PHONY += __dtbs_install_prep
__dtbs_install_prep:
ifeq ("$(dtbinst-root)", "$(obj)")
	$(Q)if [ -d $(INSTALL_DTBS_PATH).old ]; then rm -rf $(INSTALL_DTBS_PATH).old; fi
	$(Q)if [ -d $(INSTALL_DTBS_PATH) ]; then mv $(INSTALL_DTBS_PATH) $(INSTALL_DTBS_PATH).old; fi
	$(Q)mkdir -p $(INSTALL_DTBS_PATH)
endif

dtbinst-files	:= $(dtb-y)
dtbinst-dirs	:= $(dts-dirs)

# Helper targets for Installing DTBs into the boot directory
quiet_cmd_dtb_install =	INSTALL $<
      cmd_dtb_install =	mkdir -p $(2); cp $< $(2)

install-dir = $(patsubst $(dtbinst-root)%,$(INSTALL_DTBS_PATH)%,$(obj))

$(dtbinst-files) $(dtbinst-dirs): | __dtbs_install_prep

$(dtbinst-files): %.dtb: $(obj)/%.dtb
	$(call cmd,dtb_install,$(install-dir))

$(dtbinst-dirs):
	$(Q)$(MAKE) $(dtbinst)=$(obj)/$@

PHONY += $(dtbinst-files) $(dtbinst-dirs)
__dtbs_install: $(dtbinst-files) $(dtbinst-dirs)

.PHONY: $(PHONY)
TestCase(unittest.TestCase): """Allow use of xmlrunner to generate Jenkins compatible output without using xmlrunner to actually run tests. Usage: suite = unittest.TestSuite() suite.addTest(MockTestCase('Test1 passed ', True, 'Test1')) suite.addTest(MockTestCase('Test2 failed because...', False, 'Test2')) xmlrunner.XMLTestRunner(...).run(suite) """ def __init__(self, msg, is_pass, test_name): #remember the things self.msg = msg self.is_pass = is_pass #dynamically create a test method with the right name #but point the method at our generic test method setattr(MockTestCase, test_name, self.generic_test) super(MockTestCase, self).__init__(test_name) def generic_test(self): """Provide a generic function that raises or not based 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. """ args = parse_arguments() # configure settings 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 settings.load_from_dict(args) if args['conf_file']: settings.load_from_file(args['conf_file']) if args['load_env']: settings.load_from_env() # reload command line parameters since these should take higher priority # than both a settings file and environment variables settings.load_from_dict(args) settings.setValue('mode', args['mode']) # update paths to trafficgens if required if settings.getValue('mode') == 'trafficgen': functions.settings_update_paths() # if required, handle list-* operations handle_list_options(args) configure_logging(settings.getValue('VERBOSITY')) # check and fix locale check_and_set_locale() # configure trafficgens if args['trafficgen']: trafficgens = Loader().get_trafficgens() if args['trafficgen'] not in trafficgens: _LOGGER.error('There are no trafficgens matching \'%s\' found in' ' \'%s\'. Exiting...', args['trafficgen'], settings.getValue('TRAFFICGEN_DIR')) sys.exit(1) # configuration validity checks if args['vswitch']: 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: _LOGGER.error('there are no vnfs matching \'%s\' found in' ' \'%s\'. exiting...', args['vnf'], settings.getValue('VNF_DIR')) sys.exit(1) if args['loadgen']: loadgens = Loader().get_loadgens() if args['loadgen'] not in loadgens: _LOGGER.error('There are no loadgens matching \'%s\' found in' ' \'%s\'. Exiting...', args['loadgen'], settings.getValue('LOADGEN_DIR')) sys.exit(1) 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: vsperf_finalize() raise RuntimeError("Invalid network card PCI ID: '{}'".format(nic)) 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) settings.setValue('RESULTS_PATH', results_path) # create results directory if not os.path.exists(results_path): _LOGGER.info("Creating result directory: " + results_path) os.makedirs(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 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() settings_snapshot = copy.deepcopy(settings.__dict__) for cfg in selected_tests: test_name = cfg.get('Name', '') 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...") finally: settings.restore_from_dict(settings_snapshot) # 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 = str(settings.getValue('OPNFV_INSTALLER')).lower() opnfv_url = settings.getValue('OPNFV_URL') pkg_list = settings.getValue('PACKAGE_LIST') int_data = {'pod': pod_name, 'build_tag': get_build_tag(), 'installer': installer_name, 'pkg_list': pkg_list, 'db_url': opnfv_url, # pass vswitch name from configuration to be used for failed # TCs; In case of successful TCs it is safer to use vswitch # name from CSV as TC can override global configuration 'vswitch': str(settings.getValue('VSWITCH')).lower()} tc_names = [tc['Name'] for tc in selected_tests] opnfvdashboard.results2opnfv_dashboard(tc_names, results_path, int_data) # cleanup before exit vsperf_finalize() if __name__ == "__main__": main()