diff --git a/bazel/py_itf_plugin.bzl b/bazel/py_itf_plugin.bzl index 967ef25..39c30be 100644 --- a/bazel/py_itf_plugin.bzl +++ b/bazel/py_itf_plugin.bzl @@ -12,6 +12,7 @@ # ******************************************************************************* """Bazel rule for defining ITF test plugins.""" +load("@bazel_skylib//rules:common_settings.bzl", "BuildSettingInfo") load("@rules_python//python:defs.bzl", "PyInfo") PyItfPluginInfo = provider( @@ -34,6 +35,15 @@ def _py_itf_plugin_impl(ctx): arg = arg.replace("$(location ", "$(rootpath ").replace("$(locations ", "$(rootpaths ") resolved_args.append(ctx.expand_location(arg, targets = all_data_targets)) + # Append args driven by string_flag build settings. Each mapped flag emits + # its arg template (with '{}' replaced by the flag value) only when the + # value is non-empty. This lets a value be set via `--//path/to:flag=/dir` + # instead of a hard-coded macro attribute at each call site. + for flag_target, arg_template in ctx.attr.string_flag_args.items(): + value = flag_target[BuildSettingInfo].value + if value: + resolved_args.append(arg_template.format(value)) + # Collect all plugin files and runfiles plugin_file_depsets = [] plugin_runfiles = ctx.runfiles() @@ -83,6 +93,13 @@ py_itf_plugin = rule( doc = "Additional CLI arguments. Supports $(location ...) referencing plugin_data targets.", default = [], ), + "string_flag_args": attr.label_keyed_string_dict( + doc = "Maps a string_flag build setting to a CLI arg template. " + + "'{}' is substituted with the flag value and the arg is " + + "emitted only when the value is non-empty.", + default = {}, + providers = [BuildSettingInfo], + ), "plugin_data": attr.label_list( doc = "Data files built for target configuration.", default = [], diff --git a/score/itf/plugins/BUILD b/score/itf/plugins/BUILD index 8a0d516..91b39f3 100644 --- a/score/itf/plugins/BUILD +++ b/score/itf/plugins/BUILD @@ -10,6 +10,7 @@ # # SPDX-License-Identifier: Apache-2.0 # ******************************************************************************* +load("@bazel_skylib//rules:common_settings.bzl", "bool_flag", "string_flag") load("@itf_pip//:requirements.bzl", "requirement") load("@rules_python//python:defs.bzl", "py_library") load("//bazel:py_itf_plugin.bzl", "py_itf_plugin") @@ -38,12 +39,42 @@ py_library( # ---- ITF plugin targets (used by py_itf_test symbolic macro) ---- +# Enable copying core dump files out of the container before teardown. +# Toggle with: bazel test --//score/itf/plugins:extract_core +bool_flag( + name = "extract_core", + build_setting_default = False, + visibility = ["//visibility:public"], +) + +config_setting( + name = "extract_core_enabled", + flag_values = {":extract_core": "True"}, + visibility = ["//visibility:public"], +) + +# Override the directory extracted core dumps are written to. When empty the +# docker plugin falls back to $TEST_UNDECLARED_OUTPUTS_DIR/cores. +# Set with: bazel test --//score/itf/plugins:core_output_dir=/abs/path +string_flag( + name = "core_output_dir", + build_setting_default = "", + visibility = ["//visibility:public"], +) + py_itf_plugin( name = "docker_plugin", enabled_plugins = [ "score.itf.plugins.docker", ], + plugin_args = select({ + ":extract_core_enabled": ["--extract-core"], + "//conditions:default": [], + }), py_library = ":docker", + string_flag_args = { + ":core_output_dir": "--core-output-dir={}", + }, visibility = ["//visibility:public"], ) diff --git a/score/itf/plugins/docker.py b/score/itf/plugins/docker.py index ca1cd8f..b8d7068 100644 --- a/score/itf/plugins/docker.py +++ b/score/itf/plugins/docker.py @@ -62,6 +62,21 @@ def pytest_addoption(parser): help="Directory to write extracted coverage files. " "Defaults to $TEST_UNDECLARED_OUTPUTS_DIR/sysroot or /tmp/sysroot.", ) + parser.addoption( + "--extract-core", + action="store_true", + default=False, + help="Copy core dump files from the container to the host before teardown.", + ) + parser.addoption( + "--core-output-dir", + default=os.path.join( + os.environ.get("TEST_UNDECLARED_OUTPUTS_DIR", "/tmp"), + "cores", + ), + help="Directory to write extracted core dump files. " + "Defaults to $TEST_UNDECLARED_OUTPUTS_DIR/cores or /tmp/cores.", + ) class DockerAsyncProcess(AsyncProcess): @@ -335,6 +350,34 @@ def _extract_coverage_from_container(target, output_base): logger.warning(f"Failed to extract {remote_path}", exc_info=True) +def _extract_core_from_container(target, output_base): + """Extract core dump files created inside the container.""" + logger.info(f"Attempting core extraction to {output_base}") + os.makedirs(output_base, exist_ok=True) + # Look for core files in typical locations, being specific to avoid false positives + exit_code, output = target.execute( + "(ls -1 /core* 2>/dev/null || true) && (ls -1 /opt/*/core* 2>/dev/null || true) && (ls -1 /root/core* 2>/dev/null || true) && (ls -1 /tmp/core* 2>/dev/null || true)" + ) + + core_paths = [line.strip() for line in output.decode().splitlines() if line.strip()] + logger.info(f"Found {len(core_paths)} core files: {core_paths}") + if not core_paths: + return + + for remote_path in core_paths: + local_path = os.path.join(output_base, remote_path.lstrip("/")) + if not os.path.realpath(local_path).startswith(os.path.realpath(output_base)): + logger.warning(f"Skipping path traversal attempt: {remote_path}") + continue + os.makedirs(os.path.dirname(local_path), exist_ok=True) + try: + logger.info(f"Extracting core from {remote_path} to {local_path}") + target.download(remote_path, local_path) + logger.info(f"Successfully extracted {remote_path}") + except Exception: + logger.warning(f"Failed to extract core file {remote_path}", exc_info=True) + + @pytest.fixture(scope=determine_target_scope) def target_init(request, _docker_configuration): print(_docker_configuration) @@ -397,6 +440,17 @@ def target_init(request, _docker_configuration): ) except Exception: logger.warning("Coverage extraction failed", exc_info=True) + try: + extract_core_enabled = request.config.getoption("extract_core") + logger.info(f"Core extraction enabled: {extract_core_enabled}") + if target is not None and extract_core_enabled: + logger.info(f"Extracting cores to {request.config.getoption('core_output_dir')}") + _extract_core_from_container( + target, + request.config.getoption("core_output_dir"), + ) + except Exception: + logger.warning("Core extraction failed", exc_info=True) try: try: container.stop(timeout=1) diff --git a/score/itf/plugins/qemu/__init__.py b/score/itf/plugins/qemu/__init__.py index 928b335..b6445f1 100644 --- a/score/itf/plugins/qemu/__init__.py +++ b/score/itf/plugins/qemu/__init__.py @@ -11,6 +11,7 @@ # SPDX-License-Identifier: Apache-2.0 # ******************************************************************************* import logging +import os import socket import pytest @@ -31,6 +32,55 @@ def pytest_addoption(parser): help="Path to json file with target configurations.", ) parser.addoption("--qemu-image", action="store", help="Path to a QEMU image") + parser.addoption( + "--extract-core", + action="store_true", + default=False, + help="Copy core dump files from the QEMU target to the host before teardown.", + ) + parser.addoption( + "--core-output-dir", + default=os.path.join( + os.environ.get("TEST_UNDECLARED_OUTPUTS_DIR", "/tmp"), + "cores", + ), + help="Directory to write extracted core dump files. " + "Defaults to $TEST_UNDECLARED_OUTPUTS_DIR/cores or /tmp/cores.", + ) + + +def _extract_cores_from_qemu(target, output_base): + """Extract core dump files from a QEMU (QNX) target via SSH/SFTP.""" + logger.info(f"Attempting core extraction to {output_base}") + os.makedirs(output_base, exist_ok=True) + # Search common core file locations (works on QNX and Linux guests). + _exit_code, output = target.execute( + "(ls -1 /core* 2>/dev/null || true)" + " && (ls -1 /opt/*/core* 2>/dev/null || true)" + " && (ls -1 /root/core* 2>/dev/null || true)" + " && (ls -1 /tmp/core* 2>/dev/null || true)" + " && (ls -1 /data/*/core* 2>/dev/null || true)" + " && (ls -1 /tmp/*.core /tmp/*.core.gz /var/*.core /var/*.core.gz 2>/dev/null || true)" + " && (ls -1 /opt/*/*.core /opt/*/*.core.gz /root/*.core /root/*.core.gz /data/*/*.core /data/*/*.core.gz 2>/dev/null || true)" + ) + + core_paths = [line.strip() for line in output.decode().splitlines() if line.strip()] + logger.info(f"Found {len(core_paths)} core files: {core_paths}") + if not core_paths: + return + + for remote_path in core_paths: + local_path = os.path.join(output_base, remote_path.lstrip("/")) + if not os.path.realpath(local_path).startswith(os.path.realpath(output_base)): + logger.warning(f"Skipping path traversal attempt: {remote_path}") + continue + os.makedirs(os.path.dirname(local_path), exist_ok=True) + try: + logger.info(f"Extracting core from {remote_path} to {local_path}") + target.download(remote_path, local_path) + logger.info(f"Successfully extracted {remote_path}") + except Exception: + logger.warning(f"Failed to extract core file {remote_path}", exc_info=True) @pytest.fixture(scope="session") @@ -55,3 +105,14 @@ def target_init(config, request, dlt): with qemu_target(config) as qemu: pre_tests_phase(qemu) yield qemu + try: + extract_core_enabled = request.config.getoption("extract_core") + logger.info(f"Core extraction enabled: {extract_core_enabled}") + if qemu is not None and extract_core_enabled: + logger.info(f"Extracting cores to {request.config.getoption('core_output_dir')}") + _extract_cores_from_qemu( + qemu, + request.config.getoption("core_output_dir"), + ) + except Exception: + logger.warning("Core extraction failed", exc_info=True)