#!/usr/bin/env python3 import sys import shutil import os import fnmatch import unittest import time import threading import traceback import signal import re from multiprocessing import Process, Pipe, get_context from multiprocessing.queues import Queue from multiprocessing.managers import BaseManager from config import config, num_cpus, available_cpus, max_vpp_cpus from vpp_papi import VPPApiJSONFiles from asfframework import ( VppTestRunner, get_testcase_doc_name, get_test_description, get_failed_testcase_linkname, get_testcase_dirname, ) from framework import VppTestCase from test_result_code import TestResultCode from debug import spawn_gdb from log import ( get_parallel_logger, double_line_delim, RED, YELLOW, GREEN, colorize, single_line_delim, ) from discover_tests import discover_tests import sanity_run_vpp from subprocess import check_output, CalledProcessError from util import check_core_path, get_core_path, is_core_present # timeout which controls how long the child has to finish after seeing # a core dump in test temporary directory. If this is exceeded, parent assumes # that child process is stuck (e.g. waiting for event from vpp) and kill # the child core_timeout = 3 class StreamQueue(Queue): def write(self, msg): self.put(msg) def flush(self): sys.__stdout__.flush() sys.__stderr__.flush() def fileno(self): return self._writer.fileno() class StreamQueueManager(BaseManager): pass StreamQueueManager.register("StreamQueue", StreamQueue) class TestResult(dict): def __init__(self, testcase_suite, testcases_by_id=None): super(TestResult, self).__init__() for trc in list(TestResultCode): self[trc] = [] self.crashed = False self.testcase_suite = testcase_suite self.testcases = [testcase for testcase in testcase_suite] self.testcases_by_id = testcases_by_id def was_successful(self): return ( 0 == len(self[TestResultCode.FAIL]) == len(self[TestResultCode.ERROR]) == len(self[TestResultCode.UNEXPECTED_PASS]) and len(self[TestResultCode.PASS]) + len(self[TestResultCode.SKIP]) + len(self[TestResultCode.SKIP_CPU_SHORTAGE]) + len(self[TestResultCode.EXPECTED_FAIL]) == self.testcase_suite.countTestCases() ) def no_tests_run(self): return 0 == len(self[TestResultCode.TEST_RUN]) def process_result(self, test_id, result): self[result].append(test_id) def suite_from_failed(self): rerun_ids = set([]) for testcase in self.testcase_suite: tc_id = testcase.id() if ( tc_id not in self[TestResultCode.PASS] + self[TestResultCode.SKIP] + self[TestResultCode.SKIP_CPU_SHORTAGE] + self[TestResultCode.EXPECTED_FAIL] ): rerun_ids.add(tc_id) if rerun_ids: return suite_from_failed(self.testcase_suite, rerun_ids) def get_testcase_names(self, test_id): # could be tearDownClass (test_ipsec_esp.TestIpsecEsp1) setup_teardown_match = re.match( r"((tearDownClass)|(setUpClass)) \((.+\..+)\)", test_id ) if setup_teardown_match: test_name, _, _, testcase_name = setup_teardown_match.groups() if len(testcase_name.split(".")) == 2: for key in self.testcases_by_id.keys(): if key.startswith(testcase_name): testcase_name = key break testcase_name = self._get_testcase_doc_name(testcase_name) else: test_name = self._get_test_description(test_id) testcase_name = self._get_testcase_doc_name(test_id) return testcase_name, test_name def _get_test_description(self, test_id): if test_id in self.testcases_by_id: desc = get_test_description(descriptions, self.testcases_by_id[test_id]) else: desc = test_id return desc def _get_testcase_doc_name(self, test_id): if test_id in self.testcases_by_id: doc_name = get_testcase_doc_name(self.testcases_by_id[test_id]) else: doc_name = test_id return doc_name def test_runner_wrapper( suite, keep_alive_pipe, stdouterr_queue, finished_pipe, result_pipe, logger ): sys.stdout = stdouterr_queue sys.stderr = stdouterr_queue VppTestCase.parallel_handler = logger.handlers[0] result = VppTestRunner( keep_alive_pipe=keep_alive_pipe, descriptions=descriptions, verbosity=config.verbose, result_pipe=result_pipe, failfast=config.failfast, print_summary=False, ).run(suite) finished_pipe.send(result.wasSuccessful()) finished_pipe.close() keep_alive_pipe.close() class TestCaseWrapper(object): def __init__(self, testcase_suite, manager): self.keep_alive_parent_end, self.keep_alive_child_end = Pipe(duplex=False) self.finished_parent_end, self.finished_child_end = Pipe(duplex=False) self.result_parent_end, self.result_child_end = Pipe(duplex=False) self.testcase_suite = testcase_suite self.stdouterr_queue = manager.StreamQueue(ctx=get_context()) self.logger = get_parallel_logger(self.stdouterr_queue) self.child = Process( target=test_runner_wrapper, args=( testcase_suite, self.keep_alive_child_end, self.stdouterr_queue, self.finished_child_end, self.result_child_end, self.logger, ), ) self.child.start() self.last_test_temp_dir = None self.last_test_vpp_binary = None self._last_test = None self.last_test_id = None self.vpp_pid = None self.last_heard = time.time() self.core_detected_at = None self.testcases_by_id = {} self.testclasess_with_core = {} for testcase in self.testcase_suite: self.testcases_by_id[testcase.id()] = testcase self.result = TestResult(testcase_suite, self.testcases_by_id) @property def last_test(self): return self._last_test @last_test.setter def last_test(self, test_id): self.last_test_id = test_id if test_id in self.testcases_by_id: testcase = self.testcases_by_id[test_id] self._last_test = testcase.shortDescription() if not self._last_test: self._last_test = str(testcase) else: self._last_test = test_id def add_testclass_with_core(self): if self.last_test_id in self.testcases_by_id: test = self.testcases_by_id[self.last_test_id] class_name = unittest.util.strclass(test.__class__) test_name = "'{}' ({})".format( get_test_description(descriptions, test), self.last_test_id ) else: test_name = self.last_test_id class_name = re.match( r"((tearDownClass)|(setUpClass)) " r"\((.+\..+)\)", test_name ).groups()[3] if class_name not in self.testclasess_with_core: self.testclasess_with_core[class_name] = ( test_name, self.last_test_vpp_binary, self.last_test_temp_dir, ) def close_pipes(self): self.keep_alive_child_end.close() self.finished_child_end.close() self.result_child_end.close() self.keep_alive_parent_end.close() self.finished_parent_end.close() self.result_parent_end.close() def was_successful(self): return self.result.was_successful() @property def cpus_used(self): return self.testcase_suite.cpus_used def get_assigned_cpus(self): return self.testcase_suite.get_assigned_cpus() def stdouterr_reader_wrapper( unread_testcases, finished_unread_testcases, read_testcases ): read_testcase = None while read_testcases.is_set() or unread_testcases: if finished_unread_testcases: read_testcase = finished_unread_testcases.pop() unread_testcases.remove(read_testcase) elif unread_testcases: read_testcase = unread_testcases.pop() if read_testcase: data = "" while data is not None: sys.stdout.write(data) data = read_testcase.stdouterr_queue.get() read_testcase.stdouterr_queue.close() finished_unread_testcases.discard(read_testcase) read_testcase = None def handle_failed_suite(logger, last_test_temp_dir, vpp_pid, vpp_binary): if last_test_temp_dir: # Need to create link in case of a timeout or core dump without failure lttd = os.path.basename(last_test_temp_dir) link_path = os.path.join(config.failed_dir, f"{lttd}-FAILED") if not os.path.exists(link_path): os.symlink(last_test_temp_dir, link_path) logger.error( "Symlink to failed testcase directory: %s -> %s" % (link_path, lttd) ) # Report core existence core_path = get_core_path(last_test_temp_dir) if os.path.exists(core_path): logger.error( "Core-file exists in test temporary directory: %s!" % core_path ) check_core_path(logger, core_path) logger.debug("Running 'file %s':" % core_path) try: info = check_output(["file", core_path]) logger.debug(info) except CalledProcessError as e: logger.error( "Subprocess returned with return code " "while running `file' utility on core-file " "returned: " "rc=%s", e.returncode, ) except OSError as e: logger.error( "Subprocess returned with OS error while " "running 'file' utility " "on core-file: " "(%s) %s", e.errno, e.strerror, ) except Exception as e: logger.exception("Unexpected error running `file' utility on core-file") logger.error(f"gdb {vpp_binary} {core_path}") if vpp_pid: # Copy api post mortem api_post_mortem_path = "/tmp/api_post_mortem.%d" % vpp_pid if os.path.isfile(api_post_mortem_path): logger.error( "Copying api_post_mortem.%d to %s" % (vpp_pid, last_test_temp_dir) ) shutil.copy2(api_post_mortem_path, last_test_temp_dir) def check_and_handle_core(vpp_binary, tempdir, core_crash_test): if is_core_present(tempdir): if debug_core: print( "VPP core detected in %s. Last test running was %s" % (tempdir, core_crash_test) ) print(single_line_delim) spawn_gdb(vpp_binary, get_core_path(tempdir)) print(single_line_delim) elif config.compress_core: print("Compressing core-file in test directory `%s'" % tempdir) os.system("gzip %s" % get_core_path(tempdir)) def handle_cores(failed_testcases): for failed_testcase in failed_testcases: tcs_with_core = failed_testcase.testclasess_with_core if tcs_with_core: for test, vpp_binary, tempdir in tcs_with_core.values(): check_and_handle_core(vpp_binary, tempdir, test) def process_finished_testsuite( wrapped_testcase_suite, finished_testcase_suites, failed_wrapped_testcases, results ): results.append(wrapped_testcase_suite.result) finished_testcase_suites.add(wrapped_testcase_suite) stop_run = False if config.failfast and not wrapped_testcase_suite.was_successful(): stop_run = True if not wrapped_testcase_suite.was_successful(): failed_wrapped_testcases.add(wrapped_testcase_suite) handle_failed_suite( wrapped_testcase_suite.logger, wrapped_testcase_suite.last_test_temp_dir, wrapped_testcase_suite.vpp_pid, wrapped_testcase_suite.last_test_vpp_binary, ) return stop_run def run_forked(testcase_suites): wrapped_testcase_suites = set() solo_testcase_suites = [] # suites are unhashable, need to use list results = [] unread_testcases = set() finished_unread_testcases = set() manager = StreamQueueManager() manager.start() tests_running = 0 free_cpus = list(available_cpus) def on_suite_start(tc): nonlocal tests_running nonlocal free_cpus tests_running = tests_running + 1 def on_suite_finish(tc): nonlocal tests_running nonlocal free_cpus tests_running = tests_running - 1 assert tests_running >= 0 free_cpus.extend(tc.get_assigned_cpus()) def run_suite(suite): nonlocal manager nonlocal wrapped_testcase_suites nonlocal unread_testcases nonlocal free_cpus suite.assign_cpus(free_cpus[: suite.cpus_used]) free_cpus = free_cpus[suite.cpus_used :] wrapper = TestCaseWrapper(suite, manager) wrapped_testcase_suites.add(wrapper) unread_testcases.add(wrapper) on_suite_start(suite) def can_run_suite(suite): return tests_running < max_concurrent_tests and ( suite.cpus_used <= len(free_cpus) or suite.cpus_used > max_vpp_cpus ) while free_cpus and testcase_suites: a_suite = testcase_suites[0] if a_suite.is_tagged_run_solo: a_suite = testcase_suites.pop(0) solo_testcase_suites.append(a_suite) continue if can_run_suite(a_suite): a_suite = testcase_suites.pop(0) run_suite(a_suite) else: break if tests_running == 0 and solo_testcase_suites: a_suite = solo_testcase_suites.pop(0) run_suite(a_suite) read_from_testcases = threading.Event() read_from_testcases.set() stdouterr_thread = threading.Thread( target=stdouterr_reader_wrapper, args=(unread_testcases, finished_unread_testcases, read_from_testcases), ) stdouterr_thread.start() failed_wrapped_testcases = set() stop_run = False try: while wrapped_testcase_suites or testcase_suites: finished_testcase_suites = set() for wrapped_testcase_suite in wrapped_testcase_suites: while wrapped_testcase_suite.result_parent_end.poll(): wrapped_testcase_suite.result.process_result( *wrapped_testcase_suite.result_parent_end.recv() ) wrapped_testcase_suite.last_heard = time.time() while wrapped_testcase_suite.keep_alive_parent_end.poll(): ( wrapped_testcase_suite.last_test, wrapped_testcase_suite.last_test_vpp_binary, wrapped_testcase_suite.last_test_temp_dir, wrapped_testcase_suite.vpp_pid, ) = wrapped_testcase_suite.keep_alive_parent_end.recv() wrapped_testcase_suite.last_heard = time.time() if wrapped_testcase_suite.finished_parent_end.poll(): wrapped_testcase_suite.finished_parent_end.recv() wrapped_testcase_suite.last_heard = time.time() stop_run = ( process_finished_testsuite( wrapped_testcase_suite, finished_testcase_suites, failed_wrapped_testcases, results, ) or stop_run ) continue fail = False if wrapped_testcase_suite.last_heard + config.timeout < time.time(): fail = True wrapped_testcase_suite.logger.critical( "Child test runner process timed out " "(last test running was `%s' in `%s')!" % ( wrapped_testcase_suite.last_test, wrapped_testcase_suite.last_test_temp_dir, ) ) elif not wrapped_testcase_suite.child.is_alive(): fail = True wrapped_testcase_suite.logger.critical( "Child test runner process unexpectedly died " "(last test running was `%s' in `%s')!" % ( wrapped_testcase_suite.last_test, wrapped_testcase_suite.last_test_temp_dir, ) ) elif ( wrapped_testcase_suite.last_test_temp_dir and wrapped_testcase_suite.last_test_vpp_binary ): if is_core_present(wrapped_testcase_suite.last_test_temp_dir): wrapped_testcase_suite.add_testclass_with_core() if wrapped_testcase_suite.core_detected_at is None: wrapped_testcase_suite.core_detected_at = time.time() elif ( wrapped_testcase_suite.core_detected_at + core_timeout < time.time() ): wrapped_testcase_suite.logger.critical( "Child test runner process unresponsive and " "core-file exists in test temporary directory " "(last test running was `%s' in `%s')!" % ( wrapped_testcase_suite.last_test, wrapped_testcase_suite.last_test_temp_dir, ) ) fail = True if fail: wrapped_testcase_suite.child.terminate() try: # terminating the child process tends to leave orphan # VPP process around if wrapped_testcase_suite.vpp_pid: os.kill(wrapped_testcase_suite.vpp_pid, signal.SIGTERM) except OSError: # already dead pass wrapped_testcase_suite.result.crashed = True wrapped_testcase_suite.result.process_result( wrapped_testcase_suite.last_test_id, TestResultCode.ERROR ) stop_run = ( process_finished_testsuite( wrapped_testcase_suite, finished_testcase_suites, failed_wrapped_testcases, results, ) or stop_run ) for finished_testcase in finished_testcase_suites: # Somewhat surprisingly, the join below may # timeout, even if client signaled that # it finished - so we note it just in case. join_start = time.time() finished_testcase.child.join(test_finished_join_timeout) join_end = time.time() if join_end - join_start >= test_finished_join_timeout: finished_testcase.logger.error( "Timeout joining finished test: %s (pid %d)" % (finished_testcase.last_test, finished_testcase.child.pid) ) finished_testcase.close_pipes() wrapped_testcase_suites.remove(finished_testcase) finished_unread_testcases.add(finished_testcase) finished_testcase.stdouterr_queue.put(None) on_suite_finish(finished_testcase) if stop_run: while testcase_suites: results.append(TestResult(testcase_suites.pop(0))) elif testcase_suites: a_suite = testcase_suites[0] while a_suite and a_suite.is_tagged_run_solo: testcase_suites.pop(0) solo_testcase_suites.append(a_suite) if testcase_suites: a_suite = testcase_suites[0] else: a_suite = None if a_suite and can_run_suite(a_suite): testcase_suites.pop(0) run_suite(a_suite) if solo_testcase_suites and tests_running == 0: a_suite = solo_testcase_suites.pop(0) run_suite(a_suite) time.sleep(0.1) except Exception: for wrapped_testcase_suite in wrapped_testcase_suites: wrapped_testcase_suite.child.terminate() wrapped_testcase_suite.stdouterr_queue.put(None) raise finally: read_from_testcases.clear() stdouterr_thread.join(config.timeout) manager.shutdown() handle_cores(failed_wrapped_testcases) return results class TestSuiteWrapper(unittest.TestSuite): cpus_used = 0 def __init__(self): return super().__init__() def addTest(self, test): self.cpus_used = max(self.cpus_used, test.get_cpus_required()) super().addTest(test) def assign_cpus(self, cpus): self.cpus = cpus def _handleClassSetUp(self, test, result): if not test.__class__.skipped_due_to_cpu_lack: test.assign_cpus(self.cpus) super()._handleClassSetUp(test, result) def get_assigned_cpus(self): return self.cpus class SplitToSuitesCallback: def __init__(self, filter_callback): self.suites = {} self.suite_name = "default" self.filter_callback = filter_callback self.filtered = TestSuiteWrapper() def __call__(self, file_name, cls, method): test_method = cls(method) if self.filter_callback(file_name, cls.__name__, method): self.suite_name = file_name + cls.__name__ if self.suite_name not in self.suites: self.suites[self.suite_name] = TestSuiteWrapper() self.suites[self.suite_name].is_tagged_run_solo = False self.suites[self.suite_name].addTest(test_method) if test_method.is_tagged_run_solo(): self.suites[self.suite_name].is_tagged_run_solo = True else: self.filtered.addTest(test_method) def parse_test_filter(test_filter): f = test_filter filter_file_name = None filter_class_name = None filter_func_name = None if f: if "." in f: parts = f.split(".") if len(parts) > 3: raise Exception(f"Invalid test filter: {test_filter}") if len(parts) > 2: if parts[2] not in ("*", ""): filter_func_name = parts[2] if parts[1] not in ("*", ""): filter_class_name = parts[1] if parts[0] not in ("*", ""): if parts[0].startswith("test_"): filter_file_name = parts[0] else: filter_file_name = "test_%s" % parts[0] else: if f.startswith("test_"): filter_file_name = f else: filter_file_name = "test_%s" % f if filter_file_name: filter_file_name = "%s.py" % filter_file_name return filter_file_name, filter_class_name, filter_func_name def filter_tests(tests, filter_cb): result = TestSuiteWrapper() for t in tests: if isinstance(t, unittest.suite.TestSuite): # this is a bunch of tests, recursively filter... x = filter_tests(t, filter_cb) if x.countTestCases() > 0: result.addTest(x) elif isinstance(t, unittest.TestCase): # this is a single test parts = t.id().split(".") # t.id() for common cases like this: # test_classifier.TestClassifier.test_acl_ip # apply filtering only if it is so if len(parts) == 3: if not filter_cb(parts[0], parts[1], parts[2]): continue result.addTest(t) else: # unexpected object, don't touch it result.addTest(t) return result class FilterByTestOption: def __init__(self, filters): self.filters = filters def __call__(self, file_name, class_name, func_name): def test_one( filter_file_name, filter_class_name, filter_func_name, file_name, class_name, func_name, ): if filter_file_name: fn_match = fnmatch.fnmatch(file_name, filter_file_name) if not fn_match: return False if filter_class_name and class_name != filter_class_name: return False if filter_func_name and func_name != filter_func_name: return False return True for filter_file_name, filter_class_name, filter_func_name in self.filters: if test_one( filter_file_name, filter_class_name, filter_func_name, file_name, class_name, func_name, ): return True return False class FilterByClassList: def __init__(self, classes_with_filenames): self.classes_with_filenames = classes_with_filenames def __call__(self, file_name, class_name, func_name): return ".".join([file_name, class_name]) in self.classes_with_filenames def suite_from_failed(suite, failed): failed = {x.rsplit(".", 1)[0] for x in failed} filter_cb = FilterByClassList(failed) suite = filter_tests(suite, filter_cb) return suite class AllResults(dict): def __init__(self): super(AllResults, self).__init__() self.all_testcases = 0 self.results_per_suite = [] for trc in list(TestResultCode): self[trc] = 0 self.rerun = [] self.testsuites_no_tests_run = [] def add_results(self, result): self.results_per_suite.append(result) for trc in list(TestResultCode): self[trc] += len(result[trc]) def add_result(self, result): retval = 0 self.all_testcases += result.testcase_suite.countTestCases() self.add_results(result) if result.no_tests_run(): self.testsuites_no_tests_run.append(result.testcase_suite) if result.crashed: retval = -1 else: retval = 1 elif not result.was_successful(): retval = 1 if retval != 0: self.rerun.append(result.testcase_suite) return retval def print_results(self): print("") print(double_line_delim) print("TEST RESULTS:") def indent_results(lines): lines = list(filter(None, lines)) maximum = max(lines, key=lambda x: x.index(":")) maximum = 4 + maximum.index(":") for l in lines: padding = " " * (maximum - l.index(":")) print(f"{padding}{l}") indent_results( [ f"Scheduled tests: {self.all_testcases}", f"Executed tests: {self[TestResultCode.TEST_RUN]}", f"Passed tests: {colorize(self[TestResultCode.PASS], GREEN)}", ( f"Expected failures: {colorize(self[TestResultCode.EXPECTED_FAIL], GREEN)}" if self[TestResultCode.EXPECTED_FAIL] else None ), ( f"Skipped tests: {colorize(self[TestResultCode.SKIP], YELLOW)}" if self[TestResultCode.SKIP] else None ), ( f"Not Executed tests: {colorize(self.not_executed, RED)}" if self.not_executed else None ), ( f"Failures: {colorize(self[TestResultCode.FAIL], RED)}" if self[TestResultCode.FAIL] else None ), ( f"Unexpected passes: {colorize(self[TestResultCode.UNEXPECTED_PASS], RED)}" if self[TestResultCode.UNEXPECTED_PASS] else None ), ( f"Errors: {colorize(self[TestResultCode.ERROR], RED)}" if self[TestResultCode.ERROR] else None ), ( "Tests skipped due to lack of CPUS: " f"{colorize(self[TestResultCode.SKIP_CPU_SHORTAGE], YELLOW)}" if self[TestResultCode.SKIP_CPU_SHORTAGE] else None ), ] ) if self.all_failed > 0: print("FAILURES AND ERRORS IN TESTS:") for result in self.results_per_suite: old_testcase_name = None for tr_code, headline in ( (TestResultCode.FAIL, "FAILURE"), (TestResultCode.ERROR, "ERROR"), (TestResultCode.UNEXPECTED_PASS, "UNEXPECTED PASS"), ): if not result[tr_code]: continue for failed_test_id in result[tr_code]: new_testcase_name, test_name = result.get_testcase_names( failed_test_id ) if new_testcase_name != old_testcase_name: print( f" Testcase name: {colorize(new_testcase_name, RED)}" ) old_testcase_name = new_testcase_name print( f" {headline}: {colorize(test_name, RED)} [{failed_test_id}]" ) if self.testsuites_no_tests_run: print("TESTCASES WHERE NO TESTS WERE SUCCESSFULLY EXECUTED:") tc_classes = set() for testsuite in self.testsuites_no_tests_run: for testcase in testsuite: tc_classes.add(get_testcase_doc_name(testcase)) for tc_class in tc_classes: print(" {}".format(colorize(tc_class, RED))) if self[TestResultCode.SKIP_CPU_SHORTAGE]: print() print( colorize( " SOME TESTS WERE SKIPPED BECAUSE THERE ARE NOT" " ENOUGH CPUS AVAILABLE", YELLOW, ) ) print(double_line_delim) print("") @property def not_executed(self): return self.all_testcases - self[TestResultCode.TEST_RUN] @property def all_failed(self): return ( self[TestResultCode.FAIL] + self[TestResultCode.ERROR] + self[TestResultCode.UNEXPECTED_PASS] ) def parse_results(results): """ Prints the number of scheduled, executed, not executed, passed, failed, errored and skipped tests and details about failed and errored tests. Also returns all suites where any test failed. :param results: :return: """ results_per_suite = AllResults() crashed = False failed = False for result in results: result_code = results_per_suite.add_result(result) if result_code == 1: failed = True elif result_code == -1: crashed = True results_per_suite.print_results() if crashed: return_code = -1 elif failed: return_code = 1 else: return_code = 0 return return_code, results_per_suite.rerun if __name__ == "__main__": print(f"Config is: {config}") if config.api_preload: VPPApiJSONFiles.load_api(apidir=config.extern_apidir + [config.vpp_install_dir]) if config.sanity: print("Running sanity test case.") try: rc = sanity_run_vpp.main() if rc != 0: sys.exit(rc) except Exception as e: print(traceback.format_exc()) print("Couldn't run sanity test case.") sys.exit(-1) test_finished_join_timeout = 15 debug_gdb = config.debug in ["gdb", "gdbserver", "attach"] debug_core = config.debug == "core" run_interactive = debug_gdb or config.step or config.force_foreground max_concurrent_tests = 0 print(f"OS reports {num_cpus} available cpu(s).") test_jobs = config.jobs if test_jobs == "auto": if run_interactive: max_concurrent_tests = 1 print("Interactive mode required, running tests consecutively.") else: max_concurrent_tests = num_cpus print( f"Running at most {max_concurrent_tests} python test " "processes concurrently." ) else: max_concurrent_tests = test_jobs print( f"Running at most {max_concurrent_tests} python test processes " "concurrently as set by 'TEST_JOBS'." ) print(f"Using at most {max_vpp_cpus} cpus for VPP threads.") if run_interactive and max_concurrent_tests > 1: raise NotImplementedError( "Running tests interactively (DEBUG is gdb[server] or ATTACH or " "STEP is set) in parallel (TEST_JOBS is more than 1) is not " "supported" ) descriptions = True print("Running tests using custom test runner.") filters = [(parse_test_filter(f)) for f in config.filter.split(",")] print( "Selected filters: ", "|".join( f"file={filter_file}, class={filter_class}, function={filter_func}" for filter_file, filter_class, filter_func in filters ), ) filter_cb = FilterByTestOption(filters) cb = SplitToSuitesCallback(filter_cb) for d in config.test_src_dir: print("Adding tests from directory tree %s" % d) discover_tests(d, cb) # suites are not hashable, need to use list suites = [] tests_amount = 0 for testcase_suite in cb.suites.values(): tests_amount += testcase_suite.countTestCases() if testcase_suite.cpus_used > max_vpp_cpus: # here we replace test functions with lambdas to just skip them # but we also replace setUp/tearDown functions to do nothing # so that the test can be "started" and "stopped", so that we can # still keep those prints (test description - SKIP), which are done # in stopTest() (for that to trigger, test function must run) for t in testcase_suite: for m in dir(t): if m.startswith("test_"): setattr(t, m, lambda: t.skipTest("not enough cpus")) setattr(t.__class__, "setUpClass", lambda: None) setattr(t.__class__, "tearDownClass", lambda: None) setattr(t, "setUp", lambda: None) setattr(t, "tearDown", lambda: None) t.__class__.skipped_due_to_cpu_lack = True suites.append(testcase_suite) print( "%s out of %s tests match specified filters" % (tests_amount, tests_amount + cb.filtered.countTestCases()) ) if not config.extended: print("Not running extended tests (some tests will be skipped)") attempts = config.retries + 1 if attempts > 1: print("Perform %s attempts to pass the suite..." % attempts) if run_interactive and suites: # don't fork if requiring interactive terminal print("Running tests in foreground in the current process") full_suite = unittest.TestSuite() free_cpus = list(available_cpus) cpu_shortage = False for suite in suites: if suite.cpus_used <= max_vpp_cpus: suite.assign_cpus(free_cpus[: suite.cpus_used]) else: suite.assign_cpus([]) cpu_shortage = True full_suite.addTests(suites) result = VppTestRunner( verbosity=config.verbose, failfast=config.failfast, print_summary=True ).run(full_suite) was_successful = result.wasSuccessful() if not was_successful: for test_case_info in result.failed_test_cases_info: handle_failed_suite( test_case_info.logger, test_case_info.tempdir, test_case_info.vpp_pid, config.vpp, ) if test_case_info in result.core_crash_test_cases_info: check_and_handle_core( test_case_info.vpp_bin_path, test_case_info.tempdir, test_case_info.core_crash_test, ) if cpu_shortage: print() print( colorize( "SOME TESTS WERE SKIPPED BECAUSE THERE ARE NOT" " ENOUGH CPUS AVAILABLE", YELLOW, ) ) print() sys.exit(not was_successful) else: print( "Running each VPPTestCase in a separate background process" f" with at most {max_concurrent_tests} parallel python test " "process(es)" ) exit_code = 0 while suites and attempts > 0: for suite in suites: failed_link = get_failed_testcase_linkname( config.failed_dir, f"{get_testcase_dirname(suite._tests[0].__class__.__name__)}", ) if os.path.islink(failed_link): os.unlink(failed_link) results = run_forked(suites) exit_code, suites = parse_results(results) attempts -= 1 if exit_code == 0: print("Test run was successful") else: print("%s attempt(s) left." % attempts) sys.exit(exit_code)