import os
import sys
import argparse
+import re
import time
import datetime
import shutil
import unittest
-import xmlrunner
import locale
import copy
import glob
import subprocess
-
-sys.dont_write_bytecode = True
-
+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 PerformanceTestCase
from testcases import IntegrationTestCase
from tools import functions
from tools.pkt_gen import trafficgen
from tools.opnfvdashboard import opnfvdashboard
-from tools.pkt_gen.trafficgen.trafficgenhelper import TRAFFIC_DEFAULTS
-import core.component_factory as component_factory
+
+sys.dont_write_bytecode = True
VERBOSITY_LEVELS = {
'debug': logging.DEBUG,
'critical': logging.CRITICAL
}
-_TEMPLATE_RST = {'head' : 'tools/report/report_head.rst',
- 'foot' : 'tools/report/report_foot.rst',
+_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' : 'tools/report/report_tmp_caption.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)
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())
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):
- _LOGGER.error("SRIOV cannot be enabled for NIC %s", nic)
- raise
+ raise RuntimeError('SRIOV cannot be enabled for NIC {}'.format(nic))
else:
_LOGGER.debug("SRIOV enabled for NIC %s", nic)
for nic in nic_list:
if networkcard.is_sriov_nic(nic):
if not networkcard.set_sriov_numvfs(nic.split('|')[0], 0):
- _LOGGER.error("SRIOV cannot be disabled for NIC %s", nic)
- raise
+ raise RuntimeError('SRIOV cannot be disabled for NIC {}'.format(nic))
else:
_LOGGER.debug("SRIOV disabled for NIC %s", nic.split('|')[0])
# 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('conf/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
- functions.settings_update_paths()
+ 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)
if args['vnf'] not in vnfs:
_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['exact_test_name'] and args['tests']:
_LOGGER.error("Cannot specify tests with both positional args and --test.")
sys.exit(1)
- # sriov handling
- settings.setValue('SRIOV_ENABLED', enable_sriov(settings.getValue('WHITELIST_NICS')))
-
# 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)
'driver' : networkcard.get_driver(tmp_nic),
'device' : networkcard.get_device_name(tmp_nic)})
else:
- _LOGGER.error("Invalid network card PCI ID: '%s'", nic)
vsperf_finalize()
- raise
+ 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))
- # update global settings
- guest_loopback = get_test_param('guest_loopback', None)
- if guest_loopback:
- tmp_gl = []
- for dummy_i in range(len(settings.getValue('GUEST_LOOPBACK'))):
- tmp_gl.append(guest_loopback)
- settings.setValue('GUEST_LOOPBACK', tmp_gl)
-
- settings.setValue('mode', args['mode'])
-
# generate results directory name
date = datetime.datetime.fromtimestamp(time.time())
results_dir = "results_" + date.strftime('%Y-%m-%d_%H-%M-%S')
_LOGGER.debug("Executing traffic generator:")
loader = Loader()
# set traffic details, so they can be passed to traffic ctl
- traffic = copy.deepcopy(TRAFFIC_DEFAULTS)
- traffic.update({'traffic_type': get_test_param('traffic_type', 'rfc2544'),
- 'bidir': get_test_param('bidirectional', 'False'),
- 'multistream': int(get_test_param('multistream', 0)),
- 'stream_type': get_test_param('stream_type', 'L4'),
- 'frame_rate': int(get_test_param('iload', 100))})
+ traffic = copy.deepcopy(settings.getValue('TRAFFIC'))
traffic_ctl = component_factory.create_traffic(
traffic['traffic_type'],
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']:
opnfv_url = settings.getValue('OPNFV_URL')
pkg_list = settings.getValue('PACKAGE_LIST')
- int_data = {'cuse': False,
- 'vanilla': False,
+ 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
- if settings.getValue('VNF').endswith('Cuse'):
- int_data['cuse'] = True
opnfvdashboard.results2opnfv_dashboard(results_path, int_data)
# cleanup before exit
if __name__ == "__main__":
main()
-