Adding core dumps to user processes

tests/integration/incorrect_config_non_reporting/test_incorrect_config_non_reporting.py

@@ -12,7 +12,7 @@
 # *******************************************************************************
 import logging
 from tests.utils.testing_utils.run_until_file_deployed import run_until_file_deployed
-from tests.utils.testing_utils.setup_test import setup_test
+from tests.utils.testing_utils.setup_test import setup_test, download_core_dumps
 from tests.utils.testing_utils.test_results import assert_test_results
 from attribute_plugin import add_test_properties
 

tests/integration/smoke/smoke.py

@@ -22,7 +22,7 @@
     test_type="interface-test",
     derivation_technique="explorative-testing",
 )
-def test_smoke(target, setup_test, assert_test_results, remote_test_dir):
+def test_smoke(setup_test, target, assert_test_results, remote_test_dir):
     """
     Objective: Verifies the basic end-to-end lifecycle flow of the launch manager, including process startup, run target transitions, and execution state reporting.
 

tests/utils/plugins/localhost.py

@@ -19,6 +19,7 @@
 import time
 from pathlib import Path
 from typing import List
+import resource
 
 import pytest
 
@@ -127,6 +128,12 @@ def download(self, remote_path: str, local_path: str) -> None:
     def execute_async(
         self, binary_path: str, args=None, cwd: str = "/", **kwargs
     ) -> LocalAsyncProcess:
+        def _set_core_unlimited():
+            """Function to set core dump limit in the subshell."""
+            resource.setrlimit(
+                resource.RLIMIT_CORE, (resource.RLIM_INFINITY, resource.RLIM_INFINITY)
+            )
+
         cmd = ["fakeroot", "--", binary_path] + (args or [])
         cmd_logger = logging.getLogger(Path(binary_path).name)
         output_lines: List[str] = []
@@ -135,6 +142,7 @@ def execute_async(
             cmd,
             stdout=subprocess.PIPE,
             stderr=subprocess.STDOUT,
+            preexec_fn=_set_core_unlimited,
             cwd=cwd,
             start_new_session=True,  # Start under a new process group
         )

tests/utils/testing_utils/setup_test.py

@@ -13,15 +13,13 @@
 import pytest
 from pathlib import Path
 import logging
+from score.itf.core.target import Target
 
 
 logger = logging.getLogger(__name__)
 
 
-@pytest.fixture(autouse=True, scope="function")
-def setup_test(request, target):
-    """Sets up the test by uploading and extracting the test binaries tar"""
-
+def setup(request: pytest.FixtureRequest, target: Target):
     # silence logs coming from docker which uses urllib3
     logging.getLogger("urllib3").setLevel(logging.WARNING)
 
@@ -41,4 +39,63 @@ def setup_test(request, target):
     res, _ = target.execute(f"tar -xf {remote_tar} -C {extract_dir}")
     assert res == 0, f"Couldn't extract tar {remote_tar}"
 
+    res, out = target.execute("ulimit -c unlimited")
+
+    res, out = target.execute("cat /proc/sys/kernel/core_pattern")
+    logger.info(f"core_pattern: {out}")
+
+    logger.warning(f"ulimit res:\n{out.decode()}")
+    if not res == 0:
+        logger.warning(
+            f"Failed to set the ulimit, core dumps might not be created:\n{out}"
+        )
+    res, out = target.execute("ulimit -c")
+    logger.info(f"ulimit set:{out.decode()}")
+
     logger.info("Test case setup finished")
+
+
+def download_core_dumps(target: Target, remote_test_dir: Path, test_output_dir: Path):
+    """ """
+    logger.info("Trying to download core dumps")
+
+    # get core dumps
+    res, stdout = target.execute(f"find {remote_test_dir} -name 'core*' -type f")
+    if res != 0:
+        logger.error("Find command errored")
+        return
+
+    core_files = stdout.decode().strip().splitlines()
+    for remote_path in core_files:
+        remote_path = remote_path.strip()
+        if not remote_path:
+            continue
+        local_path = test_output_dir / Path(remote_path).name
+        try:
+            target.download(remote_path, str(local_path))
+            logger.info(f"Downloaded core dump: {remote_path} -> {local_path}")
+        except Exception as e:
+            logger.warning(f"Failed to download core dump {remote_path}: {e}")
+
+
+def teardown(target, remote_test_dir, test_output_dir):
+    """Tears down the test case"""
+
+    logger.info("Test case teardown started")
+    download_core_dumps(target, remote_test_dir, test_output_dir)
+
+
+@pytest.fixture(autouse=True, scope="function")
+def setup_test(
+    request: pytest.FixtureRequest,
+    target: Target,
+    remote_test_dir: Path,
+    test_output_dir: Path,
+):
+    """Sets up and tears down the test"""
+
+    setup(request, target)
+
+    yield None
+
+    teardown(target, remote_test_dir, test_output_dir)