import shutil
import unittest
import xmlrunner
+import locale
sys.dont_write_bytecode = True
from conf import settings
from core.loader import Loader
from testcases import TestCase
-from tools.report import report
from tools import tasks
-from tools.collectors import collector
from tools.pkt_gen import trafficgen
-from vswitches import vswitch
-from vnfs import vnf
+from tools.opnfvdashboard import opnfvdashboard
VERBOSITY_LEVELS = {
'debug': logging.DEBUG,
help='list all system vnfs and exit')
parser.add_argument('--list-settings', action='store_true',
help='list effective settings configuration and exit')
- parser.add_argument('test', nargs='*', help='test specification(s)')
+ parser.add_argument('exact_test_name', nargs='*', help='Exact names of\
+ tests to run. E.g "vsperf phy2phy_tput phy2phy_cont"\
+ runs only the two tests with those exact names.\
+ To run all tests omit both positional args and --tests arg.')
group = parser.add_argument_group('test selection options')
group.add_argument('-f', '--test-spec', help='test specification file')
help='csv list of test parameters: key=val; e.g.'
'including pkt_sizes=x,y; duration=x; '
'rfc2544_trials=x ...')
+ group.add_argument('--opnfvpod', help='name of POD in opnfv')
args = vars(parser.parse_args())
log_file_traffic_gen = os.path.join(
settings.getValue('LOG_DIR'),
settings.getValue('LOG_FILE_TRAFFIC_GEN'))
- log_file_sys_metrics = os.path.join(
- settings.getValue('LOG_DIR'),
- settings.getValue('LOG_FILE_SYS_METRICS'))
logger = logging.getLogger()
logger.setLevel(logging.DEBUG)
def filter(self, record):
return record.getMessage().startswith(trafficgen.CMD_PREFIX)
- class SystemMetricsCommandFilter(logging.Filter):
- """Filter out strings beginning with 'gencmd :'"""
- def filter(self, record):
- return record.getMessage().startswith(collector.CMD_PREFIX)
-
cmd_logger = logging.FileHandler(filename=log_file_host_cmds)
cmd_logger.setLevel(logging.DEBUG)
cmd_logger.addFilter(CommandFilter())
gen_logger.addFilter(TrafficGenCommandFilter())
logger.addHandler(gen_logger)
- metrics_logger = logging.FileHandler(filename=log_file_sys_metrics)
- metrics_logger.setLevel(logging.DEBUG)
- metrics_logger.addFilter(SystemMetricsCommandFilter())
- logger.addHandler(metrics_logger)
-
def apply_filter(tests, tc_filter):
"""Allow a subset of tests to be conveniently selected
return result
+def check_and_set_locale():
+ """ Function will check locale settings. In case, that it isn't configured
+ properly, then default values specified by DEFAULT_LOCALE will be used.
+ """
+
+ 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",
+ system_locale, locale.getdefaultlocale())
+
class MockTestCase(unittest.TestCase):
"""Allow use of xmlrunner to generate Jenkins compatible output without
using xmlrunner to actually run tests.
# 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')))
+
configure_logging(settings.getValue('VERBOSITY'))
logger = logging.getLogger()
+ # check and fix locale
+ check_and_set_locale()
+
# configure trafficgens
if args['trafficgen']:
cfg.get('Name', '<Name not set>'))
raise
- # TODO(BOM) Apply filter to select requested tests
- all_tests = apply_filter(all_tests, args['tests'])
-
# if required, handle list-* operations
if args['list']:
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)
+
# create results directory
if not os.path.exists(results_path):
logger.info("Creating result directory: " + results_path)
os.makedirs(results_path)
- suite = unittest.TestSuite()
-
# run tests
- for test in all_tests:
+ suite = unittest.TestSuite()
+ for test in selected_tests:
try:
test.run()
suite.addTest(MockTestCase('', True, test.name))
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)
- else:
- for file in files_list:
- # generate report from all csv files
- if file[-3:] == 'csv':
- results_csv = os.path.join(results_path, file)
- if os.path.isfile(results_csv) and os.access(results_csv, os.R_OK):
- report.generate(testcases, results_csv)
if __name__ == "__main__":
main()