Skip to content

junitxml: preserve setup/call captured output for interlaced reports #14082

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
CortexShadow wants to merge 8 commits into
base: main
Choose a base branch
from
Open

junitxml: preserve setup/call captured output for interlaced reports #14082

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
9020127
Add regression test for junitxml interlaced captured output
CortexShadow Jan 1, 2026
da5e637
Make junitxml interlacing independent of xdist-only report attrs
CortexShadow Jan 1, 2026
0f40c1c
Add Ilya Abdolmanafi to AUTHORS
CortexShadow Jan 1, 2026
6c89840
[pre-commit.ci] auto fixes from pre-commit.com hooks
pre-commit-ci[bot] Jan 1, 2026
56a6789
Add changelog entry for #14078
CortexShadow Jan 1, 2026
9361b9b
Handle junitxml interlaced output without assuming unique nodeid
CortexShadow Jan 4, 2026
5a88b61
Add junitxml test for nodeid collision safety
CortexShadow Jan 4, 2026
aae191c
[pre-commit.ci] auto fixes from pre-commit.com hooks
pre-commit-ci[bot] Jan 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions AUTHORS
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -199,6 +199,7 @@ Hugo van Kemenade
Hui Wang (coldnight)
Ian Bicking
Ian Lesperance
Ilya Abdolmanafi
Ilya Konstantinov
Ionuț Turturică
Isaac Virshup
Expand Down
1 change: 1 addition & 0 deletions changelog/14078.bugfix.rst
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Fix JUnit XML output to include captured stdout/stderr from setup and call phases when reports are interlaced without xdist metadata.
164 changes: 125 additions & 39 deletions src/_pytest/junitxml.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,6 +83,34 @@ def merge_family(left, right) -> None:
families["xunit2"] = families["_base"]


class _ReportOutput:
def __init__(
self, report: TestReport, stdout: str, stderr: str, log: str, stream_id: int
) -> None:
self.capstdout = stdout
self.capstderr = stderr
self.caplog = log
self.passed = report.passed
self.stream_id = stream_id


class _CapturedOutput:
def __init__(self) -> None:
self.out = ""
self.err = ""
self.log = ""
self.last_out = ""
self.last_err = ""
self.last_log = ""


class _OutputState:
def __init__(self) -> None:
self.current: _CapturedOutput | None = None
self.current_id: int | None = None
self.next_id = 0


class _NodeReporter:
def __init__(self, nodeid: str | TestReport, xml: LogXML) -> None:
self.id = nodeid
Expand Down Expand Up @@ -156,7 +184,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

Expand All @@ -182,7 +210,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)
Expand Down Expand Up @@ -481,14 +511,85 @@ def __init__(
self.node_reporters_ordered: list[_NodeReporter] = []
self.global_properties: list[tuple[str, str]] = []

# List of reports that failed on call but teardown is pending.
self.open_reports: list[TestReport] = []
# Reports that failed on call but teardown is pending.
self.open_reports: dict[tuple[tuple[str, object], int], TestReport] = {}
self.cnt_double_fail_tests = 0
self._output_state: dict[tuple[str, object], _OutputState] = {}
self._ambiguous_output: set[tuple[str, object]] = set()

# Replaces convenience family with real family.
if self.family == "legacy":
self.family = "xunit1"

def _report_key(self, report: TestReport) -> tuple[str, object]:
# Nodeid is not guaranteed unique; avoid xdist-only worker_id/item_index,
# and include the worker node when present to disambiguate.
return report.nodeid, getattr(report, "node", None)
Copy link
Member

@RonnyPfannschmidt RonnyPfannschmidt Jan 4, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

theres various 34d party plugins that make nodeid non-unique

these are key parts why xdist does the messy nodeindex/list stuff

Copy link
Contributor Author

@CortexShadow CortexShadow Jan 4, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for pointing this out.

I’ve updated the handling to avoid assuming unique nodeids and to handle collisions conservatively, and added a regression test for the collision case.

Copy link

@tymonx tymonx Jan 4, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this could be simplify and backward compatible by hashing nodeid and optionally with worker_id and index_item attributes if they are present.

def _report_key(self, report: TestReport) -> str:
    """Get unique key from test report."""
    # Use hashlib.sha1 or simple concatenating attributes as string is enough?
    return f"{report.nodeid}{getattr(report, 'worker_id', '')}{getattr(report, 'index_item', '')}"

Then interlaced logic will work in non-xdist (cocotb use case) and with xdist.

By default, pytest is generating unique nodeid by using file path, function name and parametrized arguments []. If some 3rd party plugin is altering/generating nodeid that is not unique, they can use these additional xdist attributes (worker_id, index_item) or this can be classified as bug in 3rd party plugin. The whole point of having id is to have some kind of unique identifier.


@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 _mark_output_ambiguous(self, key: tuple[str, object]) -> None:
self._ambiguous_output.add(key)
self._output_state.pop(key, None)
for report_key in list(self.open_reports):
if report_key[0] == key:
del self.open_reports[report_key]

def _current_output_stream(
self, report: TestReport
) -> tuple[_CapturedOutput, int] | None:
key = self._report_key(report)
if key in self._ambiguous_output:
return None
state = self._output_state.setdefault(key, _OutputState())
if report.when == "setup":
if state.current is not None:
self._mark_output_ambiguous(key)
return None
state.current = _CapturedOutput()
state.current_id = state.next_id
state.next_id += 1
elif state.current is None:
state.current = _CapturedOutput()
state.current_id = state.next_id
state.next_id += 1
assert state.current is not None
assert state.current_id is not None
return state.current, state.current_id

def _finish_output_stream(self, report: TestReport) -> None:
if report.when != "teardown":
return
key = self._report_key(report)
if key in self._ambiguous_output:
return
state = self._output_state.get(key)
if state is None:
return
state.current = None
state.current_id = None

def _report_output(self, report: TestReport) -> _ReportOutput | None:
stream = self._current_output_stream(report)
if stream is None:
return None
captured, stream_id = stream
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 _ReportOutput(
report, captured.out, captured.err, captured.log, stream_id
)

def finalize(self, report: TestReport) -> None:
nodeid = getattr(report, "nodeid", report)
# Local hack to handle xdist report order.
Expand Down Expand Up @@ -552,28 +653,21 @@ 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)
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
)
),
None,
)
if close_report:
if report_output is not None:
report_key = self._report_key(report)
close_report = self.open_reports.pop(
(report_key, report_output.stream_id), None
)
else:
close_report = None
if close_report is not None:
# We need to open new testcase in case we have failure in
# call and error in teardown in order to follow junit
# schema.
Expand All @@ -582,9 +676,12 @@ def pytest_runtest_logreport(self, report: TestReport) -> None:
reporter = self._opentestcase(report)
if report.when == "call":
reporter.append_failure(report)
self.open_reports.append(report)
if report_output is not None:
report_key = self._report_key(report)
self.open_reports[(report_key, report_output.stream_id)] = report
if not self.log_passing_tests:
reporter.write_captured_output(report)
if report_output is not None:
reporter.write_captured_output(report_output)
else:
reporter.append_error(report)
elif report.skipped:
Expand All @@ -593,25 +690,14 @@ 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)
if report_output is not None:
reporter.write_captured_output(report_output)

self.finalize(report)
report_wid = getattr(report, "worker_id", None)
report_ii = getattr(report, "item_index", None)
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
)
),
None,
)
if close_report:
self.open_reports.remove(close_report)
if report_output is not None:
report_key = self._report_key(report)
self.open_reports.pop((report_key, report_output.stream_id), None)
self._finish_output_stream(report)

def update_testcase_duration(self, report: TestReport) -> None:
"""Accumulate total duration for nodeid from given report and update
Expand Down
Loading