linux/tools/testing/selftests/net/lib/py/ksft.py

# SPDX-License-Identifier: GPL-2.0

import fnmatch
import functools
import getopt
import inspect
import os
import signal
import sys
import time
import traceback
from collections import namedtuple
from .consts import KSFT_MAIN_NAME
from . import utils

KSFT_RESULT = None
KSFT_RESULT_ALL = True
KSFT_DISRUPTIVE = True


class KsftFailEx(Exception):
    pass


class KsftSkipEx(Exception):
    pass


class KsftXfailEx(Exception):
    pass


class KsftTerminate(KeyboardInterrupt):
    pass


class _KsftArgs:
    def __init__(self):
        self.list_tests = False
        self.filters = []

        try:
            opts, _ = getopt.getopt(sys.argv[1:], 'hlt:T:')
        except getopt.GetoptError as e:
            print(e, file=sys.stderr)
            sys.exit(1)

        for opt, val in opts:
            if opt == '-h':
                print(f"Usage: {sys.argv[0]} [-h|-l] [-t|-T name]\n"
                      f"\t-h       print help\n"
                      f"\t-l       list tests (filtered, if filters were specified)\n"
                      f"\t-t name  include test\n"
                      f"\t-T name  exclude test",
                      file=sys.stderr)
                sys.exit(0)
            elif opt == '-l':
                self.list_tests = True
            elif opt == '-t':
                self.filters.append((True, val))
            elif opt == '-T':
                self.filters.append((False, val))


@functools.lru_cache()
def _ksft_supports_color():
    if os.environ.get("NO_COLOR") is not None:
        return False
    if not hasattr(sys.stdout, "isatty") or not sys.stdout.isatty():
        return False
    if os.environ.get("TERM") == "dumb":
        return False
    return True


def ksft_pr(*objs, **kwargs):
    """
    Print logs to stdout.

    Behaves like print() but log lines will be prefixed
    with # to prevent breaking the TAP output formatting.

    Extra arguments (on top of what print() supports):
      line_pfx - add extra string before each line
    """
    sep = kwargs.pop("sep", " ")
    pfx = kwargs.pop("line_pfx", "")
    pfx = "#" + (" " + pfx if pfx else "")
    kwargs["flush"] = True

    text = sep.join(str(obj) for obj in objs)
    prefixed = f"\n{pfx} ".join(text.split('\n'))
    print(pfx, prefixed, **kwargs)


def _fail(*args):
    global KSFT_RESULT
    KSFT_RESULT = False

    stack = inspect.stack()
    started = False
    for frame in reversed(stack[2:]):
        # Start printing from the test case function
        if not started:
            if frame.function == 'ksft_run':
                started = True
            continue

        ksft_pr("Check| At " + frame.filename + ", line " + str(frame.lineno) +
                ", in " + frame.function + ":")
        ksft_pr("Check|     " + frame.code_context[0].strip())
    ksft_pr(*args)


def ksft_eq(a, b, comment=""):
    global KSFT_RESULT
    if a != b:
        _fail("Check failed", a, "!=", b, comment)


def ksft_ne(a, b, comment=""):
    global KSFT_RESULT
    if a == b:
        _fail("Check failed", a, "==", b, comment)


def ksft_true(a, comment=""):
    if not a:
        _fail("Check failed", a, "does not eval to True", comment)


def ksft_not_none(a, comment=""):
    if a is None:
        _fail("Check failed", a, "is None", comment)


def ksft_in(a, b, comment=""):
    if a not in b:
        _fail("Check failed", a, "not in", b, comment)


def ksft_not_in(a, b, comment=""):
    if a in b:
        _fail("Check failed", a, "in", b, comment)


def ksft_is(a, b, comment=""):
    if a is not b:
        _fail("Check failed", a, "is not", b, comment)


def ksft_ge(a, b, comment=""):
    if a < b:
        _fail("Check failed", a, "<", b, comment)


def ksft_gt(a, b, comment=""):
    if a <= b:
        _fail("Check failed", a, "<=", b, comment)


def ksft_lt(a, b, comment=""):
    if a >= b:
        _fail("Check failed", a, ">=", b, comment)


class ksft_raises:
    def __init__(self, expected_type):
        self.exception = None
        self.expected_type = expected_type

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is None:
            _fail(f"Expected exception {str(self.expected_type.__name__)}, none raised")
        elif self.expected_type != exc_type:
            _fail(f"Expected exception {str(self.expected_type.__name__)}, raised {str(exc_type.__name__)}")
        self.exception = exc_val
        # Suppress the exception if its the expected one
        return self.expected_type == exc_type


def ksft_busy_wait(cond, sleep=0.005, deadline=1, comment=""):
    end = time.monotonic() + deadline
    while True:
        if cond():
            return
        if time.monotonic() > end:
            _fail("Waiting for condition timed out", comment)
            return
        time.sleep(sleep)


def ktap_result(ok, cnt=1, case_name="", comment=""):
    global KSFT_RESULT_ALL
    KSFT_RESULT_ALL = KSFT_RESULT_ALL and ok

    res = ""
    if not ok:
        res += "not "
    res += "ok "
    res += str(cnt) + " "
    res += KSFT_MAIN_NAME
    if case_name:
        res += "." + case_name
    if comment:
        res += " # " + comment
    if _ksft_supports_color():
        if comment.startswith(("SKIP", "XFAIL")):
            color = "\033[33m"
        elif ok:
            color = "\033[32m"
        else:
            color = "\033[31m"
        res = color + res + "\033[0m"
    print(res, flush=True)


def _ksft_defer_arm(state):
    """ Allow or disallow the use of defer() """
    utils.GLOBAL_DEFER_ARMED = state


def ksft_flush_defer():
    global KSFT_RESULT

    i = 0
    qlen_start = len(utils.GLOBAL_DEFER_QUEUE)
    while utils.GLOBAL_DEFER_QUEUE:
        i += 1
        entry = utils.GLOBAL_DEFER_QUEUE.pop()
        try:
            entry.exec_only()
        except Exception:
            ksft_pr(f"Exception while handling defer / cleanup (callback {i} of {qlen_start})!")
            ksft_pr(traceback.format_exc(), line_pfx="Defer Exception|")
            KSFT_RESULT = False


KsftCaseFunction = namedtuple("KsftCaseFunction",
                              ['name', 'original_func', 'variants'])


def ksft_disruptive(func):
    """
    Decorator that marks the test as disruptive (e.g. the test
    that can down the interface). Disruptive tests can be skipped
    by passing DISRUPTIVE=False environment variable.
    """

    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        if not KSFT_DISRUPTIVE:
            raise KsftSkipEx("marked as disruptive")
        return func(*args, **kwargs)
    return wrapper


class KsftNamedVariant:
    """ Named string name + argument list tuple for @ksft_variants """

    def __init__(self, name, *params):
        self.params = params
        self.name = name or "_".join([str(x) for x in self.params])


def ksft_variants(params):
    """
    Decorator defining the sets of inputs for a test.
    The parameters will be included in the name of the resulting sub-case.
    Parameters can be either single object, tuple or a KsftNamedVariant.
    The argument can be a list or a generator.

    Example:

    @ksft_variants([
        (1, "a"),
        (2, "b"),
        KsftNamedVariant("three", 3, "c"),
    ])
    def my_case(cfg, a, b):
        pass # ...

    ksft_run(cases=[my_case], args=(cfg, ))

    Will generate cases:
        my_case.1_a
        my_case.2_b
        my_case.three
    """

    return lambda func: KsftCaseFunction(func.__name__, func, params)


def ksft_setup(env):
    """
    Setup test framework global state from the environment.
    """

    def get_bool(env, name):
        value = env.get(name, "").lower()
        if value in ["yes", "true"]:
            return True
        if value in ["no", "false"]:
            return False
        try:
            return bool(int(value))
        except Exception:
            raise Exception(f"failed to parse {name}")

    if "DISRUPTIVE" in env:
        global KSFT_DISRUPTIVE
        KSFT_DISRUPTIVE = get_bool(env, "DISRUPTIVE")

    return env


def _ksft_intr(signum, frame):
    # ksft runner.sh sends 2 SIGTERMs in a row on a timeout
    # if we don't ignore the second one it will stop us from handling cleanup
    global term_cnt
    term_cnt += 1
    if term_cnt == 1:
        raise KsftTerminate()
    else:
        ksft_pr(f"Ignoring SIGTERM (cnt: {term_cnt}), already exiting...")


def _ksft_name_matches(name, pattern):
    if '*' in pattern or '?' in pattern or '[' in pattern:
        return fnmatch.fnmatchcase(name, pattern)
    return name == pattern


def _ksft_test_enabled(name, filters):
    has_positive = False
    for include, pattern in filters:
        has_positive |= include
        if _ksft_name_matches(name, pattern):
            return include
    return not has_positive


def _ksft_generate_test_cases(cases, globs, case_pfx, args, cli_args):
    """Generate a filtered list of (func, args, name) tuples.

    If -l is given, prints matching test names and exits.
    """

    cases = cases or []
    test_cases = []

    # If using the globs method find all relevant functions
    if globs and case_pfx:
        for key, value in globs.items():
            if not callable(value):
                continue
            for prefix in case_pfx:
                if key.startswith(prefix):
                    cases.append(value)
                    break

    for func in cases:
        if isinstance(func, KsftCaseFunction):
            # Parametrized test - create case for each param
            for param in func.variants:
                if not isinstance(param, KsftNamedVariant):
                    if not isinstance(param, tuple):
                        param = (param, )
                    param = KsftNamedVariant(None, *param)

                test_cases.append((func.original_func,
                                   (*args, *param.params),
                                   func.name + "." + param.name))
        else:
            test_cases.append((func, args, func.__name__))

    if cli_args.filters:
        test_cases = [tc for tc in test_cases
                      if _ksft_test_enabled(tc[2], cli_args.filters)]

    if cli_args.list_tests:
        for _, _, name in test_cases:
            print(name)
        sys.exit(0)

    return test_cases


def ksft_run(cases=None, globs=None, case_pfx=None, args=()):
    cli_args = _KsftArgs()
    test_cases = _ksft_generate_test_cases(cases, globs, case_pfx, args,
                                           cli_args)

    global term_cnt
    term_cnt = 0
    prev_sigterm = signal.signal(signal.SIGTERM, _ksft_intr)

    totals = {"pass": 0, "fail": 0, "skip": 0, "xfail": 0}

    global KSFT_RESULT
    if KSFT_RESULT is not None:
        raise RuntimeError("ksft_run() can't be called multiple times.")

    print("TAP version 13", flush=True)
    print("1.." + str(len(test_cases)), flush=True)

    cnt = 0
    stop = False
    for func, args, name in test_cases:
        KSFT_RESULT = True
        cnt += 1
        comment = ""
        cnt_key = ""

        _ksft_defer_arm(True)
        try:
            func(*args)
        except KsftSkipEx as e:
            comment = "SKIP " + str(e)
            cnt_key = 'skip'
        except KsftXfailEx as e:
            comment = "XFAIL " + str(e)
            cnt_key = 'xfail'
        except BaseException as e:
            stop |= isinstance(e, KeyboardInterrupt)
            ksft_pr(traceback.format_exc(), line_pfx="Exception|")
            if stop:
                ksft_pr(f"Stopping tests due to {type(e).__name__}.")
            KSFT_RESULT = False
            cnt_key = 'fail'
        _ksft_defer_arm(False)

        try:
            ksft_flush_defer()
        except BaseException as e:
            ksft_pr(traceback.format_exc(), line_pfx="Exception|")
            if isinstance(e, KeyboardInterrupt):
                ksft_pr()
                ksft_pr("WARN: defer() interrupted, cleanup may be incomplete.")
                ksft_pr("      Attempting to finish cleanup before exiting.")
                ksft_pr("      Interrupt again to exit immediately.")
                ksft_pr()
                stop = True
            # Flush was interrupted, try to finish the job best we can
            ksft_flush_defer()

        if not cnt_key:
            cnt_key = 'pass' if KSFT_RESULT else 'fail'

        ktap_result(KSFT_RESULT, cnt, name, comment=comment)
        totals[cnt_key] += 1

        if stop:
            break

    signal.signal(signal.SIGTERM, prev_sigterm)

    print(
        f"# Totals: pass:{totals['pass']} fail:{totals['fail']} xfail:{totals['xfail']} xpass:0 skip:{totals['skip']} error:0"
    )


def ksft_exit():
    global KSFT_RESULT_ALL
    sys.exit(0 if KSFT_RESULT_ALL else 1)