- make sure we close and remove all handlers for named logs on each reuse.
- test module level exec.log no longer truncated to last test case output
- cleanup the log names, and make sure they are present in all exec logs
- keep separate exec logs for each pytest worker when running in distributed mode
- disabled code due to CI infra can't handle it: add per test case exec logs
Signed-off-by: Christian Hopps <chopps@labn.net>
"""
# pylint: disable=consider-using-f-string
+import contextlib
import glob
import logging
import os
import subprocess
import sys
import time
+from pathlib import Path
import lib.fixtures
import pytest
pass
+# Remove this and use munet version when we move to pytest_asyncio
+@contextlib.contextmanager
+def chdir(ndir, desc=""):
+ odir = os.getcwd()
+ os.chdir(ndir)
+ if desc:
+ logging.debug("%s: chdir from %s to %s", desc, odir, ndir)
+ try:
+ yield
+ finally:
+ if desc:
+ logging.debug("%s: chdir back from %s to %s", desc, ndir, odir)
+ os.chdir(odir)
+
+
+@contextlib.contextmanager
+def log_handler(basename, logpath):
+ topolog.logstart(basename, logpath)
+ try:
+ yield
+ finally:
+ topolog.logfinish(basename, logpath)
+
+
def pytest_addoption(parser):
"""
Add topology-only option to the topology tester. This option makes pytest
pytest.fail("memleaks found for daemons: " + " ".join(daemons))
+@pytest.fixture(autouse=True, scope="module")
+def module_autouse(request):
+ basename = get_test_logdir(request.node.nodeid, True)
+ logdir = Path(topotest.g_pytest_config.option.rundir) / basename
+ logpath = logdir / "exec.log"
+
+ subprocess.check_call("mkdir -p -m 1777 {}".format(logdir), shell=True)
+
+ with log_handler(basename, logpath):
+ sdir = os.path.dirname(os.path.realpath(request.fspath))
+ with chdir(sdir, "module autouse fixture"):
+ yield
+
+
@pytest.fixture(autouse=True, scope="module")
def module_check_memtest(request):
yield
check_for_memleaks()
-def pytest_runtest_logstart(nodeid, location):
- # location is (filename, lineno, testname)
- topolog.logstart(nodeid, location, topotest.g_pytest_config.option.rundir)
-
-
-def pytest_runtest_logfinish(nodeid, location):
- # location is (filename, lineno, testname)
- topolog.logfinish(nodeid, location)
+#
+# Disable per test function logging as FRR CI system can't handle it.
+#
+# @pytest.fixture(autouse=True, scope="function")
+# def function_autouse(request):
+# # For tests we actually use the logdir name as the logfile base
+# logbase = get_test_logdir(nodeid=request.node.nodeid, module=False)
+# logbase = os.path.join(topotest.g_pytest_config.option.rundir, logbase)
+# logpath = Path(logbase)
+# path = Path(f"{logpath.parent}/exec-{logpath.name}.log")
+# subprocess.check_call("mkdir -p -m 1777 {}".format(logpath.parent), shell=True)
+# with log_handler(request.node.nodeid, path):
+# yield
@pytest.hookimpl(hookwrapper=True)
os.environ["PYTEST_TOPOTEST_WORKER"] = ""
is_xdist = os.environ["PYTEST_XDIST_MODE"] != "no"
is_worker = False
+ wname = ""
else:
- os.environ["PYTEST_TOPOTEST_WORKER"] = os.environ["PYTEST_XDIST_WORKER"]
+ wname = os.environ["PYTEST_XDIST_WORKER"]
+ os.environ["PYTEST_TOPOTEST_WORKER"] = wname
is_xdist = True
is_worker = True
if not config.getoption("--log-file") and not config.getini("log_file"):
config.option.log_file = os.path.join(rundir, "exec.log")
+ # Handle pytest-xdist each worker get's it's own top level log file
+ # `exec-worker-N.log`
+ if wname:
+ wname = wname.replace("gw", "worker-")
+ cpath = Path(config.option.log_file).absolute()
+ config.option.log_file = f"{cpath.parent}/{cpath.stem}-{wname}{cpath.suffix}"
+ elif is_xdist:
+ cpath = Path(config.option.log_file).absolute()
+ config.option.log_file = f"{cpath.parent}/{cpath.stem}-xdist{cpath.suffix}"
+
# 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:
@pytest.fixture(autouse=True, scope="session")
def setup_session_auto():
+ # Aligns logs nicely
+ logging.addLevelName(logging.WARNING, " WARN")
+ logging.addLevelName(logging.INFO, " INFO")
+
if "PYTEST_TOPOTEST_WORKER" not in os.environ:
is_worker = False
elif not os.environ["PYTEST_TOPOTEST_WORKER"]:
g_mnet_inst = None
- def __init__(self, rundir=None, pytestconfig=None):
+ def __init__(self, rundir=None, pytestconfig=None, logger=None):
"""
Create a Micronet.
"""
# os.umask(0)
super(Mininet, self).__init__(
- pid=False, rundir=rundir, pytestconfig=pytestconfig
+ pid=False, rundir=rundir, pytestconfig=pytestconfig, logger=logger
)
# From munet/munet/native.py
"""Return the command to use for ExaBGP version < 4."""
if commander is None:
- commander = Commander("topogen")
+ commander = Commander("exabgp", logger=logging.getLogger("exabgp"))
def exacmd_version_ok(exacmd):
logger.debug("checking %s for exabgp < version 4", exacmd)
exacmd = py2_path + " -m exabgp"
if exacmd_version_ok(exacmd):
return exacmd
- py2_path = commander.get_exec_path("python")
+ py2_path = commander.get_exec_path("python")
if py2_path:
exacmd = py2_path + " -m exabgp"
if exacmd_version_ok(exacmd):
# Mininet(Micronet) to build the actual topology.
assert not inspect.isclass(topodef)
- self.net = Mininet(rundir=self.logdir, pytestconfig=topotest.g_pytest_config)
+ self.net = Mininet(
+ rundir=self.logdir,
+ pytestconfig=topotest.g_pytest_config,
+ logger=topolog.get_logger("mu", log_level="debug"),
+ )
# Adjust the parent namespace
topotest.fix_netns_limits(self.net)
# pylint: disable=too-few-public-methods
def __init__(self, tgen, name, **params):
+ logger = topolog.get_logger(name, log_level="debug")
super(TopoSwitch, self).__init__(tgen, name, **params)
- tgen.net.add_switch(name)
+ tgen.net.add_switch(name, logger=logger)
def __str__(self):
gear = super(TopoSwitch, self).__str__()
import logging
import os
-import subprocess
-import sys
-
-if sys.version_info[0] > 2:
- pass
-else:
- pass
try:
from xdist import is_xdist_controller
return False
-BASENAME = "topolog"
-
# Helper dictionary to convert Topogen logging levels to Python's logging.
DEBUG_TOPO2LOGGING = {
"debug": logging.DEBUG,
"error": logging.ERROR,
"critical": logging.CRITICAL,
}
-FORMAT = "%(asctime)s.%(msecs)03d %(levelname)s: %(name)s: %(message)s"
+FORMAT = "%(asctime)s %(levelname)s: %(name)s: %(message)s"
handlers = {}
-logger = logging.getLogger("topolog")
+logger = logging.getLogger("topo")
+
+
+# Remove this and use munet version when we move to pytest_asyncio
+def get_test_logdir(nodeid=None, module=False):
+ """Get log directory relative pathname."""
+ xdist_worker = os.getenv("PYTEST_XDIST_WORKER", "")
+ mode = os.getenv("PYTEST_XDIST_MODE", "no")
+
+ # nodeid: all_protocol_startup/test_all_protocol_startup.py::test_router_running
+ # may be missing "::testname" if module is True
+ if not nodeid:
+ nodeid = os.environ["PYTEST_CURRENT_TEST"].split(" ")[0]
+
+ cur_test = nodeid.replace("[", "_").replace("]", "_")
+ if module:
+ idx = cur_test.rfind("::")
+ path = cur_test if idx == -1 else cur_test[:idx]
+ testname = ""
+ else:
+ path, testname = cur_test.split("::")
+ testname = testname.replace("/", ".")
+ path = path[:-3].replace("/", ".")
+ # We use different logdir paths based on how xdist is running.
+ if mode == "each":
+ if module:
+ return os.path.join(path, "worker-logs", xdist_worker)
+ return os.path.join(path, testname, xdist_worker)
+ assert mode in ("no", "load", "loadfile", "loadscope"), f"Unknown dist mode {mode}"
+ return path if module else os.path.join(path, testname)
-def set_handler(l, target=None):
+
+def set_handler(lg, target=None):
if target is None:
h = logging.NullHandler()
else:
h.setFormatter(logging.Formatter(fmt=FORMAT))
# Don't filter anything at the handler level
h.setLevel(logging.DEBUG)
- l.addHandler(h)
+ lg.addHandler(h)
return h
-def set_log_level(l, level):
+def set_log_level(lg, level):
"Set the logging level."
# Messages sent to this logger only are created if this level or above.
log_level = DEBUG_TOPO2LOGGING.get(level, level)
- l.setLevel(log_level)
+ lg.setLevel(log_level)
-def get_logger(name, log_level=None, target=None):
- l = logging.getLogger("{}.{}".format(BASENAME, name))
+def reset_logger(lg):
+ while lg.handlers:
+ x = lg.handlers.pop()
+ x.close()
+ lg.removeHandler(x)
- if log_level is not None:
- set_log_level(l, log_level)
- if target is not None:
- set_handler(l, target)
-
- return l
-
-
-# nodeid: all_protocol_startup/test_all_protocol_startup.py::test_router_running
-
-
-def get_test_logdir(nodeid=None):
- """Get log directory relative pathname."""
- xdist_worker = os.getenv("PYTEST_XDIST_WORKER", "")
- mode = os.getenv("PYTEST_XDIST_MODE", "no")
+def get_logger(name, log_level=None, target=None, reset=True):
+ lg = logging.getLogger(name)
- if not nodeid:
- nodeid = os.environ["PYTEST_CURRENT_TEST"].split(" ")[0]
+ if reset:
+ reset_logger(lg)
- cur_test = nodeid.replace("[", "_").replace("]", "_")
- path, testname = cur_test.split("::")
- path = path[:-3].replace("/", ".")
+ if log_level is not None:
+ set_log_level(lg, log_level)
- # We use different logdir paths based on how xdist is running.
- if mode == "each":
- return os.path.join(path, testname, xdist_worker)
- elif mode == "load":
- return os.path.join(path, testname)
- else:
- assert (
- mode == "no" or mode == "loadfile" or mode == "loadscope"
- ), "Unknown dist mode {}".format(mode)
+ if target is not None:
+ set_handler(lg, target)
- return path
+ return lg
-def logstart(nodeid, location, rundir):
+def logstart(nodeid, logpath):
"""Called from pytest before module setup."""
-
- mode = os.getenv("PYTEST_XDIST_MODE", "no")
worker = os.getenv("PYTEST_TOPOTEST_WORKER", "")
+ wstr = f" on worker {worker}" if worker else ""
+ handler_id = nodeid + worker
+ logpath = logpath.absolute()
- # We only per-test log in the workers (or non-dist)
- if not worker and mode != "no":
- return
+ logging.debug("logstart: adding logging for %s%s at %s", nodeid, wstr, logpath)
+ root_logger = logging.getLogger()
+ handler = logging.FileHandler(logpath, mode="w")
+ handler.setFormatter(logging.Formatter(FORMAT))
- handler_id = nodeid + worker
- assert handler_id not in handlers
-
- rel_log_dir = get_test_logdir(nodeid)
- exec_log_dir = os.path.join(rundir, rel_log_dir)
- subprocess.check_call(
- "mkdir -p {0} && chmod 1777 {0}".format(exec_log_dir), shell=True
- )
- exec_log_path = os.path.join(exec_log_dir, "exec.log")
-
- # Add test based exec log handler
- h = set_handler(logger, exec_log_path)
- handlers[handler_id] = h
-
- if worker:
- logger.info(
- "Logging on worker %s for %s into %s", worker, handler_id, exec_log_path
- )
- else:
- logger.info("Logging for %s into %s", handler_id, exec_log_path)
+ root_logger.addHandler(handler)
+ handlers[handler_id] = handler
+ logging.debug("logstart: added logging for %s%s at %s", nodeid, wstr, logpath)
+ return handler
-def logfinish(nodeid, location):
- """Called from pytest after module teardown."""
- # This function may not be called if pytest is interrupted.
+def logfinish(nodeid, logpath):
+ """Called from pytest after module teardown."""
worker = os.getenv("PYTEST_TOPOTEST_WORKER", "")
- handler_id = nodeid + worker
+ wstr = f" on worker {worker}" if worker else ""
+
+ root_logger = logging.getLogger()
- if handler_id in handlers:
- # Remove test based exec log handler
- if worker:
- logger.info("Closing logs for %s", handler_id)
+ handler_id = nodeid + worker
+ if handler_id not in handlers:
+ logging.critical("can't find log handler to remove")
+ else:
+ logging.debug(
+ "logfinish: removing logging for %s%s at %s", nodeid, wstr, logpath
+ )
h = handlers[handler_id]
- logger.removeHandler(handlers[handler_id])
+ root_logger.removeHandler(h)
h.flush()
h.close()
del handlers[handler_id]
+ logging.debug(
+ "logfinish: removed logging for %s%s at %s", nodeid, wstr, logpath
+ )
console_handler = set_handler(logger, None)
import sys
import tempfile
import time
+import logging
from collections.abc import Mapping
from copy import deepcopy
def get_logs_path(rundir):
- logspath = topolog.get_test_logdir()
+ logspath = topolog.get_test_logdir(module=True)
return os.path.join(rundir, logspath)
def sysctl_atleast(commander, variable, min_value, raises=False):
try:
if commander is None:
- commander = micronet.Commander("topotest")
+ logger = logging.getLogger("topotest")
+ commander = micronet.Commander("sysctl", logger=logger)
+
return _sysctl_atleast(commander, variable, min_value)
except subprocess.CalledProcessError as error:
logger.warning(
def sysctl_assure(commander, variable, value, raises=False):
try:
if commander is None:
- commander = micronet.Commander("topotest")
+ logger = logging.getLogger("topotest")
+ commander = micronet.Commander("sysctl", logger=logger)
return _sysctl_assure(commander, variable, value)
except subprocess.CalledProcessError as error:
logger.warning(
projects / matthieu / frr.git / commitdiff