src/tools

#!/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)
Age	Commit message (Expand)	Author	Files	Lines
2020-10-13	stats: counters data model	Ole Troan	3	-12/+162
2020-10-06	feature: Add packet trace API	Jon Loeliger	1	-4/+10
2020-09-18	vat: add infrastructure to align vnet test code and plugin test code	Ole Troan	1	-10/+11
2020-09-11	build: fix build for Debian 9 and Debian 10	Benoît Ganne	1	-4/+5
2020-09-01	vppapigen: supports backwards compatible marking of enums	Ole Troan	1	-18/+38
2020-08-25	api: register endian handlers for reply messages	Ole Troan	1	-0/+13
2020-08-21	vppapigen: Fix missing api dir	Nathan Skrzypczak	1	-0/+1
2020-08-19	vppinfra: minor tweaks for cgo interoperation	Dave Barach	2	-11/+11
2020-08-17	vppapigen: generate define for module_semver	Ole Troan	1	-0/+7
2020-08-13	vppapigen: remove unreachable code	Ole Troan	1	-1/+0
2020-08-11	vppapigen: add support for empty options	Ole Troan	2	-3/+17
2020-08-06	misc: harmonize names	Dave Barach	2	-10/+10
2020-07-23	vppapigen: missing crcs in user-defined types	Ole Troan	2	-3/+403
2020-06-17	docs: add more FEATURE.yaml documentation	Dave Barach	1	-0/+12
2020-05-27	api: make vpp api handlers endian independent	Ole Troan	1	-4/+15
2020-05-25	api: add new stream message convention	Ole Troan	2	-1/+8
2020-05-19	vppapigen: handle new api file for crcchekcer	Ole Troan	1	-2/+9
2020-05-16	vppapigen: add assert for python version	Ole Troan	1	-2/+4
2020-05-15	g2: fix the g2 build for Ubuntu 20.04	Dave Barach	1	-2/+1
2020-05-09	vppapigen: api crc checker	Ole Troan	2	-23/+67
2020-05-08	build: various improvements	Damjan Marion	1	-1/+1
2020-05-07	misc: deprecate elftool	Damjan Marion	3	-500/+0
2020-04-06	misc: fix python sonarcloud BLOCKER level issues	Paul Vinciguerra	1	-1/+1
2020-04-04	misc: strcpy be gone	Dave Barach	2	-22/+20
2020-04-04	misc: sprintf be gone	Dave Barach	7	-164/+163
2020-03-04	vppapigen: crc is a negative value for some messages when using python 2.7	Mark Nelson	1	-2/+2
2020-02-26	vppapigen: list imports in JSON	Ondrej Fabry	1	-0/+7
2020-02-26	api: improve api string safety	Jakub Grajciar	1	-3/+2
2020-02-04	vppapigen: fix options representation	Ole Troan	1	-1/+1
2020-01-30	vppapigen: update markdown documentation	Paul Vinciguerra	1	-7/+10
2019-12-10	api: multiple connections per process	Dave Barach	1	-1/+1
2019-12-07	lacp: fix control_ping from plugins	Ole Troan	1	-0/+4
2019-11-24	vppapigen: clean up typos in c generated code	Paul Vinciguerra	1	-4/+4
2019-11-24	vppapigen: fix i64 format string	Paul Vinciguerra	1	-1/+1
2019-11-24	vppapigen: fix typo on f64 endian_string	Paul Vinciguerra	1	-4/+4
2019-10-10	acl: remove api boilerplate	Ole Troan	1	-0/+3
2019-10-10	memif: remove api boilerplate	Ole Troan	1	-0/+2
2019-10-09	api: autogenerate event handler functions for *_test.c	Ole Troan	1	-0/+17
2019-09-25	api: split api generated files	Ole Troan	3	-74/+261
2019-09-19	api: split vl_api_prefix into two	Ole Troan	3	-47/+39
2019-09-18	vppapigen: fix missing vla check for union class	Ole Troan	2	-52/+87
2019-09-18	vppapigen: fix tests and run on verify	Paul Vinciguerra	2	-26/+56
2019-09-16	api: autogenerate api trace print/endian	Ole Troan	4	-86/+434
2019-09-10	misc: fix shebang with missing env	Nathan Skrzypczak	1	-1/+1
2019-09-03	api: enforce vla is last and fixed string type	Ole Troan	2	-11/+74
2019-08-23	vppapitrace: add text output	Ole Troan	1	-8/+65
2019-08-20	vppapigen: remove support for legacy typedefs	Paul Vinciguerra	1	-1/+3
2019-08-20	vppapigen: remove python2 support	Paul Vinciguerra	1	-28/+30
2019-08-19	vppapigen map: raise ValueError when fieldname is python keyword	Paul Vinciguerra	1	-0/+4
2019-08-08	api: vppapitrace JSON/API trace converter	Ole Troan	2	-0/+435