frr/tests/topotests/conftest.py

"""
Topotest conftest.py file.
"""

import glob
import os
import pdb
import re
import subprocess
import sys
import time
import resource

import pytest
import lib.fixtures
from lib import topolog
from lib.micronet import Commander, proc_error
from lib.micronet_cli import cli
from lib.micronet_compat import Mininet, cleanup_current, cleanup_previous
from lib.topogen import diagnose_env, get_topogen
from lib.topolog import logger
from lib.topotest import g_extra_config as topotest_extra_config
from lib.topotest import json_cmp_result


def pytest_addoption(parser):
    """
    Add topology-only option to the topology tester. This option makes pytest
    only run the setup_module() to setup the topology without running any tests.
    """
    parser.addoption(
        "--asan-abort",
        action="store_true",
        help="Configure address sanitizer to abort process on error",
    )

    parser.addoption(
        "--cli-on-error",
        action="store_true",
        help="Mininet cli on test failure",
    )

    parser.addoption(
        "--gdb-breakpoints",
        metavar="SYMBOL[,SYMBOL...]",
        help="Comma-separated list of functions to set gdb breakpoints on",
    )

    parser.addoption(
        "--gdb-daemons",
        metavar="DAEMON[,DAEMON...]",
        help="Comma-separated list of daemons to spawn gdb on, or 'all'",
    )

    parser.addoption(
        "--gdb-routers",
        metavar="ROUTER[,ROUTER...]",
        help="Comma-separated list of routers to spawn gdb on, or 'all'",
    )

    parser.addoption(
        "--pause",
        action="store_true",
        help="Pause after each test",
    )

    parser.addoption(
        "--pause-on-error",
        action="store_true",
        help="Do not pause after (disables default when --shell or -vtysh given)",
    )

    parser.addoption(
        "--no-pause-on-error",
        dest="pause_on_error",
        action="store_false",
        help="Do not pause after (disables default when --shell or -vtysh given)",
    )

    rundir_help = "directory for running in and log files"
    parser.addini("rundir", rundir_help, default="/tmp/topotests")
    parser.addoption("--rundir", metavar="DIR", help=rundir_help)

    parser.addoption(
        "--shell",
        metavar="ROUTER[,ROUTER...]",
        help="Comma-separated list of routers to spawn shell on, or 'all'",
    )

    parser.addoption(
        "--shell-on-error",
        action="store_true",
        help="Spawn shell on all routers on test failure",
    )

    parser.addoption(
        "--strace-daemons",
        metavar="DAEMON[,DAEMON...]",
        help="Comma-separated list of daemons to strace, or 'all'",
    )

    parser.addoption(
        "--topology-only",
        action="store_true",
        default=False,
        help="Only set up this topology, don't run tests",
    )

    parser.addoption(
        "--valgrind-extra",
        action="store_true",
        help="Generate suppression file, and enable more precise (slower) valgrind checks",
    )

    parser.addoption(
        "--valgrind-memleaks",
        action="store_true",
        help="Run all daemons under valgrind for memleak detection",
    )

    parser.addoption(
        "--vtysh",
        metavar="ROUTER[,ROUTER...]",
        help="Comma-separated list of routers to spawn vtysh on, or 'all'",
    )

    parser.addoption(
        "--vtysh-on-error",
        action="store_true",
        help="Spawn vtysh on all routers on test failure",
    )


def check_for_memleaks():
    assert topotest_extra_config["valgrind_memleaks"]

    leaks = []
    tgen = get_topogen()
    latest = []
    existing = []
    if tgen is not None:
        logdir = tgen.logdir
        if hasattr(tgen, "valgrind_existing_files"):
            existing = tgen.valgrind_existing_files
        latest = glob.glob(os.path.join(logdir, "*.valgrind.*"))

    daemons = set()
    for vfile in latest:
        if vfile in existing:
            continue
        existing.append(vfile)
        with open(vfile, encoding="ascii") as vf:
            vfcontent = vf.read()
            match = re.search(r"ERROR SUMMARY: (\d+) errors", vfcontent)
            if match and match.group(1) != "0":
                emsg = "{} in {}".format(match.group(1), vfile)
                leaks.append(emsg)
                daemons.add(re.match(r".*\.valgrind\.(.*)\.\d+", vfile).group(1))

    if tgen is not None:
        tgen.valgrind_existing_files = existing

    if leaks:
        logger.error("valgrind memleaks found:\n\t%s", "\n\t".join(leaks))
        pytest.fail("valgrind memleaks found for daemons: " + " ".join(daemons))


def pytest_runtest_logstart(nodeid, location):
    # location is (filename, lineno, testname)
    topolog.logstart(nodeid, location, topotest_extra_config["rundir"])


def pytest_runtest_logfinish(nodeid, location):
    # location is (filename, lineno, testname)
    topolog.logfinish(nodeid, location)


@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_call(item: pytest.Item) -> None:
    "Hook the function that is called to execute the test."

    # For topology only run the CLI then exit
    if topotest_extra_config["topology_only"]:
        get_topogen().cli()
        pytest.exit("exiting after --topology-only")

    # Let the default pytest_runtest_call execute the test function
    yield

    # Check for leaks if requested
    if topotest_extra_config["valgrind_memleaks"]:
        check_for_memleaks()


def pytest_assertrepr_compare(op, left, right):
    """
    Show proper assertion error message for json_cmp results.
    """
    del op

    json_result = left
    if not isinstance(json_result, json_cmp_result):
        json_result = right
        if not isinstance(json_result, json_cmp_result):
            return None

    return json_result.gen_report()


def pytest_configure(config):
    """
    Assert that the environment is correctly configured, and get extra config.
    """

    if "PYTEST_XDIST_WORKER" not in os.environ:
        os.environ["PYTEST_XDIST_MODE"] = config.getoption("dist", "no")
        os.environ["PYTEST_TOPOTEST_WORKER"] = ""
        is_xdist = os.environ["PYTEST_XDIST_MODE"] != "no"
        is_worker = False
    else:
        os.environ["PYTEST_TOPOTEST_WORKER"] = os.environ["PYTEST_XDIST_WORKER"]
        is_xdist = True
        is_worker = True

    resource.setrlimit(
        resource.RLIMIT_CORE, (resource.RLIM_INFINITY, resource.RLIM_INFINITY)
    )
    # -----------------------------------------------------
    # Set some defaults for the pytest.ini [pytest] section
    # ---------------------------------------------------

    rundir = config.getoption("--rundir")
    if not rundir:
        rundir = config.getini("rundir")
    if not rundir:
        rundir = "/tmp/topotests"
    if not config.getoption("--junitxml"):
        config.option.xmlpath = os.path.join(rundir, "topotests.xml")
    xmlpath = config.option.xmlpath

    # Save an existing topotest.xml
    if os.path.exists(xmlpath):
        fmtime = time.localtime(os.path.getmtime(xmlpath))
        suffix = "-" + time.strftime("%Y%m%d%H%M%S", fmtime)
        commander = Commander("pytest")
        mv_path = commander.get_exec_path("mv")
        commander.cmd_status([mv_path, xmlpath, xmlpath + suffix])

    topotest_extra_config["rundir"] = rundir

    # Set the log_file (exec) to inside the rundir if not specified
    if not config.getoption("--log-file") and not config.getini("log_file"):
        config.option.log_file = os.path.join(rundir, "exec.log")

    # Turn on live logging if user specified verbose and the config has a CLI level set
    if config.getoption("--verbose") and not is_xdist and not config.getini("log_cli"):
        if config.getoption("--log-cli-level", None) is None:
            # By setting the CLI option to the ini value it enables log_cli=1
            cli_level = config.getini("log_cli_level")
            if cli_level is not None:
                config.option.log_cli_level = cli_level

    have_tmux = bool(os.getenv("TMUX", ""))
    have_screen = not have_tmux and bool(os.getenv("STY", ""))
    have_xterm = not have_tmux and not have_screen and bool(os.getenv("DISPLAY", ""))
    have_windows = have_tmux or have_screen or have_xterm
    have_windows_pause = have_tmux or have_xterm
    xdist_no_windows = is_xdist and not is_worker and not have_windows_pause

    def assert_feature_windows(b, feature):
        if b and xdist_no_windows:
            pytest.exit(
                "{} use requires byobu/TMUX/XTerm under dist {}".format(
                    feature, os.environ["PYTEST_XDIST_MODE"]
                )
            )
        elif b and not is_xdist and not have_windows:
            pytest.exit("{} use requires byobu/TMUX/SCREEN/XTerm".format(feature))

    # ---------------------------------------
    # Record our options in global dictionary
    # ---------------------------------------

    topotest_extra_config["rundir"] = rundir

    asan_abort = config.getoption("--asan-abort")
    topotest_extra_config["asan_abort"] = asan_abort

    gdb_routers = config.getoption("--gdb-routers")
    gdb_routers = gdb_routers.split(",") if gdb_routers else []
    topotest_extra_config["gdb_routers"] = gdb_routers

    gdb_daemons = config.getoption("--gdb-daemons")
    gdb_daemons = gdb_daemons.split(",") if gdb_daemons else []
    topotest_extra_config["gdb_daemons"] = gdb_daemons
    assert_feature_windows(gdb_routers or gdb_daemons, "GDB")

    gdb_breakpoints = config.getoption("--gdb-breakpoints")
    gdb_breakpoints = gdb_breakpoints.split(",") if gdb_breakpoints else []
    topotest_extra_config["gdb_breakpoints"] = gdb_breakpoints

    cli_on_error = config.getoption("--cli-on-error")
    topotest_extra_config["cli_on_error"] = cli_on_error
    assert_feature_windows(cli_on_error, "--cli-on-error")

    shell = config.getoption("--shell")
    topotest_extra_config["shell"] = shell.split(",") if shell else []
    assert_feature_windows(shell, "--shell")

    strace = config.getoption("--strace-daemons")
    topotest_extra_config["strace_daemons"] = strace.split(",") if strace else []

    shell_on_error = config.getoption("--shell-on-error")
    topotest_extra_config["shell_on_error"] = shell_on_error
    assert_feature_windows(shell_on_error, "--shell-on-error")

    topotest_extra_config["valgrind_extra"] = config.getoption("--valgrind-extra")
    topotest_extra_config["valgrind_memleaks"] = config.getoption("--valgrind-memleaks")

    vtysh = config.getoption("--vtysh")
    topotest_extra_config["vtysh"] = vtysh.split(",") if vtysh else []
    assert_feature_windows(vtysh, "--vtysh")

    vtysh_on_error = config.getoption("--vtysh-on-error")
    topotest_extra_config["vtysh_on_error"] = vtysh_on_error
    assert_feature_windows(vtysh_on_error, "--vtysh-on-error")

    pause_on_error = vtysh or shell or config.getoption("--pause-on-error")
    if config.getoption("--no-pause-on-error"):
        pause_on_error = False

    topotest_extra_config["pause_on_error"] = pause_on_error
    assert_feature_windows(pause_on_error, "--pause-on-error")

    pause = config.getoption("--pause")
    topotest_extra_config["pause"] = pause
    assert_feature_windows(pause, "--pause")

    topology_only = config.getoption("--topology-only")
    if topology_only and is_xdist:
        pytest.exit("Cannot use --topology-only with distributed test mode")
    topotest_extra_config["topology_only"] = topology_only

    # Check environment now that we have config
    if not diagnose_env(rundir):
        pytest.exit("environment has errors, please read the logs")


@pytest.fixture(autouse=True, scope="session")
def setup_session_auto():
    if "PYTEST_TOPOTEST_WORKER" not in os.environ:
        is_worker = False
    elif not os.environ["PYTEST_TOPOTEST_WORKER"]:
        is_worker = False
    else:
        is_worker = True

    logger.debug("Before the run (is_worker: %s)", is_worker)
    if not is_worker:
        cleanup_previous()
    yield
    if not is_worker:
        cleanup_current()
    logger.debug("After the run (is_worker: %s)", is_worker)


def pytest_runtest_setup(item):
    module = item.parent.module
    script_dir = os.path.abspath(os.path.dirname(module.__file__))
    os.environ["PYTEST_TOPOTEST_SCRIPTDIR"] = script_dir


def pytest_runtest_makereport(item, call):
    "Log all assert messages to default logger with error level"

    # Nothing happened
    if call.when == "call":
        pause = topotest_extra_config["pause"]
    else:
        pause = False

    title = "unset"

    if call.excinfo is None:
        error = False
    else:
        parent = item.parent
        modname = parent.module.__name__

        # Treat skips as non errors, don't pause after
        if call.excinfo.typename == "Skipped":
            pause = False
            error = False
            logger.info(
                'test skipped at "{}/{}": {}'.format(
                    modname, item.name, call.excinfo.value
                )
            )
        else:
            error = True
            # Handle assert failures
            parent._previousfailed = item  # pylint: disable=W0212
            logger.error(
                'test failed at "{}/{}": {}'.format(
                    modname, item.name, call.excinfo.value
                )
            )
            title = "{}/{}".format(modname, item.name)

            # We want to pause, if requested, on any error not just test cases
            # (e.g., call.when == "setup")
            if not pause:
                pause = (
                    topotest_extra_config["pause_on_error"]
                    or topotest_extra_config["pause"]
                )

            # (topogen) Set topology error to avoid advancing in the test.
            tgen = get_topogen()
            if tgen is not None:
                # This will cause topogen to report error on `routers_have_failure`.
                tgen.set_error("{}/{}".format(modname, item.name))

    commander = Commander("pytest")
    isatty = sys.stdout.isatty()
    error_cmd = None

    if error and topotest_extra_config["vtysh_on_error"]:
        error_cmd = commander.get_exec_path(["vtysh"])
    elif error and topotest_extra_config["shell_on_error"]:
        error_cmd = os.getenv("SHELL", commander.get_exec_path(["bash"]))

    if error_cmd:
        is_tmux = bool(os.getenv("TMUX", ""))
        is_screen = not is_tmux and bool(os.getenv("STY", ""))
        is_xterm = not is_tmux and not is_screen and bool(os.getenv("DISPLAY", ""))

        channel = None
        win_info = None
        wait_for_channels = []
        wait_for_procs = []
        # Really would like something better than using this global here.
        # Not all tests use topogen though so get_topogen() won't work.
        for node in Mininet.g_mnet_inst.hosts.values():
            pause = True

            if is_tmux:
                channel = (
                    "{}-{}".format(os.getpid(), Commander.tmux_wait_gen)
                    if not isatty
                    else None
                )
                Commander.tmux_wait_gen += 1
                wait_for_channels.append(channel)

            pane_info = node.run_in_window(
                error_cmd,
                new_window=win_info is None,
                background=True,
                title="{} ({})".format(title, node.name),
                name=title,
                tmux_target=win_info,
                wait_for=channel,
            )
            if is_tmux:
                if win_info is None:
                    win_info = pane_info
            elif is_xterm:
                assert isinstance(pane_info, subprocess.Popen)
                wait_for_procs.append(pane_info)

        # Now wait on any channels
        for channel in wait_for_channels:
            logger.debug("Waiting on TMUX channel %s", channel)
            commander.cmd_raises([commander.get_exec_path("tmux"), "wait", channel])
        for p in wait_for_procs:
            logger.debug("Waiting on TMUX xterm process %s", p)
            o, e = p.communicate()
            if p.wait():
                logger.warning("xterm proc failed: %s:", proc_error(p, o, e))

    if error and topotest_extra_config["cli_on_error"]:
        # Really would like something better than using this global here.
        # Not all tests use topogen though so get_topogen() won't work.
        if Mininet.g_mnet_inst:
            cli(Mininet.g_mnet_inst, title=title, background=False)
        else:
            logger.error("Could not launch CLI b/c no mininet exists yet")

    while pause and isatty:
        try:
            user = raw_input(
                'PAUSED, "cli" for CLI, "pdb" to debug, "Enter" to continue: '
            )
        except NameError:
            user = input('PAUSED, "cli" for CLI, "pdb" to debug, "Enter" to continue: ')
        user = user.strip()

        if user == "cli":
            cli(Mininet.g_mnet_inst)
        elif user == "pdb":
            pdb.set_trace()
        elif user:
            print('Unrecognized input: "%s"' % user)
        else:
            break


#
# Add common fixtures available to all tests as parameters
#

tgen = pytest.fixture(lib.fixtures.tgen)
topo = pytest.fixture(lib.fixtures.topo)