# information regarding copyright ownership.
from functools import partial
-import logging
import os
from pathlib import Path
import re
# ----------------------- Globals definition -----------------------------
-LOG_FORMAT = "%(asctime)s %(levelname)7s:%(name)s %(message)s"
XDIST_WORKER = os.environ.get("PYTEST_XDIST_WORKER", "")
FILE_DIR = os.path.abspath(Path(__file__).parent)
ENV_RE = re.compile(b"([^=]+)=(.*)")
"upforwd/",
]
PRIORITY_TESTS_RE = re.compile("|".join(PRIORITY_TESTS))
-CONFTEST_LOGGER = logging.getLogger("conftest")
SYSTEM_TEST_DIR_GIT_PATH = "bin/tests/system"
SYSTEM_TEST_NAME_RE = re.compile(f"{SYSTEM_TEST_DIR_GIT_PATH}" + r"/([^/]+)")
SYMLINK_REPLACEMENT_RE = re.compile(r"/tests(_.*)\.py")
# ---------------------- Module initialization ---------------------------
-def init_pytest_conftest_logger(conftest_logger):
- """
- This initializes the conftest logger which is used for pytest setup
- and configuration before tests are executed -- aka any logging in this
- file that is _not_ module-specific.
- """
- conftest_logger.setLevel(logging.DEBUG)
- file_handler = logging.FileHandler("pytest.conftest.log.txt")
- file_handler.setLevel(logging.DEBUG)
- file_handler.setFormatter(logging.Formatter(LOG_FORMAT))
- conftest_logger.addHandler(file_handler)
-
-
-init_pytest_conftest_logger(CONFTEST_LOGGER)
-
-
-def avoid_duplicated_logs():
- """
- Remove direct root logger output to file descriptors.
- This default is causing duplicates because all our messages go through
- regular logging as well and are thus displayed twice.
- """
- todel = []
- for handler in logging.root.handlers:
- if handler.__class__ == logging.StreamHandler:
- # Beware: As for pytest 7.2.2, LiveLogging and LogCapture
- # handlers inherit from logging.StreamHandler
- todel.append(handler)
- for handler in todel:
- logging.root.handlers.remove(handler)
-
-
def parse_env(env_bytes):
"""Parse the POSIX env format into Python dictionary."""
out = {}
stdout=subprocess.PIPE,
)
except subprocess.CalledProcessError as exc:
- CONFTEST_LOGGER.error("failed to get shell env: %s", exc)
+ isctest.log.error("failed to get shell env: %s", exc)
raise exc
env_bytes = proc.stdout
return parse_env(env_bytes)
# FUTURE: Remove conf.sh entirely and define all variables in pytest only.
CONF_ENV = get_env_bytes(". ./conf.sh && env")
os.environb.update(CONF_ENV)
-CONFTEST_LOGGER.debug("variables in env: %s", ", ".join([str(key) for key in CONF_ENV]))
+isctest.log.debug("variables in env: %s", ", ".join([str(key) for key in CONF_ENV]))
# --------------------------- pytest hooks -------------------------------
try:
import xdist.scheduler.loadscope # pylint: disable=unused-import
except ImportError:
- CONFTEST_LOGGER.debug(
+ isctest.log.debug(
"xdist is too old and does not have "
"scheduler.loadscope, disabling parallelism"
)
# is otherwise and invalid character for a system test name.
match = SYSTEM_TEST_NAME_RE.search(str(path))
if match is None:
- CONFTEST_LOGGER.warning("unexpected test path: %s (ignored)", path)
+ isctest.log.warning("unexpected test path: %s (ignored)", path)
return True
system_test_name = match.groups()[0]
return "_" in system_test_name
return path.parent.name
-@pytest.fixture(scope="module")
-def mlogger(system_test_name):
- """Logging facility specific to this test module."""
- avoid_duplicated_logs()
- return logging.getLogger(system_test_name)
-
-
def _get_marker(node, marker):
try:
# pytest >= 4.x
watcher.wait_for_line("all zones loaded")
-@pytest.fixture
+@pytest.fixture(autouse=True)
def logger(request, system_test_name):
- """Logging facility specific to a particular test."""
- return logging.getLogger(f"{system_test_name}.{request.node.name}")
+ """Sets up logging facility specific to a particular test."""
+ isctest.log.init_test_logger(system_test_name, request.node.name)
+ yield
+ isctest.log.deinit_test_logger()
@pytest.fixture(scope="module")
def system_test_dir(
- request, env, system_test_name, mlogger
+ request, env, system_test_name
): # pylint: disable=too-many-statements,too-many-locals
"""
Temporary directory for executing the test.
except AttributeError:
# This may happen if pytest execution is interrupted and
# pytest_runtest_makereport() is never called.
- mlogger.debug("can't obtain test results, test run was interrupted")
+ isctest.log.debug("can't obtain test results, test run was interrupted")
return "error"
test_results = {
node.nodeid: all_test_results[node.nodeid]
assert len(test_results)
messages = []
for node, result in test_results.items():
- mlogger.debug("%s %s", result.outcome.upper(), node)
+ isctest.log.debug("%s %s", result.outcome.upper(), node)
messages.extend(result.messages)
for message in messages:
- mlogger.debug("\n" + message)
+ isctest.log.debug("\n" + message)
failed = any(res.outcome == "failed" for res in test_results.values())
skipped = any(res.outcome == "skipped" for res in test_results.values())
if failed:
unlink(symlink_dst)
symlink_dst.symlink_to(os.path.relpath(testdir, start=system_test_root))
- # Configure logger to write to a file inside the temporary test directory
- mlogger.handlers.clear()
- mlogger.setLevel(logging.DEBUG)
- handler = logging.FileHandler(testdir / "pytest.log.txt", mode="w")
- formatter = logging.Formatter(LOG_FORMAT)
- handler.setFormatter(formatter)
- mlogger.addHandler(handler)
+ isctest.log.init_module_logger(system_test_name, testdir)
# System tests are meant to be executed from their directory - switch to it.
old_cwd = os.getcwd()
os.chdir(testdir)
- mlogger.debug("switching to tmpdir: %s", testdir)
+ isctest.log.info("switching to tmpdir: %s", testdir)
try:
yield testdir # other fixtures / tests will execute here
finally:
os.chdir(old_cwd)
- mlogger.debug("changed workdir to: %s", old_cwd)
+ isctest.log.debug("changed workdir to: %s", old_cwd)
result = get_test_result()
# Clean temporary dir unless it should be kept
keep = False
if request.config.getoption("--noclean"):
- mlogger.debug(
+ isctest.log.debug(
"--noclean requested, keeping temporary directory %s", testdir
)
keep = True
elif result == "failed":
- mlogger.debug(
+ isctest.log.debug(
"test failure detected, keeping temporary directory %s", testdir
)
keep = True
elif not request.node.stash[FIXTURE_OK]:
- mlogger.debug(
+ isctest.log.debug(
"test setup/teardown issue detected, keeping temporary directory %s",
testdir,
)
keep = True
if keep:
- mlogger.info(
+ isctest.log.info(
"test artifacts in: %s", symlink_dst.relative_to(system_test_root)
)
else:
- mlogger.debug("deleting temporary directory")
- handler.flush()
- handler.close()
+ isctest.log.debug("deleting temporary directory")
+
+ isctest.log.deinit_module_logger()
+ if not keep:
shutil.rmtree(testdir)
unlink(symlink_dst)
def _run_script( # pylint: disable=too-many-arguments
env,
- mlogger,
system_test_dir: Path,
interpreter: str,
script: str,
cwd = os.getcwd()
if not path.exists():
raise FileNotFoundError(f"script {script} not found in {cwd}")
- mlogger.debug("running script: %s %s %s", interpreter, script, " ".join(args))
- mlogger.debug(" workdir: %s", cwd)
+ isctest.log.debug("running script: %s %s %s", interpreter, script, " ".join(args))
+ isctest.log.debug(" workdir: %s", cwd)
returncode = 1
cmd = [interpreter, script] + args
) as proc:
if proc.stdout:
for line in proc.stdout:
- mlogger.info(" %s", line.rstrip("\n"))
+ isctest.log.info(" %s", line.rstrip("\n"))
proc.communicate()
returncode = proc.returncode
if returncode:
raise subprocess.CalledProcessError(returncode, cmd)
- mlogger.debug(" exited with %d", returncode)
+ isctest.log.debug(" exited with %d", returncode)
@pytest.fixture(scope="module")
-def shell(env, system_test_dir, mlogger):
+def shell(env, system_test_dir):
"""Function to call a shell script with arguments."""
- return partial(_run_script, env, mlogger, system_test_dir, env["SHELL"])
+ return partial(_run_script, env, system_test_dir, env["SHELL"])
@pytest.fixture(scope="module")
-def perl(env, system_test_dir, mlogger):
+def perl(env, system_test_dir):
"""Function to call a perl script with arguments."""
- return partial(_run_script, env, mlogger, system_test_dir, env["PERL"])
+ return partial(_run_script, env, system_test_dir, env["PERL"])
@pytest.fixture(scope="module")
def system_test( # pylint: disable=too-many-arguments,too-many-statements
request,
env: Dict[str, str],
- mlogger,
system_test_dir,
shell,
perl,
try:
perl("testsock.pl", ["-p", env["PORT"]])
except subprocess.CalledProcessError as exc:
- mlogger.error("testsock.pl: exited with code %d", exc.returncode)
+ isctest.log.error("testsock.pl: exited with code %d", exc.returncode)
pytest.skip("Network interface aliases not set up.")
def check_prerequisites():
except FileNotFoundError:
pass # setup.sh is optional
except subprocess.CalledProcessError as exc:
- mlogger.error("Failed to run test setup")
+ isctest.log.error("Failed to run test setup")
pytest.fail(f"setup.sh exited with {exc.returncode}")
def start_servers():
try:
perl("start.pl", ["--port", env["PORT"], system_test_dir.name])
except subprocess.CalledProcessError as exc:
- mlogger.error("Failed to start servers")
+ isctest.log.error("Failed to start servers")
pytest.fail(f"start.pl exited with {exc.returncode}")
def stop_servers():
try:
perl("stop.pl", [system_test_dir.name])
except subprocess.CalledProcessError as exc:
- mlogger.error("Failed to stop servers")
+ isctest.log.error("Failed to stop servers")
get_core_dumps()
pytest.fail(f"stop.pl exited with {exc.returncode}")
try:
shell("get_core_dumps.sh", [system_test_dir.name])
except subprocess.CalledProcessError as exc:
- mlogger.error("Found core dumps or sanitizer reports")
+ isctest.log.error("Found core dumps or sanitizer reports")
pytest.fail(f"get_core_dumps.sh exited with {exc.returncode}")
os.environ.update(env) # Ensure pytests have the same env vars as shell tests.
- mlogger.info(f"test started: {request.node.name}")
+ isctest.log.info(f"test started: {request.node.name}")
port = int(env["PORT"])
- mlogger.info("using port range: <%d, %d>", port, port + PORTS_PER_TEST - 1)
+ isctest.log.info("using port range: <%d, %d>", port, port + PORTS_PER_TEST - 1)
if not hasattr(request.node, "stash"): # compatibility with pytest<7.0.0
request.node.stash = {} # use regular dict instead of pytest.Stash
setup_test()
try:
start_servers()
- mlogger.debug("executing test(s)")
+ isctest.log.debug("executing test(s)")
yield
finally:
- mlogger.debug("test(s) finished")
+ isctest.log.debug("test(s) finished")
stop_servers()
get_core_dumps()
request.node.stash[FIXTURE_OK] = True
@pytest.fixture
-def servers(ports, logger, system_test_dir):
+def servers(ports, system_test_dir):
instances = {}
for entry in system_test_dir.rglob("*"):
if entry.is_dir():
named_ports = isctest.instance.NamedPorts(
dns=int(ports["PORT"]), rndc=int(ports["CONTROLPORT"])
)
- instance = isctest.instance.NamedInstance(dir_name, named_ports, logger)
+ instance = isctest.instance.NamedInstance(dir_name, named_ports)
instances[dir_name] = instance
except ValueError:
continue
--- /dev/null
+# Copyright (C) Internet Systems Consortium, Inc. ("ISC")
+#
+# SPDX-License-Identifier: MPL-2.0
+#
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0. If a copy of the MPL was not distributed with this
+# file, you can obtain one at https://mozilla.org/MPL/2.0/.
+#
+# See the COPYRIGHT file distributed with this work for additional
+# information regarding copyright ownership.
+
+import logging
+from pathlib import Path
+from typing import Dict, Optional
+
+
+CONFTEST_LOGGER = logging.getLogger("conftest")
+LOG_FORMAT = "%(asctime)s %(levelname)7s:%(name)s %(message)s"
+
+LOGGERS = {
+ "conftest": None,
+ "module": None,
+ "test": None,
+} # type: Dict[str, Optional[logging.Logger]]
+
+
+def init_conftest_logger():
+ """
+ This initializes the conftest logger which is used for pytest setup
+ and configuration before tests are executed -- aka any logging in this
+ file that is _not_ module-specific.
+ """
+ LOGGERS["conftest"] = logging.getLogger("conftest")
+ LOGGERS["conftest"].setLevel(logging.DEBUG)
+ file_handler = logging.FileHandler("pytest.conftest.log.txt")
+ file_handler.setFormatter(logging.Formatter(LOG_FORMAT))
+ LOGGERS["conftest"].addHandler(file_handler)
+
+
+def avoid_duplicated_logs():
+ """
+ Remove direct root logger output to file descriptors.
+ This default is causing duplicates because all our messages go through
+ regular logging as well and are thus displayed twice.
+ """
+ todel = []
+ for handler in logging.root.handlers:
+ if handler.__class__ == logging.StreamHandler:
+ # Beware: As for pytest 7.2.2, LiveLogging and LogCapture
+ # handlers inherit from logging.StreamHandler
+ todel.append(handler)
+ for handler in todel:
+ logging.root.handlers.remove(handler)
+
+
+init_conftest_logger()
+avoid_duplicated_logs()
+
+
+def init_module_logger(system_test_name: str, testdir: Path):
+ logger = logging.getLogger(system_test_name)
+ logger.handlers.clear()
+ logger.setLevel(logging.DEBUG)
+ handler = logging.FileHandler(testdir / "pytest.log.txt", mode="w")
+ handler.setFormatter(logging.Formatter(LOG_FORMAT))
+ logger.addHandler(handler)
+ LOGGERS["module"] = logger
+
+
+def deinit_module_logger():
+ for handler in LOGGERS["module"].handlers:
+ handler.flush()
+ handler.close()
+ LOGGERS["module"] = None
+
+
+def init_test_logger(system_test_name: str, test_name: str):
+ LOGGERS["test"] = logging.getLogger(f"{system_test_name}.{test_name}")
+
+
+def deinit_test_logger():
+ LOGGERS["test"] = None
+
+
+def log(lvl: int, msg: str, *args, **kwargs):
+ """Log message with the most-specific logger currently available."""
+ logger = LOGGERS["test"]
+ if logger is None:
+ logger = LOGGERS["module"]
+ if logger is None:
+ logger = LOGGERS["conftest"]
+ assert logger is not None
+ logger.log(lvl, msg, *args, **kwargs)
+
+
+def debug(msg: str, *args, **kwargs):
+ log(logging.DEBUG, msg, *args, **kwargs)
+
+
+def info(msg: str, *args, **kwargs):
+ log(logging.INFO, msg, *args, **kwargs)
+
+
+def warning(msg: str, *args, **kwargs):
+ log(logging.WARNING, msg, *args, **kwargs)
+
+
+def error(msg: str, *args, **kwargs):
+ log(logging.ERROR, msg, *args, **kwargs)
+
+
+def critical(msg: str, *args, **kwargs):
+ log(logging.CRITICAL, msg, *args, **kwargs)
projects / thirdparty / bind9.git / commitdiff
