fix: generalize JUnit XML injection for all early-exit failures

conftest.py

@@ -51,6 +51,7 @@
 from utilities.junit_ai_utils import enrich_junit_xml, setup_ai_analysis
 from utilities.logger import setup_logging
 from utilities.pytest_utils import (
+    _inject_failure_junit,
     assert_incremental_classes_fully_collected,
     config_default_storage_class,
     deploy_run_in_progress_config_map,
@@ -870,41 +871,47 @@ def pytest_collection_finish(session):
 
 
 def pytest_sessionfinish(session, exitstatus):
-    shutil.rmtree(path=session.config.option.basetemp, ignore_errors=True)
-    if not skip_if_pytest_flags_exists(pytest_config=session.config):
-        admin_client = utilities.cluster.cache_admin_client()
-        run_in_progress_config_map(client=admin_client).clean_up()
-        deploy_run_in_progress_namespace(client=admin_client).clean_up()
-
-    reporter = session.config.pluginmanager.get_plugin("terminalreporter")
-    reporter.summary_stats()
-    if session.config.getoption("--data-collector"):
-        db = Database(base_dir=session.config.getoption("--data-collector-output-dir"))
-        file_path = db.database_file_path
-        LOGGER.info(f"Removing database file path {file_path}")
-        os.remove(file_path)
-    # clean up the empty folders
-    collector_directory = py_config["data_collector"]["data_collector_base_directory"]
-    if os.path.exists(collector_directory):
-        for root, dirs, files in os.walk(collector_directory, topdown=False):
-            for _dir in dirs:
-                dir_path = os.path.join(root, _dir)
-                if not os.listdir(dir_path):
-                    shutil.rmtree(dir_path, ignore_errors=True)
-
-    # Enrich JUnit XML with AI analysis after all tests complete.
-    # Source: https://github.com/myk-org/jenkins-job-insight/blob/main/examples/pytest-junitxml/conftest_junit_ai.py
-    if session.config.option.analyze_with_ai:
-        if exitstatus == 0:
-            LOGGER.info("No test failures (exit code %d), skipping AI analysis", exitstatus)
+    try:
+        shutil.rmtree(path=session.config.option.basetemp, ignore_errors=True)
+        if not skip_if_pytest_flags_exists(pytest_config=session.config):
+            admin_client = utilities.cluster.cache_admin_client()
+            run_in_progress_config_map(client=admin_client).clean_up()
+            deploy_run_in_progress_namespace(client=admin_client).clean_up()
+
+        reporter = session.config.pluginmanager.get_plugin("terminalreporter")
+        reporter.summary_stats()
+        if session.config.getoption("--data-collector"):
+            db = Database(base_dir=session.config.getoption("--data-collector-output-dir"))
+            file_path = db.database_file_path
+            LOGGER.info(f"Removing database file path {file_path}")
+            os.remove(file_path)
+        # clean up the empty folders
+        collector_directory = py_config["data_collector"]["data_collector_base_directory"]
+        if os.path.exists(collector_directory):
+            for root, dirs, files in os.walk(collector_directory, topdown=False):
+                for _dir in dirs:
+                    dir_path = os.path.join(root, _dir)
+                    if not os.listdir(dir_path):
+                        shutil.rmtree(dir_path, ignore_errors=True)
+
+        # Enrich JUnit XML with AI analysis after all tests complete.
+        # Source: https://github.com/myk-org/jenkins-job-insight/blob/main/examples/pytest-junitxml/conftest_junit_ai.py
+        if session.config.option.analyze_with_ai:
+            if exitstatus == 0:
+                LOGGER.info("No test failures (exit code %d), skipping AI analysis", exitstatus)
 
-        else:
-            try:
-                enrich_junit_xml(session)
-            except Exception:
-                LOGGER.exception("Failed to enrich JUnit XML, original preserved")
+            else:
+                try:
+                    enrich_junit_xml(session)
+                except Exception:
+                    LOGGER.exception("Failed to enrich JUnit XML, original preserved")
+    finally:
+        try:
+            _inject_failure_junit(session=session)
+        except Exception:
+            LOGGER.exception("Failed to inject failure into JUnit XML")
 
-    session.config.option.log_listener.stop()
+        session.config.option.log_listener.stop()
 
 
 def get_all_node_markers(node: Node) -> list[str]:

utilities/pytest_utils.py

@@ -8,8 +8,20 @@
 import shutil
 import socket
 import sys
+import tempfile
 from collections import defaultdict
-from typing import Any
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from typing import TypedDict
+
+    class _FailureInfoDict(TypedDict):
+        message: str
+        log_message: str
+        return_code: int
+
+
+from xml.etree import ElementTree
 
 import pytest
 from kubernetes.dynamic import DynamicClient
@@ -52,6 +64,81 @@
 LOGGER = logging.getLogger(__name__)
 
 
+_failure_info: _FailureInfoDict | None = None
+
+
+def _inject_failure_junit(session: pytest.Session) -> None:
+    """Inject a synthetic error testcase into JUnit XML for pytest exit failures.
+
+    When exit_pytest_execution aborts the session, the exit reason may not appear
+    in the JUnit XML report. CI systems can miss the failure — either because the
+    report is empty (no tests ran) or because the exit reason is not captured as a
+    testcase. This function re-opens the XML file after LogXML writes it and adds
+    a synthetic error testcase to ensure the exit failure is always reported.
+
+    Note:
+        Must be called during pytest_sessionfinish, after LogXML has written the
+        XML file. Pluggy calls hooks in LIFO (last-in-first-out) registration
+        order: conftest is registered before the junitxml plugin, so the
+        conftest hook runs after LogXML has written its output.
+
+    Args:
+        session: The pytest session object, used to access the junitxml file path.
+    """
+    if _failure_info is None:
+        return
+
+    xml_path = getattr(session.config.option, "xmlpath", None)
+    if not xml_path or not os.path.exists(xml_path):
+        LOGGER.info("No JUnit XML file found, skipping synthetic testcase injection")
+        return
+
+    return_code = _failure_info["return_code"]
+    log_message = _failure_info["log_message"]
+
+    sanitized_name = re.sub(r"[^a-z0-9_]", "", _failure_info["message"].lower().replace(" ", "_"))
+    sanitized_name = re.sub(r"_+", "_", sanitized_name).strip("_")[:80]
+    name = sanitized_name or "execution_failure"
+
+    tree = ElementTree.parse(xml_path)
+    root = tree.getroot()
+
+    testsuite = root.find("testsuite")
+    if testsuite is None:
+        LOGGER.warning("No <testsuite> element found in JUnit XML, skipping injection")
+        return
+
+    testcase = ElementTree.SubElement(
+        testsuite,
+        "testcase",
+        attrib={"classname": "pytest_exit", "name": name, "time": "0.000"},
+    )
+    error_node = ElementTree.SubElement(
+        testcase,
+        "error",
+        attrib={"message": f"Pytest execution failed (exit code: {return_code})"},
+    )
+    # Strip XML 1.0 illegal control characters — ElementTree passes them through, corrupting the file.
+    error_node.text = re.sub(r"[^\x09\x0A\x0D\x20-\uD7FF\uE000-\uFFFD\U00010000-\U0010FFFF]", "\ufffd", log_message)
+
+    current_errors = int(testsuite.get("errors", "0"))
+    testsuite.set("errors", str(current_errors + 1))
+    current_tests = int(testsuite.get("tests", "0"))
+    testsuite.set("tests", str(current_tests + 1))
+
+    xml_dir = os.path.dirname(os.path.abspath(xml_path))
+    fd, tmp_path = tempfile.mkstemp(dir=xml_dir, suffix=".xml")
+    try:
+        with os.fdopen(fd, mode="w") as tmp_file:
+            tree.write(tmp_file, encoding="unicode", xml_declaration=True)
+        os.replace(tmp_path, xml_path)
+    except BaseException:
+        os.unlink(tmp_path)
+        raise
+
+    LOGGER.info(f"Injected synthetic failure testcase into JUnit XML (exit code: {return_code})")
+
+
 def get_base_matrix_name(matrix_name):
     match = re.match(r".*?(.*?_matrix)_(?:.*_matrix)+", matrix_name)
     if match:
@@ -424,6 +511,11 @@ def exit_pytest_execution(
         junitxml_property (pytest plugin): record_testsuite_property
         message (str): Message to log in an error file. If not provided, `log_message` will be used.
         admin_client (DynamicClient): cluster admin client
+
+    Note:
+        Records the failure details in a module-level store so that
+        _inject_failure_junit can emit a synthetic JUnit XML testcase
+        during pytest_sessionfinish.
     """
     target_location = os.path.join(get_data_collector_base_directory(), "utilities", "pytest_exit_errors")
     # collect must-gather for past 5 minutes:
@@ -443,6 +535,14 @@ def exit_pytest_execution(
         )
     if junitxml_property:
         junitxml_property(name="exit_code", value=return_code)
+
+    global _failure_info
+    _failure_info = {
+        "message": message or log_message,
+        "log_message": log_message,
+        "return_code": return_code,
+    }
+
     pytest.exit(reason=log_message, returncode=return_code)