#!/usr/bin/env python3
# SPDX-License-Identifier: BSD-2-Clause OR GPL-2.0-only
# SPDX-FileCopyrightText: Copyright Amazon.com, Inc. or its affiliates. All rights reserved
import argparse
import builtins
import os
import sys
import yaml
import pytest
from junitparser import JUnitXml
from pytest import ExitCode
def get_option_longform(option_name, option_params):
'''
get the long form command line option name of an option
'''
return option_params.get("longform", "--" + option_name.replace("_", "-"))
def get_ubertest_test_type(fabtests_testsets):
test_list = fabtests_testsets.split(",")
for test in test_list:
if test == "quick" or test == "ubertest_quick" or test == "ubertest":
return "quick"
if test == "all" or test == "ubertest_all":
return "all"
if test == "verify" or test == "ubertest_verify":
return "verify"
return None
def fabtests_testsets_to_pytest_markers(fabtests_testsets, run_mode=None):
if run_mode:
assert run_mode in ["serial", "parallel"]
test_set = set()
test_list = fabtests_testsets.split(",")
# use set() to remove duplicate test set
for test in test_list:
if test == "quick":
test_set.add("unit")
test_set.add("functional")
test_set.add("short")
test_set.add("ubertest_quick")
elif test =="ubertest":
test_set.add("ubertest_quick")
elif test == "all":
test_set.add("unit")
test_set.add("functional")
test_set.add("standard")
test_set.add("multinode")
test_set.add("ubertest_all")
elif test == "verify":
test_set.add("ubertest_verify")
else:
test_set.add(test)
markers = None
for test in test_set:
if markers is None:
markers = test[:]
else:
markers += " or " + test
if run_mode:
if run_mode == "serial":
markers = "(" + markers + ") and (serial)"
else:
assert run_mode == "parallel"
markers = "(" + markers + ") and (not serial)"
return markers
def get_default_exclusion_file(fabtests_args):
test_configs_dir = os.path.abspath(os.path.join(get_pytest_root_dir(), "..", "test_configs"))
exclusion_file = os.path.join(test_configs_dir, fabtests_args.provider,
fabtests_args.provider + ".exclude")
if not os.path.exists(exclusion_file):
return None
return exclusion_file
def get_default_ubertest_config_file(fabtests_args):
test_configs_dir = os.path.abspath(os.path.join(get_pytest_root_dir(), "..", "test_configs"))
provider = fabtests_args.provider
if provider.find(";") != -1:
core,util = fabtests_args.provider.split(";")
cfg_file = os.path.join(test_configs_dir, util, core + ".test")
else:
core = fabtests_args.provider
ubertest_test_type = get_ubertest_test_type(fabtests_args.testsets)
if not ubertest_test_type:
return None
cfg_file = os.path.join(test_configs_dir, core, ubertest_test_type + ".test")
if not os.path.exists(cfg_file):
return None
return cfg_file
def add_common_arguments(parser, shared_options):
for option_name in shared_options.keys():
option_params = shared_options[option_name]
option_longform = get_option_longform(option_name, option_params)
option_shortform = option_params.get("shortform")
option_type = option_params["type"]
option_helpmsg = option_params["help"]
option_default = option_params.get("default")
if option_type == "int" and not (option_default is None):
option_default = int(option_default)
if option_shortform:
forms = [option_shortform, option_longform]
else:
forms = [option_longform]
if option_type == "bool" or option_type == "boolean":
parser.add_argument(*forms,
dest=option_name, action="store_true",
help=option_helpmsg, default=option_default)
else:
assert option_type == "str" or option_type == "int"
parser.add_argument(*forms,
dest=option_name, type=getattr(builtins, option_type),
help=option_helpmsg, default=option_default)
def fabtests_args_to_pytest_args(fabtests_args, shared_options, run_mode):
pytest_args = []
if run_mode == "parallel":
pytest_args.append("-n")
pytest_args.append(str(fabtests_args.nworkers))
pytest_args.append("--provider=" + fabtests_args.provider)
pytest_args.append("--server-id=" + fabtests_args.server_id)
pytest_args.append("--client-id=" + fabtests_args.client_id)
# -v make pytest to print 1 line for each test
pytest_args.append("-v")
pytest_verbose_options = {
0 : "-rN", # print no extra information
1 : "-rfE", # print extra information for failed test(s)
2 : "-rfEsx", # print extra information for failed/skipped test(s)
3 : "-rA", # print extra information for all test(s) (failed/skipped/passed)
}
pytest_args.append(pytest_verbose_options[fabtests_args.verbose])
verbose_fail = fabtests_args.verbose > 0
if verbose_fail:
# Use short python trace back because it show captured stdout of failed tests
pytest_args.append("--tb=short")
else:
pytest_args.append("--tb=no")
markers = fabtests_testsets_to_pytest_markers(fabtests_args.testsets, run_mode)
pytest_args.append("-m")
pytest_args.append(markers)
if fabtests_args.expression:
pytest_args.append("-k")
pytest_args.append(fabtests_args.expression)
if fabtests_args.html:
pytest_args.append("--html")
pytest_args.append(os.path.abspath(fabtests_args.html))
pytest_args.append("--self-contained-html")
if fabtests_args.junit_xml:
pytest_args.append("--junit-xml")
file_name = os.path.abspath(fabtests_args.junit_xml)
if run_mode:
file_name += "." + run_mode
pytest_args.append(file_name)
if fabtests_args.junit_logging:
pytest_args.append("-o")
pytest_args.append("junit_logging=" + fabtests_args.junit_logging)
# add options shared between runfabtests.py and libfabric pytest
for option_name in shared_options.keys():
option_params = shared_options[option_name]
option_longform = get_option_longform(option_name, option_params)
option_type = option_params["type"]
if not hasattr(fabtests_args, option_name):
continue
option_value = getattr(fabtests_args, option_name)
if (option_value is None):
continue
if option_type == "bool" or option_type == "boolean":
assert option_value
pytest_args.append(option_longform)
else:
assert option_type == "str" or option_type == "int"
pytest_args.append(option_longform + "=" + str(option_value))
if not hasattr(fabtests_args, "exclusion_file") or not fabtests_args.exclusion_file:
default_exclusion_file = get_default_exclusion_file(fabtests_args)
if default_exclusion_file:
pytest_args.append("--exclusion-file=" + default_exclusion_file)
if not hasattr(fabtests_args, "ubertest_config_file") or not fabtests_args.ubertest_config_file:
default_ubertest_config_file = get_default_ubertest_config_file(fabtests_args)
if default_ubertest_config_file:
pytest_args.append("--ubertest-config-file=" + default_ubertest_config_file)
return pytest_args
def get_pytest_root_dir():
'''
find the pytest root directory according the location of runfabtests.py
'''
script_path = os.path.abspath(sys.argv[0])
script_dir = os.path.dirname(script_path)
if os.path.basename(script_dir) == "bin":
# runfabtests.py is part of a fabtests installation
pytest_root_dir = os.path.abspath(os.path.join(script_dir, "..", "share", "fabtests", "pytest"))
elif os.path.basename(script_dir) == "scripts":
# runfabtests.py is part of a fabtests source code
pytest_root_dir = os.path.abspath(os.path.join(script_dir, "..", "pytest"))
else:
raise RuntimeError("Error: runfabtests.py is under directory {}, "
"which is neither part of fabtests installation "
"nor part of fabetsts source code".format(script_dir))
if not os.path.exists(pytest_root_dir):
raise RuntimeError("Deduced pytest root directory {} does not exist!".format(pytest_root_dir))
return pytest_root_dir
def get_pytest_relative_case_dir(fabtests_args, pytest_root_dir):
'''
the directory that contains test cases, relative to pytest_root_dir
'''
# provider's own test directory (if exists) overrides default
pytest_case_dir = os.path.join(pytest_root_dir, fabtests_args.provider)
if os.path.exists(pytest_case_dir):
return fabtests_args.provider
assert os.path.exists(os.path.join(pytest_root_dir, "default"))
return "default"
def run(fabtests_args, shared_options, run_mode):
prev_cwd = os.getcwd()
pytest_root_dir = get_pytest_root_dir()
pytest_args = fabtests_args_to_pytest_args(fabtests_args, shared_options, run_mode)
pytest_args.append(get_pytest_relative_case_dir(fabtests_args, pytest_root_dir))
pytest_command = "cd " + pytest_root_dir + "; pytest"
for arg in pytest_args:
if arg.find(' ') != -1:
arg = "'" + arg + "'"
pytest_command += " " + arg
print(pytest_command)
# actually running tests
os.chdir(pytest_root_dir)
status = pytest.main(pytest_args)
os.chdir(prev_cwd)
return status
def main():
pytest_root_dir = get_pytest_root_dir()
# pytest/options.yaml contains the definition of a list of options that are
# shared between runfabtests.py and pytest
option_yaml = os.path.join(pytest_root_dir, "options.yaml")
if not os.path.exists(option_yaml):
print("Error: option definition yaml file {} not found!".format(option_yaml))
exit(1)
shared_options = yaml.safe_load(open(option_yaml))
parser = argparse.ArgumentParser(description="libfabric integration test runner")
parser.add_argument("provider", type=str, help="libfabric provider")
parser.add_argument("server_id", type=str, help="server ip or hostname")
parser.add_argument("client_id", type=str, help="client ip or hostname")
parser.add_argument("-t", dest="testsets", type=str, default="quick",
help="test set(s): all,quick,unit,functional,standard,short,ubertest,cuda_memory,neuron_memory (default quick)")
parser.add_argument("-v", dest="verbose", action="count", default=0,
help="verbosity level"
"-v: print extra info for failed test(s)"
"-vv: print extra info of failed/skipped test(s)"
"-vvv: print extra info of failed/skipped/passed test(s)")
parser.add_argument("--expression", type=str,
help="only run tests which match the given substring expression.")
parser.add_argument("--html", type=str, help="path to generated html report")
parser.add_argument("--junit-xml", type=str, help="path to generated junit xml report")
parser.add_argument("--junit-logging", choices=['no', 'log', 'system-out', 'system-err', 'out-err', 'all'], type=str,
help="Write captured log messages to JUnit report")
parser.add_argument("--nworkers", type=int, default=8, help="Number of parallel test workers. Defaut is 8.")
add_common_arguments(parser, shared_options)
fabtests_args = parser.parse_args()
if fabtests_args.provider not in ["efa", "shm"] and fabtests_args.nworkers > 1:
print("only efa and shm provider support parallelized tests. Setting nworkers to 1 ....")
fabtests_args.nworkers = 1
if fabtests_args.html:
print("html cannot be generated under parallel mode. Setting nworkers to 1 ....")
fabtests_args.nworkers = 1
if fabtests_args.nworkers == 1:
exit(run(fabtests_args, shared_options, None))
else:
print("Running parallelable tests in parallel mode")
parallel_status = run(fabtests_args, shared_options, "parallel")
print("Running other tests in serial mode")
serial_status = run(fabtests_args, shared_options, "serial")
if fabtests_args.junit_xml:
merged_xml = JUnitXml.fromfile(f'{fabtests_args.junit_xml}.parallel') + JUnitXml.fromfile(f'{fabtests_args.junit_xml}.serial')
merged_xml.write(f'{fabtests_args.junit_xml}')
os.unlink(fabtests_args.junit_xml + ".parallel")
os.unlink(fabtests_args.junit_xml + ".serial")
# Still return success when no tests are collected.
if parallel_status not in [ExitCode.OK, ExitCode.NO_TESTS_COLLECTED]:
exit(parallel_status)
if serial_status not in [ExitCode.OK, ExitCode.NO_TESTS_COLLECTED]:
exit(serial_status)
exit(0)
main()