+# -*- coding: utf-8 eval: (blacken-mode 1) -*-
"""
Topotest conftest.py file.
"""
import glob
import os
-import pdb
import re
import resource
import subprocess
import lib.fixtures
import pytest
-from lib.micronet_compat import Mininet
+from lib.micronet_compat import ConfigOptionsProxy, Mininet
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
from munet import cli
from munet.base import Commander, proc_error
from munet.cleanup import cleanup_current, cleanup_previous
+from munet.testing.util import pause_test
-from lib import topolog
+from lib import topolog, topotest
def pytest_addoption(parser):
def check_for_memleaks():
- assert topotest_extra_config["valgrind_memleaks"]
+ assert topotest.g_pytest_config.option.valgrind_memleaks
leaks = []
tgen = get_topogen() # pylint: disable=redefined-outer-name
@pytest.fixture(autouse=True, scope="module")
def module_check_memtest(request):
- del request # disable unused warning
yield
- if topotest_extra_config["valgrind_memleaks"]:
+ if request.config.option.valgrind_memleaks:
if get_topogen() is not None:
check_for_memleaks()
def pytest_runtest_logstart(nodeid, location):
# location is (filename, lineno, testname)
- topolog.logstart(nodeid, location, topotest_extra_config["rundir"])
+ topolog.logstart(nodeid, location, topotest.g_pytest_config.option.rundir)
def pytest_runtest_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."
- del item # disable unused warning
# For topology only run the CLI then exit
- if topotest_extra_config["topology_only"]:
+ if item.config.option.topology_only:
get_topogen().cli()
pytest.exit("exiting after --topology-only")
yield
# Check for leaks if requested
- if topotest_extra_config["valgrind_memleaks"]:
+ if item.config.option.valgrind_memleaks:
check_for_memleaks()
"""
Assert that the environment is correctly configured, and get extra config.
"""
+ topotest.g_pytest_config = ConfigOptionsProxy(config)
if config.getoption("--collect-only"):
return
# Set some defaults for the pytest.ini [pytest] section
# ---------------------------------------------------
- rundir = config.getoption("--rundir")
+ rundir = config.option.rundir
if not rundir:
rundir = config.getini("rundir")
if not rundir:
rundir = "/tmp/topotests"
+ config.option.rundir = rundir
+
if not config.getoption("--junitxml"):
config.option.xmlpath = os.path.join(rundir, "topotests.xml")
xmlpath = config.option.xmlpath
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")
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:
+ #
+ # Check for window capability if given options that require window
+ #
+ assert_feature_windows(config.option.gdb_routers, "GDB")
+ assert_feature_windows(config.option.gdb_daemons, "GDB")
+ assert_feature_windows(config.option.cli_on_error, "--cli-on-error")
+ assert_feature_windows(config.option.shell, "--shell")
+ assert_feature_windows(config.option.shell_on_error, "--shell-on-error")
+ assert_feature_windows(config.option.vtysh, "--vtysh")
+ assert_feature_windows(config.option.vtysh_on_error, "--vtysh-on-error")
+
+ if config.option.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):
# 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"]
- )
+ pause = item.config.option.pause_on_error or item.config.option.pause
# (topogen) Set topology error to avoid advancing in the test.
tgen = get_topogen() # pylint: disable=redefined-outer-name
isatty = sys.stdout.isatty()
error_cmd = None
- if error and topotest_extra_config["vtysh_on_error"]:
+ if error and item.config.option.vtysh_on_error:
error_cmd = commander.get_exec_path(["vtysh"])
- elif error and topotest_extra_config["shell_on_error"]:
+ elif error and item.config.option.shell_on_error:
error_cmd = os.getenv("SHELL", commander.get_exec_path(["bash"]))
if error_cmd:
if p.wait():
logger.warning("xterm proc failed: %s:", proc_error(p, o, e))
- if error and topotest_extra_config["cli_on_error"]:
+ if error and item.config.option.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:
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.cli(Mininet.g_mnet_inst)
- elif user == "pdb":
- pdb.set_trace() # pylint: disable=forgotten-debug-statement
- elif user:
- print('Unrecognized input: "%s"' % user)
- else:
- break
+ if pause and isatty:
+ pause_test()
#
from munet.base import BaseMunet, LinuxNamespace
+def cli_opt_list(option_list):
+ if not option_list:
+ return []
+ if isinstance(option_list, str):
+ return [x for x in option_list.split(",") if x]
+ return [x for x in option_list if x]
+
+
+def name_in_cli_opt_str(name, option_list):
+ ol = cli_opt_list(option_list)
+ return name in ol or "all" in ol
+
+
+class ConfigOptionsProxy:
+ def __init__(self, pytestconfig=None):
+ if isinstance(pytestconfig, ConfigOptionsProxy):
+ self.config = pytestconfig.config
+ else:
+ self.config = pytestconfig
+ self.option = self.config.option
+
+ def getoption(self, opt, defval=None):
+ if not self.config:
+ return defval
+
+ value = self.config.getoption(opt)
+ if value is None:
+ return defval
+
+ return value
+
+ def get_option(self, opt, defval=None):
+ return self.getoption(opt, defval)
+
+ def get_option_list(self, opt):
+ value = self.get_option(opt, "")
+ return cli_opt_list(value)
+
+ def name_in_option_list(self, name, opt):
+ optlist = self.get_option_list(opt)
+ return "all" in optlist or name in optlist
+
+
class Node(LinuxNamespace):
"""Node (mininet compat)."""
g_mnet_inst = None
- def __init__(self, rundir=None):
+ def __init__(self, rundir=None, pytestconfig=None):
"""
Create a Micronet.
"""
self.host_params = {}
self.prefix_len = 8
+ self.cfgopt = ConfigOptionsProxy(pytestconfig)
+
# SNMPd used to require this, which was set int he mininet shell
# that all commands executed from. This is goofy default so let's not
# do it if we don't have to. The snmpd.conf files have been updated
cli.add_cli_config(self, cdict)
- # shellopt = (
- # self.pytest_config.getoption("--shell") if self.pytest_config else None
- # )
- # shellopt = shellopt if shellopt is not None else ""
- # if shellopt == "all" or "." in shellopt.split(","):
- # self.run_in_window("bash")
+ shellopt = self.cfgopt.get_option_list("--shell")
+ if "all" in shellopt or "." in shellopt:
+ self.run_in_window("bash")
# This is expected by newer munet CLI code
self.config_dirname = ""
# Network Device Education Foundation, Inc. ("NetDEF")
#
+import configparser
import difflib
import errno
import functools
import glob
import json
import os
-import pdb
import platform
import re
import resource
import sys
import tempfile
import time
+from collections.abc import Mapping
from copy import deepcopy
import lib.topolog as topolog
+from lib.micronet_compat import Node
from lib.topolog import logger
-if sys.version_info[0] > 2:
- import configparser
- from collections.abc import Mapping
-else:
- import ConfigParser as configparser
- from collections import Mapping
-
from lib import micronet
-from lib.micronet_compat import Node
-g_extra_config = {}
+g_pytest_config = None
def get_logs_path(rundir):
# specified, then attempt to generate an unique logdir.
self.logdir = params.get("logdir")
if self.logdir is None:
- self.logdir = get_logs_path(g_extra_config["rundir"])
+ self.logdir = get_logs_path(g_pytest_config.getoption("--rundir"))
if not params.get("logger"):
# If logger is present topogen has already set this up
)
except Exception as ex:
logger.error("%s can't remove IPs %s", self, str(ex))
- # pdb.set_trace()
+ # breakpoint()
# assert False, "can't remove IPs %s" % str(ex)
def checkCapability(self, daemon, param):
if (daemon == "zebra") and (self.daemons["mgmtd"] == 0):
# Add mgmtd with zebra - if it exists
- try:
- mgmtd_path = os.path.join(self.daemondir, "mgmtd")
- except:
- pdb.set_trace()
+ mgmtd_path = os.path.join(self.daemondir, "mgmtd")
if os.path.isfile(mgmtd_path):
self.daemons["mgmtd"] = 1
self.daemons_options["mgmtd"] = ""
if (daemon == "zebra") and (self.daemons["staticd"] == 0):
# Add staticd with zebra - if it exists
- try:
- staticd_path = os.path.join(self.daemondir, "staticd")
- except:
- pdb.set_trace()
-
+ staticd_path = os.path.join(self.daemondir, "staticd")
if os.path.isfile(staticd_path):
self.daemons["staticd"] = 1
self.daemons_options["staticd"] = ""
# used
self.cmd("echo 100000 > /proc/sys/net/mpls/platform_labels")
- shell_routers = g_extra_config["shell"]
- if "all" in shell_routers or self.name in shell_routers:
+ if g_pytest_config.name_in_option_list(self.name, "--shell"):
self.run_in_window(os.getenv("SHELL", "bash"), title="sh-%s" % self.name)
if self.daemons["eigrpd"] == 1:
status = self.startRouterDaemons(tgen=tgen)
- vtysh_routers = g_extra_config["vtysh"]
- if "all" in vtysh_routers or self.name in vtysh_routers:
+ if g_pytest_config.name_in_option_list(self.name, "--vtysh"):
self.run_in_window("vtysh", title="vt-%s" % self.name)
if self.unified_config:
def startRouterDaemons(self, daemons=None, tgen=None):
"Starts FRR daemons for this router."
- asan_abort = g_extra_config["asan_abort"]
- gdb_breakpoints = g_extra_config["gdb_breakpoints"]
- gdb_daemons = g_extra_config["gdb_daemons"]
- gdb_routers = g_extra_config["gdb_routers"]
- valgrind_extra = g_extra_config["valgrind_extra"]
- valgrind_memleaks = g_extra_config["valgrind_memleaks"]
- strace_daemons = g_extra_config["strace_daemons"]
+ asan_abort = bool(g_pytest_config.option.asan_abort)
+ gdb_breakpoints = g_pytest_config.get_option_list("--gdb-breakpoints")
+ gdb_daemons = g_pytest_config.get_option_list("--gdb-daemons")
+ gdb_routers = g_pytest_config.get_option_list("--gdb-routers")
+ valgrind_extra = bool(g_pytest_config.option.valgrind_extra)
+ valgrind_memleaks = bool(g_pytest_config.option.valgrind_memleaks)
+ strace_daemons = g_pytest_config.get_option_list("--strace-daemons")
# Get global bundle data
if not self.path_exists("/etc/frr/support_bundle_commands.conf"):
self.reportCores = True
# XXX: glue code forward ported from removed function.
- if self.version == None:
+ if self.version is None:
self.version = self.cmd(
os.path.join(self.daemondir, "bgpd") + " -v"
).split()[2]
projects / mirror / frr.git / commitdiff