junitxml: preserve setup/call captured output for interlaced reports

AUTHORS

@@ -199,6 +199,7 @@ Hugo van Kemenade
 Hui Wang (coldnight)
 Ian Bicking
 Ian Lesperance
+Ilya Abdolmanafi
 Ilya Konstantinov
 Ionuț Turturică
 Isaac Virshup

changelog/14078.bugfix.rst

@@ -0,0 +1 @@
+Fix JUnit XML output to include captured stdout/stderr from setup and call phases when reports are interlaced without xdist metadata.

src/_pytest/junitxml.py

@@ -83,6 +83,24 @@ def merge_family(left, right) -> None:
 families["xunit2"] = families["_base"]
 
 
+class _ReportOutput:
+    def __init__(self, report: TestReport, stdout: str, stderr: str, log: str) -> None:
+        self.capstdout = stdout
+        self.capstderr = stderr
+        self.caplog = log
+        self.passed = report.passed
+
+
+class _CapturedOutput:
+    def __init__(self) -> None:
+        self.out = ""
+        self.err = ""
+        self.log = ""
+        self.last_out = ""
+        self.last_err = ""
+        self.last_log = ""
+
+
 class _NodeReporter:
     def __init__(self, nodeid: str | TestReport, xml: LogXML) -> None:
         self.id = nodeid
@@ -156,7 +174,7 @@ def _add_simple(self, tag: str, message: str, data: str | None = None) -> None:
         node.text = bin_xml_escape(data)
         self.append(node)
 
-    def write_captured_output(self, report: TestReport) -> None:
+    def write_captured_output(self, report: TestReport | _ReportOutput) -> None:
         if not self.xml.log_passing_tests and report.passed:
             return
 
@@ -182,7 +200,9 @@ def write_captured_output(self, report: TestReport) -> None:
     def _prepare_content(self, content: str, header: str) -> str:
         return "\n".join([header.center(80, "-"), content, ""])
 
-    def _write_content(self, report: TestReport, content: str, jheader: str) -> None:
+    def _write_content(
+        self, report: TestReport | _ReportOutput, content: str, jheader: str
+    ) -> None:
         tag = ET.Element(jheader)
         tag.text = bin_xml_escape(content)
         self.append(tag)
@@ -484,11 +504,40 @@ def __init__(
         # List of reports that failed on call but teardown is pending.
         self.open_reports: list[TestReport] = []
         self.cnt_double_fail_tests = 0
+        self._captured_output: dict[tuple[str, object], _CapturedOutput] = {}
 
         # Replaces convenience family with real family.
         if self.family == "legacy":
             self.family = "xunit1"
 
+    def _report_key(self, report: TestReport) -> tuple[str, object]:
+        # Nodeid is stable across phases; avoid xdist-only worker_id/item_index,
+        # and include the worker node when present to disambiguate.
+        return report.nodeid, getattr(report, "node", None)
+
+    @staticmethod
+    def _diff_captured_output(previous: str, current: str) -> str:
+        if not current:
+            return ""
+        if current.startswith(previous):
+            return current[len(previous) :]
+        return current
+
+    def _update_captured_output(self, report: TestReport) -> _CapturedOutput:
+        key = self._report_key(report)
+        captured = self._captured_output.setdefault(key, _CapturedOutput())
+        captured.out += self._diff_captured_output(captured.last_out, report.capstdout)
+        captured.err += self._diff_captured_output(captured.last_err, report.capstderr)
+        captured.log += self._diff_captured_output(captured.last_log, report.caplog)
+        captured.last_out = report.capstdout
+        captured.last_err = report.capstderr
+        captured.last_log = report.caplog
+        return captured
+
+    def _report_output(self, report: TestReport) -> _ReportOutput:
+        captured = self._update_captured_output(report)
+        return _ReportOutput(report, captured.out, captured.err, captured.log)
+
     def finalize(self, report: TestReport) -> None:
         nodeid = getattr(report, "nodeid", report)
         # Local hack to handle xdist report order.
@@ -552,24 +601,19 @@ def pytest_runtest_logreport(self, report: TestReport) -> None:
             -> teardown node1
         """
         close_report = None
+        report_output = self._report_output(report)
         if report.passed:
             if report.when == "call":  # ignore setup/teardown
                 reporter = self._opentestcase(report)
                 reporter.append_pass(report)
         elif report.failed:
             if report.when == "teardown":
-                # The following vars are needed when xdist plugin is used.
-                report_wid = getattr(report, "worker_id", None)
-                report_ii = getattr(report, "item_index", None)
+                report_key = self._report_key(report)
                 close_report = next(
                     (
                         rep
                         for rep in self.open_reports
-                        if (
-                            rep.nodeid == report.nodeid
-                            and getattr(rep, "item_index", None) == report_ii
-                            and getattr(rep, "worker_id", None) == report_wid
-                        )
+                        if self._report_key(rep) == report_key
                     ),
                     None,
                 )
@@ -584,7 +628,7 @@ def pytest_runtest_logreport(self, report: TestReport) -> None:
                 reporter.append_failure(report)
                 self.open_reports.append(report)
                 if not self.log_passing_tests:
-                    reporter.write_captured_output(report)
+                    reporter.write_captured_output(report_output)
             else:
                 reporter.append_error(report)
         elif report.skipped:
@@ -593,25 +637,21 @@ def pytest_runtest_logreport(self, report: TestReport) -> None:
         self.update_testcase_duration(report)
         if report.when == "teardown":
             reporter = self._opentestcase(report)
-            reporter.write_captured_output(report)
+            reporter.write_captured_output(report_output)
 
             self.finalize(report)
-            report_wid = getattr(report, "worker_id", None)
-            report_ii = getattr(report, "item_index", None)
+            report_key = self._report_key(report)
             close_report = next(
                 (
                     rep
                     for rep in self.open_reports
-                    if (
-                        rep.nodeid == report.nodeid
-                        and getattr(rep, "item_index", None) == report_ii
-                        and getattr(rep, "worker_id", None) == report_wid
-                    )
+                    if self._report_key(rep) == report_key
                 ),
                 None,
             )
             if close_report:
                 self.open_reports.remove(close_report)
+            self._captured_output.pop(report_key, None)
 
     def update_testcase_duration(self, report: TestReport) -> None:
         """Accumulate total duration for nodeid from given report and update

testing/test_junitxml.py

@@ -9,6 +9,7 @@
 from typing import cast
 from typing import TYPE_CHECKING
 from xml.dom import minidom
+import xml.etree.ElementTree as ET
 
 import xmlschema
 
@@ -1749,6 +1750,121 @@ def test_esc(my_setup):
     assert "#x1B[31mred#x1B[m" in snode.text
 
 
+@pytest.mark.parametrize(
+    ("junit_logging", "expect_out", "expect_err"),
+    [
+        ("all", True, True),
+        ("system-out", True, False),
+        ("system-err", False, True),
+    ],
+)
+def test_interlaced_reports_capture_output(
+    pytester: Pytester,
+    junit_logging: str,
+    expect_out: bool,
+    expect_err: bool,
+) -> None:
+    pytester.makeconftest(
+        """
+        import pytest
+        from _pytest.runner import call_and_report
+
+        _reports = []
+
+        @pytest.hookimpl(tryfirst=True)
+        def pytest_runtest_protocol(item, nextitem):
+            item.ihook.pytest_runtest_logstart(
+                nodeid=item.nodeid, location=item.location
+            )
+            reports = [call_and_report(item, "setup", log=False)]
+            if reports[0].passed:
+                reports.append(call_and_report(item, "call", log=False))
+            reports.append(
+                call_and_report(item, "teardown", log=False, nextitem=nextitem)
+            )
+            item.ihook.pytest_runtest_logfinish(
+                nodeid=item.nodeid, location=item.location
+            )
+
+            _reports.append(reports)
+            if nextitem is not None:
+                return True
+
+            ihook = item.ihook
+            for reports in _reports:
+                ihook.pytest_runtest_logreport(report=reports[0])
+            for reports in _reports:
+                if len(reports) == 3:
+                    ihook.pytest_runtest_logreport(report=reports[1])
+            for reports in reversed(_reports):
+                ihook.pytest_runtest_logreport(report=reports[-1])
+            return True
+
+        @pytest.hookimpl(hookwrapper=True, tryfirst=True)
+        def pytest_runtest_logreport(report):
+            if report.when in ("setup", "call", "teardown"):
+                sections = []
+                for name, content in report.sections:
+                    if name.startswith("Captured "):
+                        if name.endswith(f" {report.when}"):
+                            sections.append((name, content))
+                    else:
+                        sections.append((name, content))
+                report.sections = sections
+            yield
+        """
+    )
+    pytester.makepyfile(
+        """
+        import sys
+        import pytest
+
+        @pytest.fixture
+        def setup_output(request):
+            print(f"SETUP_STDOUT_{request.node.name}")
+            sys.stderr.write(f"SETUP_STDERR_{request.node.name}\\n")
+
+        def test_one(setup_output):
+            print("CALL_STDOUT_test_one")
+            sys.stderr.write("CALL_STDERR_test_one\\n")
+
+        def test_two(setup_output):
+            print("CALL_STDOUT_test_two")
+            sys.stderr.write("CALL_STDERR_test_two\\n")
+        """
+    )
+
+    xml_path = pytester.path.joinpath("junit.xml")
+    result = pytester.runpytest(
+        f"--junitxml={xml_path}",
+        "--override-ini=junit_family=xunit1",
+        f"--override-ini=junit_logging={junit_logging}",
+    )
+    assert result.ret == 0
+
+    root = ET.parse(xml_path).getroot()
+    system_out_text = "".join(node.text or "" for node in root.iter("system-out"))
+    system_err_text = "".join(node.text or "" for node in root.iter("system-err"))
+
+    stdout_strings = [
+        "SETUP_STDOUT_test_one",
+        "CALL_STDOUT_test_one",
+        "SETUP_STDOUT_test_two",
+        "CALL_STDOUT_test_two",
+    ]
+    stderr_strings = [
+        "SETUP_STDERR_test_one",
+        "CALL_STDERR_test_one",
+        "SETUP_STDERR_test_two",
+        "CALL_STDERR_test_two",
+    ]
+
+    for expected in stdout_strings:
+        assert (expected in system_out_text) is expect_out
+    for expected in stderr_strings:
+        assert (expected in system_err_text) is expect_err
+
+
 @parametrize_families
 def test_logging_passing_tests_disabled_does_not_log_test_output(
     pytester: Pytester, run_and_parse: RunAndParse, xunit_family: str