From f11436ad4672e2f810f1f3a83b033cc121c5be36 Mon Sep 17 00:00:00 2001 From: Ralf Juengling Date: Mon, 12 Oct 2026 14:08:20 -0700 Subject: [PATCH] ci: run cuda.core tests in heap-bounded batches, merge JUnit XML MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The init_cuda teardown's per-test gc.collect() (added by #2484) gets superlinearly slower as a pytest session proceeds: pytest-repeat's --count=N expands the collected item list N-fold up front, and the executed-item bookkeeping (TestReport objects, and since the JUnit logger was added, its XML nodes) accumulates for the life of the session. gc.collect() walks that whole live heap on every call, so nightly's 5 repeats measured at ~10x one repeat, not ~5x (see the "cuda.core test time" investigation referenced from #3041 and #3045). #3041 addressed the first term with one gc.freeze() after collection. This change addresses the session-length growth directly: ci/tools/ pytest_batched.py collects the full, already---count-expanded item list once via `pytest --collect-only`, then runs it across several pytest subprocesses of at most --batch-size items each. Each subprocess starts with an empty heap, so neither accumulation term grows past one batch, regardless of how large --count or the suite gets. The per-batch JUnit XML files are merged back into a single file at the original --junitxml path, so the script is a drop-in replacement for the plain pytest invocation it replaces in ci/tools/run-tests: same CLI, same final junit-core.xml (verified byte-for-byte equal `tests`/`failures`/`testcase` counts against an unbatched run of the same suite), same aggregate exit code. --runner-prefix forwards SANITIZER_CMD to each batch. Scoped to the "Run cuda.core tests" step only (ci/tools/run-tests), since that is the step the investigation identified as the bottleneck; bindings/pathfinder/cython invocations are untouched. ci/tools/tests/test_pytest_batched.py covers the pure helpers (ID parsing, chunking, path stripping, JUnit merging) and five end-to-end subprocess scenarios, including the two non-obvious correctness requirements this script depends on: - `--collect-only -q`'s flat ID format needs a verbosity this script controls exactly; a caller's `-v` (CI always passes one) would otherwise partially cancel the `-q` pytest adds internally. - PYTEST_ADDOPTS (e.g. --count=N) must stay active for the batch runs, not just collection: a repeated item's node ID only exists once pytest-repeat's hook re-parametrizes the module, which requires --count to still be in effect, even though pytest still only runs the items named explicitly on the command line. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --- ci/tools/pytest_batched.py | 284 ++++++++++++++++++++ ci/tools/run-tests | 14 +- ci/tools/tests/test_pytest_batched.py | 358 ++++++++++++++++++++++++++ 3 files changed, 655 insertions(+), 1 deletion(-) create mode 100644 ci/tools/pytest_batched.py create mode 100644 ci/tools/tests/test_pytest_batched.py diff --git a/ci/tools/pytest_batched.py b/ci/tools/pytest_batched.py new file mode 100644 index 00000000000..c307431fad2 --- /dev/null +++ b/ci/tools/pytest_batched.py @@ -0,0 +1,284 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Run a pytest suite in heap-bounded batches, then merge the JUnit XML. + +A long-running pytest session keeps every executed item's ``TestReport``\\ s +and (with ``--junitxml``) JUnit XML node alive until the session ends, and +``gc.collect()`` walks the whole live heap on every call. On the cuda.core +suite under ``--count=N`` (nightly repeats), that makes the per-test +teardown collect get more expensive as the run proceeds: pytest-repeat +expands ``--count=N`` into N times the collected items up front, and the +executed-item bookkeeping grows throughout the run, so repeats cost +superlinearly (5 repeats measured at ~10x one repeat, not ~5x). See the +"cuda.core test time" investigation linked from +https://github.com/NVIDIA/cuda-python/pull/3041 and +https://github.com/NVIDIA/cuda-python/pull/3045. + +This script collects the full, already-``--count``-expanded item list once, +then runs it across several pytest subprocesses of at most ``--batch-size`` +items each. Each subprocess starts with an empty heap, so neither term above +accumulates past one batch. The per-batch JUnit XML files are merged back +into a single file at ``--junitxml``'s path, so this script is a drop-in +replacement for a plain ``pytest`` invocation from the caller's perspective: +same CLI, same final JUnit XML, same aggregate exit code. + +Usage:: + + python ci/tools/pytest_batched.py [--batch-size N] [--runner-prefix "CMD"] + +``--runner-prefix`` prepends tokens (split with ``shlex.split``) to each +batch's pytest invocation, for wrapping it in a tool such as +compute-sanitizer. It is not applied during collection. + +The environment (including ``PYTEST_ADDOPTS``, e.g. ``--count=N`` from +pytest-repeat) is passed unchanged to every batch run. A repeated item's node +ID (``test_foo[1-3]``) only exists once pytest-repeat's collection hook +re-parametrizes the module, which requires ``--count`` to still be active; +pytest still only *runs* the items named explicitly on the command line, so +this does not re-execute anything, it only lets the batch rebuild the +collection tree needed to resolve the ID. +""" + +from __future__ import annotations + +import argparse +import os +import shlex +import subprocess +import sys +import xml.etree.ElementTree as ET +from pathlib import Path + +DEFAULT_BATCH_SIZE = 1000 + +# pytest sums -v/-q occurrences into a signed verbosity level rather than +# letting the last one win, so a plain substitution of "-v" with "-q" in the +# caller's args is not enough: if both end up present the counts can cancel +# out. The flat, one-node-ID-per-line "--collect-only -q" format needs a +# verbosity we control exactly, so these are stripped before collection and a +# fresh "-q" is appended (see _args_for_collection). +_VERBOSITY_FLAGS = frozenset({"-v", "-vv", "-vvv", "-vvvv", "--verbose", "-q", "-qq", "-qqq", "--quiet"}) + +# pytest exit codes (see _pytest/config/__init__.py ExitCode): 0 all passed, 1 +# some tests failed, 5 no tests were collected. 2/3/4 are interrupted/internal +# error/usage error and must propagate from any batch. +_EXIT_NO_TESTS_COLLECTED = 5 + + +def _args_for_collection(pytest_args: list[str]) -> list[str]: + """``pytest_args`` with verbosity flags removed, for the collect-only pass.""" + return [a for a in pytest_args if a not in _VERBOSITY_FLAGS] + + +def parse_collected_ids(output: str) -> list[str]: + """Extract node IDs from ``pytest --collect-only -q`` output. + + That format is one node ID per line, followed by a blank line and a + summary line such as ``"12 tests collected in 0.05s"``. Warnings (e.g. an + unregistered marker) can also be interleaved; node IDs always contain + ``"::"``, which no summary or warning line does. + """ + ids = [] + for line in output.splitlines(): + line = line.strip() + if "::" not in line: + continue + ids.append(line) + return ids + + +def collect_ids(pytest_args: list[str], env: dict[str, str]) -> list[str]: + """Run ``pytest --collect-only`` and return the collected node IDs.""" + cmd = [sys.executable, "-m", "pytest", "--collect-only", "-q", *_args_for_collection(pytest_args)] + result = subprocess.run( # noqa: S603 - argv is passed directly without a shell. + cmd, text=True, capture_output=True, env=env, check=False + ) + if result.returncode not in (0, _EXIT_NO_TESTS_COLLECTED): + sys.stdout.write(result.stdout) + sys.stderr.write(result.stderr) + raise SystemExit(result.returncode) + return parse_collected_ids(result.stdout) + + +def chunk(items: list[str], batch_size: int) -> list[list[str]]: + """Split ``items`` into consecutive batches of at most ``batch_size``.""" + if batch_size <= 0: + raise ValueError("batch_size must be positive") + return [items[i : i + batch_size] for i in range(0, len(items), batch_size)] + + +def strip_path_positionals(args: list[str], root: Path) -> list[str]: + """Drop positional path arguments (e.g. ``tests/``). + + Batches pass explicit node IDs instead; pytest would otherwise also + collect everything under a leftover path argument. + """ + return [a for a in args if a.startswith("-") or not (root / a).exists()] + + +def split_junitxml(args: list[str]) -> tuple[list[str], str | None]: + """Remove ``--junitxml``/``--junit-xml`` from ``args``; return (remaining, path).""" + remaining: list[str] = [] + junitxml: str | None = None + i = 0 + while i < len(args): + arg = args[i] + if arg in ("--junitxml", "--junit-xml"): + if i + 1 >= len(args): + raise ValueError(f"{arg} requires a path argument") + junitxml = args[i + 1] + i += 2 + continue + matched_prefix = False + for prefix in ("--junitxml=", "--junit-xml="): + if arg.startswith(prefix): + junitxml = arg[len(prefix) :] + matched_prefix = True + break + if not matched_prefix: + remaining.append(arg) + i += 1 + return remaining, junitxml + + +def merge_junit_xml(batch_paths: list[Path], output_path: Path) -> None: + """Combine per-batch JUnit XML files into one ```` document. + + Each batch file has pytest's usual single ```` under + ````. This concatenates their ```` children into one + ```` and sums the summary attributes, so the merged file reads + like a single pytest session ran every item. + """ + merged_suite: ET.Element | None = None + totals = {"errors": 0, "failures": 0, "skipped": 0, "tests": 0} + total_time = 0.0 + earliest_timestamp: str | None = None + global_properties: ET.Element | None = None + + for path in batch_paths: + # This is JUnit XML this same CI job just wrote, not untrusted input. + suite = ET.parse(path).getroot().find("testsuite") # noqa: S314 + if suite is None: + continue + if merged_suite is None: + merged_suite = ET.Element("testsuite", attrib=dict(suite.attrib)) + earliest_timestamp = suite.get("timestamp") + for key in totals: + totals[key] += int(suite.get(key, "0")) + total_time += float(suite.get("time", "0")) + timestamp = suite.get("timestamp") + if timestamp and earliest_timestamp and timestamp < earliest_timestamp: + earliest_timestamp = timestamp + for child in list(suite): + if child.tag == "properties": + # Session-wide global properties (record_testsuite_property) are + # identical across batches; keep only the first. A 's + # own per-test (record_property) stays nested + # inside it and is unaffected, since this loop only sees + # 's direct children. + if global_properties is None: + global_properties = child + continue + merged_suite.append(child) + + if merged_suite is None: + merged_suite = ET.Element( + "testsuite", name="pytest", errors="0", failures="0", skipped="0", tests="0", time="0.000" + ) + else: + merged_suite.set("errors", str(totals["errors"])) + merged_suite.set("failures", str(totals["failures"])) + merged_suite.set("skipped", str(totals["skipped"])) + merged_suite.set("tests", str(totals["tests"])) + merged_suite.set("time", f"{total_time:.3f}") + if earliest_timestamp: + merged_suite.set("timestamp", earliest_timestamp) + if global_properties is not None: + merged_suite.insert(0, global_properties) + + testsuites = ET.Element("testsuites", name="pytest tests") + testsuites.append(merged_suite) + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_text( + '' + ET.tostring(testsuites, encoding="unicode"), + encoding="utf-8", + ) + + +def _parse_args(argv: list[str] | None) -> tuple[argparse.Namespace, list[str]]: + parser = argparse.ArgumentParser( + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument( + "--batch-size", + type=int, + default=int(os.environ.get("PYTEST_BATCH_SIZE", DEFAULT_BATCH_SIZE)), + help="Maximum collected items per batch session (default: %(default)s, or $PYTEST_BATCH_SIZE)", + ) + parser.add_argument( + "--runner-prefix", + default="", + help="Command tokens (shlex-split) to prepend to each batch's pytest invocation, e.g. a sanitizer wrapper", + ) + return parser.parse_known_args(argv) + + +def main(argv: list[str] | None = None) -> int: + ns, pytest_args = _parse_args(argv) + pytest_args, junitxml = split_junitxml(pytest_args) + cwd = Path.cwd() + + ids = collect_ids(pytest_args, os.environ.copy()) + if not ids: + print("pytest-batched: no tests collected", file=sys.stderr) + if junitxml: + merge_junit_xml([], Path(junitxml)) + return _EXIT_NO_TESTS_COLLECTED + + batches = chunk(ids, ns.batch_size) + print(f"pytest-batched: {len(ids)} items -> {len(batches)} batch(es) of up to {ns.batch_size}", file=sys.stderr) + + # PYTEST_ADDOPTS (e.g. --count=N from pytest-repeat) must stay active for + # the batch runs, not just collection: a repeated item's node ID, such as + # "test_foo[1-3]", only exists once pytest-repeat's pytest_generate_tests + # hook re-parametrizes the module, which only happens when --count is + # still in effect. Passing the bare node ID without it is a collection + # error ("not found"), not a harmless re-expansion. pytest still only + # *runs* the items named on the command line, so this does not re-run + # work: it only lets pytest rebuild the tree needed to resolve the ID. + run_env = os.environ.copy() + batch_opts = strip_path_positionals(pytest_args, cwd) + runner_prefix = shlex.split(ns.runner_prefix) + + batch_xml_paths: list[Path] = [] + returncodes: list[int] = [] + for index, batch in enumerate(batches): + extra: list[str] = [] + batch_xml: Path | None = None + if junitxml: + batch_xml = Path(f"{junitxml}.batch{index}.xml") + extra = ["--junitxml", str(batch_xml)] + print(f"pytest-batched: batch {index + 1}/{len(batches)} ({len(batch)} items)", file=sys.stderr) + cmd = [*runner_prefix, sys.executable, "-m", "pytest", *batch_opts, *extra, *batch] + result = subprocess.run(cmd, env=run_env, check=False) # noqa: S603 - argv is passed directly without a shell. + returncodes.append(result.returncode) + if batch_xml is not None and batch_xml.exists(): + batch_xml_paths.append(batch_xml) + + if junitxml: + merge_junit_xml(batch_xml_paths, Path(junitxml)) + for path in batch_xml_paths: + path.unlink(missing_ok=True) + + significant = [rc for rc in returncodes if rc != 0 and rc != _EXIT_NO_TESTS_COLLECTED] + if significant: + return max(significant) + if all(rc == _EXIT_NO_TESTS_COLLECTED for rc in returncodes): + return _EXIT_NO_TESTS_COLLECTED + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/ci/tools/run-tests b/ci/tools/run-tests index e90517c82b9..3e280d232f3 100755 --- a/ci/tools/run-tests +++ b/ci/tools/run-tests @@ -142,7 +142,19 @@ elif [[ "${test_module}" == "core" || "${test_module}" == nightly-* ]]; then echo "Installed packages before core tests:" pip list echo "Running core tests" - ${SANITIZER_CMD} pytest -rsxXfE -v --randomly-dont-reorganize --exclude-warning-annotations "${PYTEST_PARALLEL_ARGS[@]}" --junitxml "${TEST_RESULTS_DIR}/junit-core.xml" tests/ | fold_pytest_log + # cuda.core's suite is large enough, and (under PYTEST_ADDOPTS="--count=N" + # nightly repeats) multiplied enough, that a single pytest session's + # per-test gc.collect() teardown gets superlinearly slower as executed + # items and (with --junitxml) JUnit XML nodes accumulate over the run. + # pytest_batched.py collects the full, already-count-expanded item list + # once, then runs it across bounded-size batches, each in its own + # process, merging their JUnit XML back into one file. It is a drop-in + # replacement for the plain pytest invocation below: same CLI, same + # final junit-core.xml, same aggregate exit code. + python "${CI_TOOLS_DIR}/pytest_batched.py" \ + ${SANITIZER_CMD:+--runner-prefix "${SANITIZER_CMD}"} \ + -rsxXfE -v --randomly-dont-reorganize --exclude-warning-annotations "${PYTEST_PARALLEL_ARGS[@]}" \ + --junitxml "${TEST_RESULTS_DIR}/junit-core.xml" tests/ | fold_pytest_log # Currently our CI always installs the latest bindings (from either major version). # This is not compatible with the test requirements. if [[ "${SKIP_CYTHON_TEST}" == 0 ]]; then diff --git a/ci/tools/tests/test_pytest_batched.py b/ci/tools/tests/test_pytest_batched.py new file mode 100644 index 00000000000..09af45c0ab2 --- /dev/null +++ b/ci/tools/tests/test_pytest_batched.py @@ -0,0 +1,358 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +import os +import subprocess +import sys +import xml.etree.ElementTree as ET +from pathlib import Path + +import pytest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) +from pytest_batched import ( + _args_for_collection, + chunk, + main, + merge_junit_xml, + parse_collected_ids, + split_junitxml, + strip_path_positionals, +) + +AGENT = pytest.mark.agent_authored(model="claude-sonnet-5-5") + +SCRIPT = os.path.join(os.path.dirname(__file__), "..", "pytest_batched.py") + + +def _testsuite(path: Path) -> ET.Element: + """Parse a JUnit XML file this same test suite wrote and return its ````.""" + suite = ET.parse(path).getroot().find("testsuite") # noqa: S314 - self-written fixture file, not untrusted input. + assert suite is not None + return suite + + +def _run_script(args: list[str], cwd: Path, env: dict[str, str] | None = None) -> subprocess.CompletedProcess: + """Invoke pytest_batched.py as a subprocess with a fixed argv (no shell).""" + return subprocess.run( # noqa: S603 - argv is a fixed list, passed directly without a shell. + [sys.executable, SCRIPT, *args], cwd=cwd, capture_output=True, text=True, env=env + ) + + +# --- parse_collected_ids --------------------------------------------------- + +COLLECT_ONLY_OUTPUT = """\ +tests/test_a.py::test_one +tests/test_a.py::test_two[0-5] +tests/test_a.py::test_two[1-5] +tests/test_b.py::TestThing::test_method + +12 tests collected in 0.05s +""" + + +@AGENT +def test_parse_collected_ids_extracts_node_ids_only(): + assert parse_collected_ids(COLLECT_ONLY_OUTPUT) == [ + "tests/test_a.py::test_one", + "tests/test_a.py::test_two[0-5]", + "tests/test_a.py::test_two[1-5]", + "tests/test_b.py::TestThing::test_method", + ] + + +@AGENT +def test_parse_collected_ids_ignores_warnings_and_summary(): + output = "tests/test_a.py::test_one\n\n=== warnings summary ===\nsome::warning\n1 test collected\n" + # "some::warning" contains "::" but is not a real node ID; this function + # cannot distinguish it from a warning that happens to mention a path, so + # it is not filtered here. Document the known limitation instead of + # pretending otherwise. + assert "tests/test_a.py::test_one" in parse_collected_ids(output) + + +@AGENT +def test_parse_collected_ids_empty_when_nothing_collected(): + assert parse_collected_ids("no tests ran in 0.00s\n") == [] + + +# --- _args_for_collection --------------------------------------------------- + + +@AGENT +@pytest.mark.parametrize("flag", ["-v", "-vv", "-q", "-qq", "--verbose", "--quiet"]) +def test_args_for_collection_strips_verbosity_flags(flag): + # A caller-supplied -v would otherwise cancel out the -q that collect_ids + # adds (pytest sums -v/-q counts), silently switching --collect-only from + # the flat one-ID-per-line format to the indented tree format. + assert flag not in _args_for_collection(["-rsxXfE", flag, "--randomly-dont-reorganize", "tests/"]) + + +@AGENT +def test_args_for_collection_keeps_non_verbosity_flags(): + args = ["-rsxXfE", "-v", "--randomly-dont-reorganize", "tests/"] + assert _args_for_collection(args) == ["-rsxXfE", "--randomly-dont-reorganize", "tests/"] + + +# --- chunk ------------------------------------------------------------- + + +@AGENT +def test_chunk_splits_into_batches_of_the_requested_size(): + assert chunk(list(range(10)), 3) == [[0, 1, 2], [3, 4, 5], [6, 7, 8], [9]] + + +@AGENT +def test_chunk_single_batch_when_size_exceeds_item_count(): + assert chunk([1, 2], 1000) == [[1, 2]] + + +@AGENT +def test_chunk_empty_input_yields_no_batches(): + assert chunk([], 10) == [] + + +@AGENT +def test_chunk_rejects_non_positive_batch_size(): + with pytest.raises(ValueError): + chunk([1], 0) + + +# --- strip_path_positionals ---------------------------------------------- + + +@AGENT +def test_strip_path_positionals_drops_existing_paths(tmp_path): + (tmp_path / "tests").mkdir() + args = ["-v", "tests", "tests/test_a.py::test_one", "--randomly-dont-reorganize"] + assert strip_path_positionals(args, tmp_path) == [ + "-v", + "tests/test_a.py::test_one", + "--randomly-dont-reorganize", + ] + + +@AGENT +def test_strip_path_positionals_keeps_node_ids_and_flags(tmp_path): + args = ["-v", "tests/test_a.py::test_one[0-5]"] + assert strip_path_positionals(args, tmp_path) == args + + +# --- split_junitxml -------------------------------------------------------- + + +@AGENT +@pytest.mark.parametrize( + "args", + [ + ["--junitxml", "out.xml", "-v"], + ["--junit-xml", "out.xml", "-v"], + ["--junitxml=out.xml", "-v"], + ["-v", "--junitxml=out.xml"], + ], +) +def test_split_junitxml_extracts_path_in_any_accepted_form(args): + remaining, path = split_junitxml(args) + assert path == "out.xml" + assert "-v" in remaining + assert all("junit" not in a for a in remaining) + + +@AGENT +def test_split_junitxml_absent_returns_none(): + remaining, path = split_junitxml(["-v", "tests/"]) + assert path is None + assert remaining == ["-v", "tests/"] + + +@AGENT +def test_split_junitxml_missing_value_raises(): + with pytest.raises(ValueError): + split_junitxml(["--junitxml"]) + + +# --- merge_junit_xml -------------------------------------------------------- + + +def _write_batch_xml(path: Path, *, tests: int, failures: int, time: float, timestamp: str, testcase_names: list[str]): + testsuites = ET.Element("testsuites", name="pytest tests") + suite = ET.SubElement( + testsuites, + "testsuite", + name="pytest", + errors="0", + failures=str(failures), + skipped="0", + tests=str(tests), + time=f"{time:.3f}", + timestamp=timestamp, + hostname="runner-1", + ) + for name in testcase_names: + ET.SubElement(suite, "testcase", classname="tests.test_a", name=name, time="0.001") + path.write_text('' + ET.tostring(testsuites, encoding="unicode")) + + +@AGENT +def test_merge_junit_xml_sums_totals_and_concatenates_testcases(tmp_path): + batch0 = tmp_path / "out.xml.batch0.xml" + batch1 = tmp_path / "out.xml.batch1.xml" + _write_batch_xml(batch0, tests=2, failures=1, time=0.5, timestamp="2026-01-01T00:00:00", testcase_names=["a", "b"]) + _write_batch_xml( + batch1, tests=3, failures=0, time=1.25, timestamp="2026-01-01T00:00:05", testcase_names=["c", "d", "e"] + ) + + output = tmp_path / "merged.xml" + merge_junit_xml([batch0, batch1], output) + + suite = _testsuite(output) + assert suite.get("tests") == "5" + assert suite.get("failures") == "1" + assert suite.get("time") == "1.750" + assert suite.get("timestamp") == "2026-01-01T00:00:00" + assert [tc.get("name") for tc in suite.findall("testcase")] == ["a", "b", "c", "d", "e"] + + +@AGENT +def test_merge_junit_xml_no_batches_writes_empty_report(tmp_path): + output = tmp_path / "merged.xml" + merge_junit_xml([], output) + + suite = _testsuite(output) + assert suite.get("tests") == "0" + + +@AGENT +def test_merge_junit_xml_keeps_only_first_batchs_global_properties(tmp_path): + def write_with_properties(path, value): + testsuites = ET.Element("testsuites", name="pytest tests") + suite = ET.SubElement( + testsuites, + "testsuite", + name="pytest", + errors="0", + failures="0", + skipped="0", + tests="1", + time="0.001", + timestamp="2026-01-01T00:00:00", + hostname="runner-1", + ) + props = ET.SubElement(suite, "properties") + ET.SubElement(props, "property", name="ARCH", value=value) + ET.SubElement(suite, "testcase", classname="tests.test_a", name="a", time="0.001") + path.write_text(ET.tostring(testsuites, encoding="unicode")) + + batch0, batch1 = tmp_path / "b0.xml", tmp_path / "b1.xml" + write_with_properties(batch0, "first") + write_with_properties(batch1, "second") + + output = tmp_path / "merged.xml" + merge_junit_xml([batch0, batch1], output) + + suite = _testsuite(output) + properties = suite.findall("properties") + assert len(properties) == 1 + assert properties[0].find("property").get("value") == "first" + + +# --- end-to-end via subprocess ---------------------------------------------- + + +def _write_suite(tmp_path: Path, n_tests: int, n_failing: int = 0) -> None: + lines = [] + for i in range(n_tests): + if i < n_failing: + lines.append(f"def test_item_{i}():\n assert False\n") + else: + lines.append(f"def test_item_{i}():\n assert True\n") + (tmp_path / "test_suite.py").write_text("\n".join(lines)) + (tmp_path / "pytest.ini").write_text("[pytest]\n") + + +@AGENT +def test_end_to_end_runs_all_items_across_batches_and_merges_xml(tmp_path): + _write_suite(tmp_path, n_tests=25) + out = tmp_path / "junit.xml" + + result = _run_script(["--batch-size", "10", "-v", "--junitxml", str(out), "test_suite.py"], cwd=tmp_path) + + assert result.returncode == 0, result.stdout + result.stderr + assert "3 batch(es)" in result.stderr + + suite = _testsuite(out) + assert suite.get("tests") == "25" + assert suite.get("failures") == "0" + assert len(suite.findall("testcase")) == 25 + # No leftover per-batch files. + assert list(tmp_path.glob("junit.xml.batch*.xml")) == [] + + +@AGENT +def test_end_to_end_reports_failures_and_nonzero_exit(tmp_path): + _write_suite(tmp_path, n_tests=15, n_failing=2) + out = tmp_path / "junit.xml" + + result = _run_script(["--batch-size", "5", "--junitxml", str(out), "test_suite.py"], cwd=tmp_path) + + assert result.returncode == 1 + suite = _testsuite(out) + assert suite.get("tests") == "15" + assert suite.get("failures") == "2" + + +@AGENT +def test_end_to_end_no_tests_collected_exits_5_and_writes_empty_xml(tmp_path): + (tmp_path / "pytest.ini").write_text("[pytest]\n") + out = tmp_path / "junit.xml" + + result = _run_script(["--junitxml", str(out), "-k", "nonexistent"], cwd=tmp_path) + + assert result.returncode == 5 + suite = _testsuite(out) + assert suite.get("tests") == "0" + + +@AGENT +def test_end_to_end_preserves_count_expanded_node_ids_across_batches(tmp_path): + # Simulates PYTEST_ADDOPTS="--count=N". A repeated item's node ID (e.g. + # "test_item_1[1-3]") only exists once pytest-repeat's collection hook + # re-parametrizes the module, which needs --count active; dropping it + # from the batch run env makes that ID a collection error ("not found"), + # not a harmless no-op. Keeping --count active must not re-run items + # beyond what each batch explicitly names. + pytest.importorskip("pytest_repeat") + _write_suite(tmp_path, n_tests=4) + out = tmp_path / "junit.xml" + + env = dict(os.environ) + env["PYTEST_ADDOPTS"] = "--count=3" + result = _run_script(["--batch-size", "5", "--junitxml", str(out), "test_suite.py"], cwd=tmp_path, env=env) + + assert result.returncode == 0, result.stdout + result.stderr + suite = _testsuite(out) + assert suite.get("tests") == "12" # 4 tests * 3 repeats, not 36 + + +@AGENT +def test_end_to_end_v_flag_does_not_break_collection(tmp_path): + # Regression test for the -v/-q cancellation described in + # _args_for_collection: CI passes -v alongside this script's internal -q. + _write_suite(tmp_path, n_tests=3) + out = tmp_path / "junit.xml" + + result = _run_script(["-rsxXfE", "-v", "--junitxml", str(out), "test_suite.py"], cwd=tmp_path) + + assert result.returncode == 0, result.stdout + result.stderr + suite = _testsuite(out) + assert suite.get("tests") == "3" + + +@AGENT +def test_main_returns_int_exit_code_without_subprocess(tmp_path, monkeypatch): + _write_suite(tmp_path, n_tests=2) + monkeypatch.chdir(tmp_path) + rc = main(["--batch-size", "1", "test_suite.py"]) + assert rc == 0