#!/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 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,
value = value.strip()
if len(param):
if len(value):
- results[param] = 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
_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
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(settings.getValue('VSWITCH'), _TEMPLATE_RST['final']))
+ 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
- if settings.getValue('VSWITCH').lower() != 'none':
- pkt_processor = Loader().get_vswitches()[settings.getValue('VSWITCH')].__doc__.strip().split('\n')[0]
- else:
- pkt_processor = Loader().get_pktfwds()[settings.getValue('PKTFWD')].__doc__.strip().split('\n')[0]
report_caption = '{}\n{} {}\n{}\n\n'.format(
'============================================================',
'Performance report for',
- pkt_processor,
+ ', '.join(pkt_processors),
'============================================================')
with open(_TEMPLATE_RST['tmp'], 'w') as file_:
if retval == 0 and os.path.isfile(test_report):
_LOGGER.info('Overall test report written to "%s"', test_report)
else:
- _LOGGER.error('Generatrion of overall test report has failed.')
+ _LOGGER.error('Generation of overall test report has failed.')
# remove temporary file
os.remove(_TEMPLATE_RST['tmp'])
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])
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.
"""
settings.setValue('mode', args['mode'])
- # set dpdk and ovs paths accorfing to VNF and VSWITCH
+ # set dpdk and ovs paths according to VNF and VSWITCH
if settings.getValue('mode') != 'trafficgen':
functions.settings_update_paths()
# configuration validity checks
if args['vswitch']:
- vswitch_none = 'none' == args['vswitch'].strip().lower()
+ vswitch_none = args['vswitch'].strip().lower() == 'none'
if vswitch_none:
settings.setValue('VSWITCH', 'none')
else:
'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
_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', TRAFFIC_DEFAULTS['traffic_type']),
- 'bidir': get_test_param('bidirectional', TRAFFIC_DEFAULTS['bidir']),
- 'multistream': int(get_test_param('multistream', TRAFFIC_DEFAULTS['multistream'])),
- 'stream_type': get_test_param('stream_type', TRAFFIC_DEFAULTS['stream_type']),
- 'frame_rate': int(get_test_param('iload', TRAFFIC_DEFAULTS['frame_rate']))})
+ traffic = copy.deepcopy(settings.getValue('TRAFFIC'))
+
+ traffic = functions.check_traffic(traffic)
traffic_ctl = component_factory.create_traffic(
traffic['traffic_type'],
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>')
test = PerformanceTestCase(cfg)
test.run()
suite.addTest(MockTestCase('', True, test.name))
- #pylint: disable=broad-except
+ # 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))
if __name__ == "__main__":
main()
-
Code Review / vswitchperf.git / blobdiff