diff --git a/devtools/pytest_progress_plugin.py b/devtools/pytest_progress_plugin.py index 2ed536dd02..f1c07c6c85 100644 --- a/devtools/pytest_progress_plugin.py +++ b/devtools/pytest_progress_plugin.py @@ -156,10 +156,20 @@ def pytest_collection_modifyitems(session: Any, config: Any, items: list[Any]) - _SELECTED_COUNT = len(items) limit = _selection_nodeid_limit() selected_nodeids = [str(getattr(item, "nodeid", item)) for item in items[:limit]] + # Marker metadata is a compact routing index, not a node-id sample. Keep + # it complete so seed sharding can isolate load-sensitive/TUI nodes even + # when the human-readable node-id sample is capped at 500 entries. + selected_node_markers = { + str(getattr(item, "nodeid", item)): sorted( + {str(mark.name) for mark in getattr(item, "iter_markers", lambda: ())()} + ) + for item in items + } payload: dict[str, Any] = { "selected_count": _SELECTED_COUNT, "deselected_count": _DESELECTED_COUNT, "selected_nodeids": selected_nodeids, + "selected_node_markers": selected_node_markers, "selected_nodeids_omitted": max(0, _SELECTED_COUNT - len(selected_nodeids)), "deselected_nodeids": list(_DESELECTED_NODEIDS_SAMPLE), "deselected_nodeids_omitted": max(0, _DESELECTED_COUNT - len(_DESELECTED_NODEIDS_SAMPLE)), diff --git a/devtools/testmon_state.py b/devtools/testmon_state.py index 1929ff801b..fd3dc16e65 100644 --- a/devtools/testmon_state.py +++ b/devtools/testmon_state.py @@ -81,7 +81,12 @@ class TerminalAuthorization(StrEnum): _TERMINAL_NODE_OUTCOMES = frozenset({"passed", "failed", "error", "skipped"}) -def seed_shard_plan(nodeids: Sequence[str], *, shard_size: int) -> list[dict[str, Any]]: +def seed_shard_plan( + nodeids: Sequence[str], + *, + shard_size: int, + serial_nodeids: Sequence[str] = (), +) -> list[dict[str, Any]]: """Partition a complete node set into stable, contiguous, serial shards.""" if shard_size <= 0: raise ValueError("testmon seed shard_size must be positive") @@ -90,16 +95,29 @@ def seed_shard_plan(nodeids: Sequence[str], *, shard_size: int) -> list[dict[str if len(set(nodeids)) != len(nodeids): raise ValueError("testmon seed nodeids must be unique") ordered = tuple(sorted(nodeids)) + serial = set(serial_nodeids) + if not serial.issubset(ordered): + raise ValueError("testmon serial shard nodes must belong to the seed corpus") + chunks: list[tuple[str, list[str]]] = [] + for offset in range(0, len(ordered), shard_size): + chunk = list(ordered[offset : offset + shard_size]) + parallel = [nodeid for nodeid in chunk if nodeid not in serial] + isolated = [nodeid for nodeid in chunk if nodeid in serial] + if parallel: + chunks.append(("parallel", parallel)) + if isolated: + chunks.append(("serial", isolated)) return [ { "index": index, - "nodeids": list(ordered[offset : offset + shard_size]), - "nodeid_count": len(ordered[offset : offset + shard_size]), - "nodeid_digest": hashlib.sha256("\n".join(ordered[offset : offset + shard_size]).encode()).hexdigest(), + "nodeids": chunk, + "nodeid_count": len(chunk), + "nodeid_digest": hashlib.sha256("\n".join(chunk).encode()).hexdigest(), + "execution_mode": mode, "status": SeedShardStatus.PENDING.value, "node_outcomes": [], } - for index, offset in enumerate(range(0, len(ordered), shard_size), start=1) + for index, (mode, chunk) in enumerate(chunks, start=1) ] @@ -120,7 +138,7 @@ def validate_seed_shard_ledger( if not expected or len(set(expected)) != len(expected): return None normalized: list[dict[str, Any]] = [] - observed: list[str] = [] + observed: set[str] = set() for index, raw in enumerate(shards, start=1): if not isinstance(raw, Mapping) or raw.get("index") != index: return None @@ -170,9 +188,11 @@ def validate_seed_shard_ledger( and set(outcome_by_node) != set(nodeids) ): return None + if observed.intersection(nodeids): + return None normalized.append(dict(raw)) - observed.extend(nodeids) - if tuple(observed) != expected: + observed.update(nodeids) + if observed != set(expected): return None return normalized @@ -814,6 +834,9 @@ def inspect_testmon_database(path: Path, expected_nodeids: Sequence[str]) -> Gra ) execution_ids.add(execution_id) name = test_name + if name not in expected: + grouped = [nodeid for nodeid in expected if name.startswith(nodeid + "@")] + name = max(grouped, key=len, default=name) prior = latest.get(name) if prior is None or execution_id > prior[0]: latest[name] = (execution_id, failed == 1) diff --git a/devtools/verify.py b/devtools/verify.py index c1a09efaa4..3204087385 100644 --- a/devtools/verify.py +++ b/devtools/verify.py @@ -2631,7 +2631,19 @@ def _prepare_testmon_seed_shards( shards = ( prior_shards if prior_shards is not None - else (seed_shard_plan(expected, shard_size=TESTMON_SEED_SHARD_SIZE) if expected else []) + else ( + seed_shard_plan( + expected, + shard_size=TESTMON_SEED_SHARD_SIZE, + serial_nodeids=[ + nodeid + for nodeid, markers in (selection or {}).get("selected_node_markers", {}).items() + if "load_sensitive" in markers or "tui" in markers + ], + ) + if expected + else [] + ) ) payload = { **dict(prepared), @@ -2646,16 +2658,64 @@ def _prepare_testmon_seed_shards( return payload -def _seed_shard_command(collection_command: Sequence[str], shard: Mapping[str, Any]) -> list[str]: - """Build a dynamically balanced explicit-node pytest-testmon invocation.""" +def _seed_shard_command( + collection_command: Sequence[str], + shard: Mapping[str, Any], + *, + nodeids_file: Path, +) -> list[str]: + """Build a bounded-argv, dynamically balanced pytest-testmon invocation. + + A full shard's node IDs can exceed the host's ``execve`` argument budget + once ``systemd-run`` and the managed environment are included. Pytest's + response-file syntax keeps the authoritative node list in the run + artifact while making the child command size independent of shard size. + """ nodeids = shard.get("nodeids") if not isinstance(nodeids, list) or not nodeids: raise ValueError("testmon seed shard is missing nodeids") - command = [argument for argument in collection_command if argument != "--collect-only"] - command.extend(["--dist=worksteal", "--testmon", "--testmon-noselect", *nodeids]) + nodeids_file.parent.mkdir(parents=True, exist_ok=True) + nodeids_file.write_text("\n".join(nodeids) + "\n", encoding="utf-8") + command: list[str] = [] + skip_next = False + for argument in collection_command: + if skip_next: + skip_next = False + continue + if argument == "--collect-only": + continue + if argument in {"-n", "--numprocesses"}: + skip_next = True + continue + if argument.startswith("--numprocesses=") or (argument.startswith("-n") and len(argument) > 2): + continue + command.append(argument) + # Collection is deliberately serial, but execution is not. pytest-testmon + # has an xdist-aware controller database; retaining the managed worker pool + # here avoids turning a 20k-node seed into hours of serial fixture setup. + if shard.get("execution_mode") == "serial": + command.extend(["-n", "0", "--testmon", "--testmon-noselect", f"@{nodeids_file}"]) + else: + command.extend( + [ + "--dist=loadgroup", + *_pytest_worker_args(maximum=10), + "--testmon", + "--testmon-noselect", + f"@{nodeids_file}", + ] + ) return command +def _canonical_seed_nodeid(nodeid: str, expected_nodeids: Sequence[str]) -> str: + """Map xdist's ``nodeid@group`` reports back to the collected node ID.""" + if nodeid in expected_nodeids: + return nodeid + candidates = [expected for expected in expected_nodeids if nodeid.startswith(expected + "@")] + return max(candidates, key=len, default=nodeid) + + def _seed_shard_outcomes(shards: Sequence[Mapping[str, Any]]) -> list[dict[str, Any]]: """Flatten the shard ledger in canonical node order for legacy readers.""" outcomes: dict[str, dict[str, Any]] = {} @@ -2684,7 +2744,10 @@ def _checkpoint_testmon_seed_shard( nodeids = shard["nodeids"] artifact_dir = _safe_testmon_artifact_dir(step.get("artifact_dir")) selection = _read_json_artifact(artifact_dir / "selection.json") if artifact_dir is not None else None - selected = _seed_selection_nodeids(selection) if isinstance(selection, Mapping) else None + selected_raw = _seed_selection_nodeids(selection) if isinstance(selection, Mapping) else None + selected = ( + sorted(_canonical_seed_nodeid(nodeid, nodeids) for nodeid in selected_raw) if selected_raw is not None else None + ) database = _testmon_database_state(nodeids) prior = { str(item["nodeid"]): item @@ -2781,6 +2844,7 @@ def _seed_node_outcomes_from_events( nodeid = event.get("nodeid") if not isinstance(nodeid, str) or not nodeid: continue + nodeid = _canonical_seed_nodeid(nodeid, expected_nodeids) if event.get("event") == "test_started": started.add(nodeid) elif event.get("event") == "test_finished": @@ -3422,7 +3486,32 @@ def main(argv: list[str] | None = None) -> int: continue shard_index = int(shard["index"]) shard_label = f"pytest seed-testmon shard {shard_index}/{len(shards)}" - shard_cmd = _seed_shard_command(cmd, shard) + shard_args_path = verify_run.run_dir / "seed-shards" / f"{shard_index:04d}.args" + try: + shard_cmd = _seed_shard_command(cmd, shard, nodeids_file=shard_args_path) + except (OSError, PytestResourceError) as exc: + resource_failure_result = { + "name": shard_label, + "duration_s": 0.0, + "exit": 125, + "diagnosis": ( + "pytest_resource_refusal" + if isinstance(exc, PytestResourceError) + else "testmon_seed_args_file_write_failed" + ), + "error": str(exc), + "shard_index": shard_index, + "shard_count": len(shards), + "shard_nodeid_count": len(shard["nodeids"]), + } + step_results.append(resource_failure_result) + prepared_seed_attempt = _checkpoint_testmon_seed_shard( + prepared=prepared_seed_attempt, + shard_index=shard_index, + step=resource_failure_result, + ) + exit_code = 125 + break _warn_low_memory() shard_rc, shard_elapsed, shard_metadata = _run(shard_label, shard_cmd, run=verify_run) shard_result: dict[str, Any] = { diff --git a/tests/unit/devtools/test_testmon_state.py b/tests/unit/devtools/test_testmon_state.py index 7c9d983540..0a01f3c9a2 100644 --- a/tests/unit/devtools/test_testmon_state.py +++ b/tests/unit/devtools/test_testmon_state.py @@ -88,6 +88,7 @@ def _attempt(data: Path, *, outcomes: tuple[str, str] = ("passed", "failed")) -> def test_failed_complete_graph_is_selection_only_and_rebindable(tmp_path: Path) -> None: data = tmp_path / "testmondata" _write_graph(data, failed=True) + stamp = stamp_from_attempt(_attempt(data), data, checkout_root=tmp_path, protocol_version=PROTOCOL) assert stamp is not None @@ -107,6 +108,35 @@ def test_failed_complete_graph_is_selection_only_and_rebindable(tmp_path: Path) assert validate_stamp(stamp_path, data, checkout_root=tmp_path, protocol_version=PROTOCOL) is None +def test_seed_shard_ledger_rejects_duplicate_nodes_across_shards() -> None: + shard = { + "index": 1, + "nodeids": [NODEIDS[0]], + "nodeid_count": 1, + "nodeid_digest": hashlib.sha256(NODEIDS[0].encode()).hexdigest(), + "status": "complete", + "node_outcomes": [{"nodeid": NODEIDS[0], "outcome": "passed"}], + } + duplicate = {**shard, "index": 2} + + assert testmon_state.validate_seed_shard_ledger([shard, duplicate], expected_nodeids=[NODEIDS[0]]) is None + + +def test_testmon_database_canonicalizes_xdist_group_names(tmp_path: Path) -> None: + data = tmp_path / "testmondata" + _write_graph(data) + with sqlite3.connect(data) as connection: + connection.execute( + "UPDATE test_execution SET test_name = ? WHERE test_name = ?", + (f"{NODEIDS[0]}@web-reader", NODEIDS[0]), + ) + + graph = inspect_testmon_database(data, NODEIDS) + + assert graph.missing_nodeids == () + assert graph.recorded_count == len(NODEIDS) + + def test_omitted_interrupted_and_uncovered_nodes_fail_closed(tmp_path: Path) -> None: data = tmp_path / "testmondata" _write_graph(data) diff --git a/tests/unit/devtools/test_verify.py b/tests/unit/devtools/test_verify.py index adee9b345a..514b7e51b2 100644 --- a/tests/unit/devtools/test_verify.py +++ b/tests/unit/devtools/test_verify.py @@ -322,10 +322,9 @@ def test_seed_testmon_caps_adaptive_workers(monkeypatch: pytest.MonkeyPatch) -> assert command[command.index("-n") + 1] == "0" -def test_seed_shards_are_deterministic_and_use_one_testmon_writer( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch -) -> None: +def test_seed_shards_are_deterministic_and_use_managed_xdist(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.chdir(tmp_path) + monkeypatch.setattr("devtools.verify.adaptive_pytest_worker_count", lambda _environment: 64) expected = sorted(f"tests/test_seed.py::test_{index:03d}" for index in range(TESTMON_SEED_SHARD_SIZE + 2)) prepared = _prepare_testmon_seed_shards( {"resume": False, "expected_nodeids": []}, @@ -340,13 +339,51 @@ def test_seed_shards_are_deterministic_and_use_one_testmon_writer( assert [shard["nodeid_count"] for shard in shards] == [TESTMON_SEED_SHARD_SIZE, 2] assert shards[0]["nodeids"] == expected[:TESTMON_SEED_SHARD_SIZE] assert shards[1]["nodeids"] == expected[TESTMON_SEED_SHARD_SIZE:] - command = _seed_shard_command(["pytest", "--collect-only", "-n", "0"], shards[0]) + nodeids_file = tmp_path / "seed-shard.args" + command = _seed_shard_command(["pytest", "--collect-only", "-n", "0"], shards[0], nodeids_file=nodeids_file) assert "--collect-only" not in command - assert command[command.index("-n") + 1] == "0" + assert command[command.index("-n") + 1] == "10" assert "--testmon" in command assert "--testmon-noselect" in command - assert "--dist=worksteal" in command - assert command[-TESTMON_SEED_SHARD_SIZE:] == expected[:TESTMON_SEED_SHARD_SIZE] + assert "--dist=loadgroup" in command + assert command.count("-n") == 1 + assert command[command.index("-n") + 1] == "10" + assert command[-1] == f"@{nodeids_file}" + assert nodeids_file.read_text().splitlines() == expected[:TESTMON_SEED_SHARD_SIZE] + + +def test_seed_outcomes_normalize_xdist_group_suffix(tmp_path: Path) -> None: + expected = ["tests/test_seed.py::test_grouped"] + events = tmp_path / "events.jsonl" + events.write_text( + json.dumps( + { + "event": "test_report", + "nodeid": f"{expected[0]}@web-reader", + "when": "call", + "outcome": "passed", + } + ) + + "\n" + ) + + outcomes = _seed_node_outcomes_from_events( + events, + expected_nodeids=expected, + database={"node_outcomes": {}}, + pytest_step=None, + ) + + assert outcomes == [ + { + "nodeid": expected[0], + "outcome": "passed", + "reason": "test call passed", + "started": False, + "finished": False, + "phases": [{"when": "call", "outcome": "passed", "duration_s": None}], + } + ] def test_seed_shard_checkpoint_preserves_completed_shards_for_resume( @@ -385,10 +422,24 @@ def test_seed_shard_checkpoint_preserves_completed_shards_for_resume( artifact_dir = tmp_path / "shard-1" artifact_dir.mkdir() (artifact_dir / "selection.json").write_text( - json.dumps({"selected_count": 1, "selected_nodeids": [ordered[0]], "selected_nodeids_omitted": 0}) + json.dumps( + { + "selected_count": 1, + "selected_nodeids": [f"{ordered[0]}@web-reader"], + "selected_nodeids_omitted": 0, + } + ) ) (artifact_dir / "events.jsonl").write_text( - json.dumps({"event": "test_report", "nodeid": ordered[0], "when": "call", "outcome": "passed"}) + "\n" + json.dumps( + { + "event": "test_report", + "nodeid": f"{ordered[0]}@web-reader", + "when": "call", + "outcome": "passed", + } + ) + + "\n" ) checkpointed = _checkpoint_testmon_seed_shard(