import time
import datetime
import shutil
+import unittest
+import xmlrunner
sys.dont_write_bytecode = True
from core.loader import Loader
from testcases import TestCase
from tools import tasks
-from tools.collectors import collector
from tools.pkt_gen import trafficgen
VERBOSITY_LEVELS = {
"""
Parse and split the '--test-params' argument.
- This expects either 'x=y' or 'x' (implicit true) values.
+ 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 = {}
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-vnfs', action='store_true',
+ 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')
group.add_argument('--verbosity', choices=list_logging_levels(),
help='debug level')
group.add_argument('--trafficgen', help='traffic generator to use')
+ group.add_argument('--vswitch', help='vswitch implementation 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',
+ help='enable xUnit-formatted output')
+ group.add_argument('--xunit-dir', action=_ValidateDirAction,
+ help='output directory of xUnit-formatted output')
group.add_argument('--load-env', action='store_true',
help='enable loading of settings from the environment')
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;...')
+ help='csv list of test parameters: key=val; e.g.'
+ 'including pkt_sizes=x,y; duration=x; '
+ 'rfc2544_trials=x ...')
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
+class MockTestCase(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)
+
+
def main():
"""Main function.
"""
# 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()
settings.getValue('TRAFFICGEN_DIR'))
sys.exit(1)
+ # configure vswitch
+ 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'))
+ 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'
+ ' \'%s\'. exiting...', args['vnf'],
+ settings.getValue('vnf_dir'))
+ sys.exit(1)
+
+ if args['duration']:
+ if args['duration'].isdigit() and int(args['duration']) > 0:
+ settings.setValue('duration', args['duration'])
+ else:
+ logging.error('The selected Duration is not a number')
+ sys.exit(1)
# generate results directory name
date = datetime.datetime.fromtimestamp(time.time())
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(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)
+
# create results directory
- if not os.path.exists(results_dir):
+ if not os.path.exists(results_path):
logger.info("Creating result directory: " + results_path)
os.makedirs(results_path)
# run tests
- for test in all_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 _:
+ 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)
+
#remove directory if no result files were created.
if os.path.exists(results_path):
- if os.listdir(results_path) == []:
+ files_list = os.listdir(results_path)
+ if files_list == []:
shutil.rmtree(results_path)
if __name__ == "__main__":
Code Review / vswitchperf.git / blobdiff