""" Topotest conftest.py file. """ import glob import os import pdb import re import subprocess import sys import time 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 # ----------------------------------------------------- # 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)