""" 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 try: from _pytest._code.code import ExceptionInfo leak_check_ok = True except ImportError: leak_check_ok = False 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(): if not topotest_extra_config["valgrind_memleaks"]: return 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.*")) for vfile in latest: if vfile in existing: continue with open(vfile) 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) if leaks: if leak_check_ok: pytest.fail("Memleaks found:\n\t" + "\n\t".join(leaks)) else: logger.error("Memleaks found:\n\t" + "\n\t".join(leaks)) 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) def pytest_runtest_call(): """ This function must be run after setup_module(), it does standarized post setup routines. It is only being used for the 'topology-only' option. """ if topotest_extra_config["topology_only"]: tgen = get_topogen() if tgen is not None: # Allow user to play with the setup. tgen.cli() pytest.exit("the topology executed successfully") 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") topotest_extra_config["topology_only"] = config.getoption("--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 if call.excinfo is None and call.when == "call": try: check_for_memleaks() except: call.excinfo = ExceptionInfo() 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)