From 8e71bd3dddda9d847289f71c6932434e3e04c1aa Mon Sep 17 00:00:00 2001 From: rnetser Date: Thu, 16 Jul 2026 16:47:57 +0300 Subject: [PATCH] fix: inject synthetic JUnit XML testcase for pytest exit failures When exit_pytest_execution aborts the session, the exit reason may not appear in the JUnit XML report. Add _inject_failure_junit to create a synthetic error testcase in the XML, ensuring CI systems always see the failure. - Record failure details in _failure_info before pytest.exit() - Inject synthetic testcase into JUnit XML during pytest_sessionfinish - Use atomic write (tempfile + os.replace) to preserve XML on failure - Wrap in try/finally to guarantee injection despite teardown errors Signed-off-by: Ruth Netser Co-authored-by: PI (claude-opus-4-6-1m) Signed-off-by: rnetser --- conftest.py | 73 +++-- utilities/pytest_utils.py | 102 ++++++- utilities/unittests/test_pytest_utils.py | 366 ++++++++++++++++++++++- 3 files changed, 502 insertions(+), 39 deletions(-) diff --git a/conftest.py b/conftest.py index 62b8a08a25..3817d90bb9 100644 --- a/conftest.py +++ b/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]: diff --git a/utilities/pytest_utils.py b/utilities/pytest_utils.py index c005905463..f17c4dc1f1 100644 --- a/utilities/pytest_utils.py +++ b/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 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) diff --git a/utilities/unittests/test_pytest_utils.py b/utilities/unittests/test_pytest_utils.py index 5113232602..7aa7dbdc57 100644 --- a/utilities/unittests/test_pytest_utils.py +++ b/utilities/unittests/test_pytest_utils.py @@ -3,13 +3,14 @@ """Unit tests for pytest_utils module""" from unittest.mock import MagicMock, mock_open, patch +from xml.etree import ElementTree import pytest import utilities.constants # Circular dependencies are already mocked in conftest.py -from utilities import pytest_utils +from utilities import pytest_utils as pytest_utils_module from utilities.constants.architecture import ( AMD_64, ARM_64, @@ -488,9 +489,9 @@ def test_config_default_storage_class_both_options_valid(self): config_default_storage_class(mock_session) - assert pytest_utils.py_config["default_storage_class"] == "sc-1" - assert pytest_utils.py_config["default_volume_mode"] == "Filesystem" - assert pytest_utils.py_config["default_access_mode"] == "ReadWriteOnce" + assert pytest_utils_module.py_config["default_storage_class"] == "sc-1" + assert pytest_utils_module.py_config["default_volume_mode"] == "Filesystem" + assert pytest_utils_module.py_config["default_access_mode"] == "ReadWriteOnce" @patch( "utilities.pytest_utils.py_config", @@ -511,7 +512,7 @@ def test_config_default_storage_class_same_as_global(self): config_default_storage_class(mock_session) - assert pytest_utils.py_config["default_storage_class"] == "original-sc" + assert pytest_utils_module.py_config["default_storage_class"] == "original-sc" class TestValidateStorageClassOptions: @@ -2813,3 +2814,358 @@ def test_no_deselection_when_no_multiarch_tests(self): assert result == [item_other] config.hook.pytest_deselected.assert_not_called() + + +class TestInjectFailureJunit: + """Test cases for _inject_failure_junit (private) and _failure_info mechanism""" + + def setup_method(self): + pytest_utils_module._failure_info = None + + def teardown_method(self): + pytest_utils_module._failure_info = None + + @patch("utilities.pytest_utils.ElementTree") + def test_no_op_when_no_failure(self, mock_element_tree): + """Test _inject_failure_junit does nothing when no failure was recorded.""" + mock_session = MagicMock() + mock_session.config.option.xmlpath = None + pytest_utils_module._inject_failure_junit(session=mock_session) + mock_element_tree.parse.assert_not_called() + + @patch("utilities.pytest_utils.ElementTree") + def test_no_op_when_no_xmlpath(self, mock_element_tree): + """Test _inject_failure_junit does nothing when no junitxml path is configured.""" + pytest_utils_module._failure_info = { + "message": "Test failure", + "log_message": "Detailed failure", + "return_code": 99, + } + + mock_session = MagicMock() + mock_session.config.option.xmlpath = None + + pytest_utils_module._inject_failure_junit(session=mock_session) + mock_element_tree.parse.assert_not_called() + + def test_no_op_when_no_testsuite(self, tmp_path): + """Test _inject_failure_junit skips injection when XML has no testsuite element.""" + pytest_utils_module._failure_info = { + "message": "Test failure", + "log_message": "Detailed failure", + "return_code": 99, + } + + xml_path = tmp_path / "test-results.xml" + xml_path.write_text('') + + mock_session = MagicMock() + mock_session.config.option.xmlpath = str(xml_path) + + pytest_utils_module._inject_failure_junit(session=mock_session) + + tree = ElementTree.parse(xml_path) + assert tree.getroot().find("testsuite") is None + + def test_injects_synthetic_testcase(self, tmp_path): + """Test _inject_failure_junit creates synthetic error testcase in JUnit XML.""" + pytest_utils_module._failure_info = { + "message": "Cluster sanity failed", + "log_message": "Detailed cluster sanity failure message", + "return_code": 99, + } + + xml_path = tmp_path / "test-results.xml" + xml_path.write_text( + '' + '' + '' + "" + ) + + mock_session = MagicMock() + mock_session.config.option.xmlpath = str(xml_path) + + pytest_utils_module._inject_failure_junit(session=mock_session) + + tree = ElementTree.parse(xml_path) + root = tree.getroot() + testsuite = root.find("testsuite") + testcase = testsuite.find("testcase") + assert testcase is not None, "Synthetic testcase not found in XML" + assert testcase.get("classname") == "pytest_exit" + assert testcase.get("name") == "cluster_sanity_failed" + error_elem = testcase.find("error") + assert error_elem is not None, "Error element not found in testcase" + assert "exit code: 99" in error_elem.get("message") + assert "Detailed cluster sanity failure message" in error_elem.text + assert testsuite.get("errors") == "1" + assert testsuite.get("tests") == "1" + + def test_injects_into_non_empty_suite(self, tmp_path): + """Test _inject_failure_junit appends synthetic testcase to suite with existing tests.""" + pytest_utils_module._failure_info = { + "message": "Storage class failure", + "log_message": "Failed to set default storage class", + "return_code": 99, + } + + xml_path = tmp_path / "test-results.xml" + xml_path.write_text( + '' + '' + '' + '' + '' + 'assert False' + "" + '' + "" + "" + ) + + mock_session = MagicMock() + mock_session.config.option.xmlpath = str(xml_path) + + pytest_utils_module._inject_failure_junit(session=mock_session) + + tree = ElementTree.parse(xml_path) + root = tree.getroot() + testsuite = root.find("testsuite") + testcases = testsuite.findall("testcase") + assert len(testcases) == 4, f"Expected 4 testcases, got {len(testcases)}" + synthetic = testcases[-1] + assert synthetic.get("classname") == "pytest_exit" + assert synthetic.get("name") == "storage_class_failure" + assert testsuite.get("errors") == "1" + assert testsuite.get("tests") == "4" + assert testsuite.get("failures") == "1" + + @patch("utilities.pytest_utils.pytest.exit") + @patch("utilities.pytest_utils.get_data_collector_base_directory") + def test_exit_pytest_execution_stores_failure_info(self, mock_get_base_dir, mock_pytest_exit): + """Test exit_pytest_execution stores failure info for JUnit XML injection.""" + mock_get_base_dir.return_value = "/tmp/test" + mock_admin_client = MagicMock() + + exit_pytest_execution( + log_message="Storage check failed", + return_code=99, + message="Cluster sanity checks failed.", + admin_client=mock_admin_client, + ) + + assert pytest_utils_module._failure_info is not None + assert pytest_utils_module._failure_info["message"] == "Cluster sanity checks failed." + assert pytest_utils_module._failure_info["log_message"] == "Storage check failed" + assert pytest_utils_module._failure_info["return_code"] == 99 + mock_pytest_exit.assert_called_once_with(reason="Storage check failed", returncode=99) + + @patch("utilities.pytest_utils.pytest.exit") + @patch("utilities.pytest_utils.get_data_collector_base_directory") + def test_exit_pytest_execution_uses_log_message_when_no_message(self, mock_get_base_dir, mock_pytest_exit): + """Test exit_pytest_execution uses log_message as message when message is None.""" + mock_get_base_dir.return_value = "/tmp/test" + mock_admin_client = MagicMock() + + exit_pytest_execution( + log_message="Network sanity failed", + return_code=91, + admin_client=mock_admin_client, + ) + + assert pytest_utils_module._failure_info is not None + assert pytest_utils_module._failure_info["message"] == "Network sanity failed" + assert pytest_utils_module._failure_info["return_code"] == 91 + mock_pytest_exit.assert_called_once_with(reason="Network sanity failed", returncode=91) + + def test_sanitized_name_collapses_underscores(self, tmp_path): + """Test _inject_failure_junit collapses consecutive underscores in testcase name.""" + pytest_utils_module._failure_info = { + "message": "Cluster: sanity -- failed!", + "log_message": "Detailed failure", + "return_code": 99, + } + + xml_path = tmp_path / "test-results.xml" + xml_path.write_text( + '' + '' + '' + "" + ) + + mock_session = MagicMock() + mock_session.config.option.xmlpath = str(xml_path) + + pytest_utils_module._inject_failure_junit(session=mock_session) + + tree = ElementTree.parse(xml_path) + testsuite = tree.getroot().find("testsuite") + testcase = testsuite.find("testcase") + assert testcase.get("name") == "cluster_sanity_failed" + + def test_sanitized_name_fallback(self, tmp_path): + """Test _inject_failure_junit uses 'execution_failure' for messages with only special chars.""" + pytest_utils_module._failure_info = { + "message": "!@#$%^&*()", + "log_message": "Special chars only", + "return_code": 99, + } + + xml_path = tmp_path / "test-results.xml" + xml_path.write_text( + '' + '' + '' + "" + ) + + mock_session = MagicMock() + mock_session.config.option.xmlpath = str(xml_path) + + pytest_utils_module._inject_failure_junit(session=mock_session) + + tree = ElementTree.parse(xml_path) + testsuite = tree.getroot().find("testsuite") + testcase = testsuite.find("testcase") + assert testcase.get("name") == "execution_failure" + + def test_error_text_escapes_xml_chars(self, tmp_path): + """Test _inject_failure_junit escapes XML special characters in error text.""" + pytest_utils_module._failure_info = { + "message": "XML test", + "log_message": 'Failed with & "quotes"', + "return_code": 99, + } + + xml_path = tmp_path / "test-results.xml" + xml_path.write_text( + '' + '' + '' + "" + ) + + mock_session = MagicMock() + mock_session.config.option.xmlpath = str(xml_path) + + pytest_utils_module._inject_failure_junit(session=mock_session) + + tree = ElementTree.parse(xml_path) + testsuite = tree.getroot().find("testsuite") + testcase = testsuite.find("testcase") + error_elem = testcase.find("error") + # ElementTree handles escaping on write and unescaping on parse, + # so .text contains the original unescaped characters. + assert "" in error_elem.text + assert "&" in error_elem.text + + def test_control_chars_sanitized(self, tmp_path): + """Test _inject_failure_junit strips XML-illegal control characters from error text.""" + pytest_utils_module._failure_info = { + "message": "Control char test", + "log_message": "Failed\x07with\x08control\x00chars", + "return_code": 99, + } + + xml_path = tmp_path / "test-results.xml" + xml_path.write_text( + '' + '' + '' + "" + ) + + mock_session = MagicMock() + mock_session.config.option.xmlpath = str(xml_path) + + pytest_utils_module._inject_failure_junit(session=mock_session) + + # Verify the XML is parseable (control chars would break parsing) + tree = ElementTree.parse(xml_path) + testsuite = tree.getroot().find("testsuite") + testcase = testsuite.find("testcase") + error_elem = testcase.find("error") + assert error_elem.text is not None + assert "Failed" in error_elem.text + assert "control" in error_elem.text + # Control chars replaced with Unicode replacement character + assert "\x07" not in error_elem.text + assert "\x08" not in error_elem.text + assert "\x00" not in error_elem.text + + def test_injection_runs_despite_earlier_teardown_failure(self, tmp_path): + """Test _inject_failure_junit executes even when prior teardown raises. + + Simulates the conftest.py finally-block pattern: earlier teardown code + raises an exception, but inject still runs and writes the synthetic testcase. + """ + pytest_utils_module._failure_info = { + "message": "Cluster sanity failed", + "log_message": "Sanity check failure details", + "return_code": 99, + } + + xml_path = tmp_path / "test-results.xml" + xml_path.write_text( + '' + '' + '' + "" + ) + + mock_session = MagicMock() + mock_session.config.option.xmlpath = str(xml_path) + + # Simulate: earlier teardown raises, then finally block runs injection + with pytest.raises(RuntimeError, match="Earlier teardown failed"): + try: + raise RuntimeError("Earlier teardown failed") + finally: + pytest_utils_module._inject_failure_junit(session=mock_session) + + tree = ElementTree.parse(xml_path) + testsuite = tree.getroot().find("testsuite") + testcase = testsuite.find("testcase") + assert testcase is not None, "Synthetic testcase must be injected despite earlier failure" + assert testcase.get("classname") == "pytest_exit" + assert testsuite.get("errors") == "1" + + def test_atomic_write_preserves_original_on_failure(self, tmp_path): + """Test _inject_failure_junit preserves the original XML if write fails.""" + pytest_utils_module._failure_info = { + "message": "Test failure", + "log_message": "Details", + "return_code": 99, + } + + xml_path = tmp_path / "test-results.xml" + original_content = ( + '' + '' + '' + "" + ) + xml_path.write_text(original_content) + + mock_session = MagicMock() + mock_session.config.option.xmlpath = str(xml_path) + + # Make os.replace fail to simulate atomic write failure + with patch("utilities.pytest_utils.os.replace", side_effect=OSError("disk full")): + with pytest.raises(OSError, match="disk full"): + pytest_utils_module._inject_failure_junit(session=mock_session) + + # Original file should be preserved + assert xml_path.exists() + content = xml_path.read_text() + assert "pytest_exit" not in content, "Original XML should not be modified on write failure"