Add core dump extraction from container in docker plugin

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 = [],

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_dump
+bool_flag(
+    name = "extract_core_dump",
+    build_setting_default = False,
+    visibility = ["//visibility:public"],
+)
+
+config_setting(
+    name = "extract_core_dump_enabled",
+    flag_values = {":extract_core_dump": "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_dumps_output_dir=/abs/path
+string_flag(
+    name = "core_dumps_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_dump_enabled": ["--extract-core-dumps"],
+        "//conditions:default": [],
+    }),
     py_library = ":docker",
+    string_flag_args = {
+        ":core_dumps_output_dir": "--core-dumps-output-dir={}",
+    },
     visibility = ["//visibility:public"],
 )
 

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-dumps",
+        action="store_true",
+        default=False,
+        help="Copy core dump files from the container to the host before teardown.",
+    )
+    parser.addoption(
+        "--core-dumps-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_dumps_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_dumps")
+            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_dumps_output_dir')}")
+                _extract_core_dumps_from_container(
+                    target,
+                    request.config.getoption("core_dumps_output_dir"),
+                )
+        except Exception:
+            logger.warning("Core extraction failed", exc_info=True)
         try:
             try:
                 container.stop(timeout=1)

score/itf/plugins/qemu/__init__.py

@@ -80,6 +80,55 @@ def pytest_addoption(parser):
         help="Path to a QEMU disk image (qcow2, wic, or img). "
         "An ephemeral overlay is created so the original image is not modified.",
     )
+    parser.addoption(
+        "--extract-core-dump",
+        action="store_true",
+        default=False,
+        help="Copy core dump files from the QEMU target to the host before teardown.",
+    )
+    parser.addoption(
+        "--core-dump-output-dir",
+        default=os.path.join(
+            os.environ.get("TEST_UNDECLARED_OUTPUTS_DIR", "/tmp"),
+            "coredumps",
+        ),
+        help="Directory to write extracted core dump files. "
+        "Defaults to $TEST_UNDECLARED_OUTPUTS_DIR/coredumps or /tmp/coredumps.",
+    )
+
+
+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 dumps 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)"
+    )
+
+    coredumps_paths = [line.strip() for line in output.decode().splitlines() if line.strip()]
+    logger.info(f"Found {len(coredumps_paths)} core files: {coredumps_paths}")
+    if not coredumps_paths:
+        return
+
+    for remote_path in coredumps_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")
@@ -127,6 +176,19 @@ def target_init(config, request, dlt):
         ) as qemu:
             pre_tests_phase(qemu)
             yield qemu
+            try:
+                extract_core_enabled = request.config.getoption("extract_core_dump")
+                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_dump_output_dir')}"
+                    )
+                    _extract_cores_from_qemu(
+                        qemu,
+                        request.config.getoption("core_dump_output_dir"),
+                    )
+            except Exception:
+                logger.warning("Core extraction failed", exc_info=True)
     finally:
         if overlay_path and os.path.exists(overlay_path):
             os.unlink(overlay_path)