# ---------------------- Module initialization ---------------------------
+
def init_pytest_conftest_logger(conftest_logger):
"""
This initializes the conftest logger which is used for pytest setup
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.
for handler in todel:
logging.root.handlers.remove(handler)
+
def parse_env(env_bytes):
"""Parse the POSIX env format into Python dictionary."""
out = {}
out[match.groups()[0]] = match.groups()[1]
return out
+
def get_env_bytes(cmd):
try:
proc = subprocess.run(
env_bytes = proc.stdout
return parse_env(env_bytes)
+
# Read common environment variables for running tests from conf.sh.
# 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])
-)
+CONFTEST_LOGGER.debug("variables in env: %s", ", ".join([str(key) for key in CONF_ENV]))
# --------------------------- pytest hooks -------------------------------
+
def pytest_addoption(parser):
parser.addoption(
"--noclean",
help="don't remove the temporary test directories with artifacts",
)
+
def pytest_configure(config):
# Ensure this hook only runs on the main pytest instance if xdist is
# used to spawn other workers.
else:
config.option.dist = "loadscope"
+
def pytest_ignore_collect(path):
# System tests are executed in temporary directories inside
# bin/tests/system. These temporary directories contain all files
system_test_name = match.groups()[0]
return "_" in system_test_name
+
def pytest_collection_modifyitems(items):
"""Schedule long-running tests first to get more benefit from parallelism."""
priority = []
other.append(item)
items[:] = priority + other
+
class NodeResult:
def __init__(self, report=None):
self.outcome = None
if report.longreprtext:
self.messages.append(report.longreprtext)
+
@pytest.hookimpl(tryfirst=True, hookwrapper=True)
def pytest_runtest_makereport(item):
"""Hook that is used to expose test results to session (for use in fixtures)."""
node_result = test_results.setdefault(item.nodeid, NodeResult())
node_result.update(report)
+
# --------------------------- Fixtures -----------------------------------
+
@pytest.fixture(scope="session")
def modules():
"""
mods.append(mod)
return sorted(mods)
+
@pytest.fixture(scope="session")
def module_base_ports(modules):
"""
port_min = PORT_MIN
port_max = PORT_MAX - len(modules) * PORTS_PER_TEST
if port_max < port_min:
- raise RuntimeError(
- "not enough ports to assign unique port set to each module"
- )
+ raise RuntimeError("not enough ports to assign unique port set to each module")
# Rotate the base port value over time to detect possible test issues
# with using random ports. This introduces a very slight race condition
return {mod: base_port + i * PORTS_PER_TEST for i, mod in enumerate(modules)}
+
@pytest.fixture(scope="module")
def base_port(request, module_base_ports):
"""Start of the port range assigned to a particular test module."""
port = module_base_ports[request.fspath]
return port
+
@pytest.fixture(scope="module")
def ports(base_port):
"""Dictionary containing port names and their assigned values."""
"CONTROLPORT": str(base_port + 12),
}
+
@pytest.fixture(scope="module")
def env(ports):
"""Dictionary containing environment variables for the test."""
env["srcdir"] = f"{env['TOP_SRCDIR']}/{SYSTEM_TEST_DIR_GIT_PATH}"
return env
+
@pytest.fixture(scope="module")
def system_test_name(request):
"""Name of the system test directory."""
path = Path(request.fspath)
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)
+
@pytest.fixture
def logger(request, system_test_name):
"""Logging facility specific to a particular test."""
return logging.getLogger(f"{system_test_name}.{request.node.name}")
+
@pytest.fixture(scope="module")
def system_test_dir(
request, env, system_test_name, mlogger
shutil.rmtree(testdir)
unlink(symlink_dst)
+
def _run_script( # pylint: disable=too-many-arguments
env,
mlogger,
raise subprocess.CalledProcessError(returncode, cmd)
mlogger.debug(" exited with %d", returncode)
+
@pytest.fixture(scope="module")
def shell(env, system_test_dir, mlogger):
"""Function to call a shell script with arguments."""
return partial(_run_script, env, mlogger, system_test_dir, env["SHELL"])
+
@pytest.fixture(scope="module")
def perl(env, system_test_dir, mlogger):
"""Function to call a perl script with arguments."""
return partial(_run_script, env, mlogger, system_test_dir, env["PERL"])
+
@pytest.fixture(scope="module")
def run_tests_sh(system_test_dir, shell):
"""Utility function to execute tests.sh as a python test."""
return run_tests
+
@pytest.fixture(scope="module", autouse=True)
def system_test( # pylint: disable=too-many-arguments,too-many-statements
request,
projects / thirdparty / bind9.git / commitdiff
