From 6bc7490b957a97d5669bd4e172427f81af40a475 Mon Sep 17 00:00:00 2001 From: Jonah Gray Date: Wed, 15 Jul 2026 10:44:03 -0700 Subject: [PATCH 1/2] fix(OMN-14641): stream coverage sweep subprocess output --- scripts/ci/run_coverage_sweep_gate.py | 9 +++++---- tests/ci/test_run_coverage_sweep_gate.py | 6 ++++++ 2 files changed, 11 insertions(+), 4 deletions(-) diff --git a/scripts/ci/run_coverage_sweep_gate.py b/scripts/ci/run_coverage_sweep_gate.py index b31a2630bc..4b45078b8a 100644 --- a/scripts/ci/run_coverage_sweep_gate.py +++ b/scripts/ci/run_coverage_sweep_gate.py @@ -94,12 +94,14 @@ def generate_coverage_json( f"--cov={target_dir / 'src'}", f"--cov-report=json:{coverage_json}", ] + print( + "coverage-sweep-gate: running coverage generation command: " + " ".join(cmd), + flush=True, + ) try: proc = subprocess.run( cmd, cwd=str(target_dir), - capture_output=True, - text=True, timeout=timeout_s, check=False, ) @@ -111,8 +113,7 @@ def generate_coverage_json( if not coverage_json.is_file(): return False, ( f"coverage generation exited {proc.returncode} but produced no " - f"coverage.json at {coverage_json}. stderr(tail): " - f"{proc.stderr[-2000:]}" + f"coverage.json at {coverage_json}." ) return True, f"generated {coverage_json}" diff --git a/tests/ci/test_run_coverage_sweep_gate.py b/tests/ci/test_run_coverage_sweep_gate.py index 25447e31c5..961d5c9717 100644 --- a/tests/ci/test_run_coverage_sweep_gate.py +++ b/tests/ci/test_run_coverage_sweep_gate.py @@ -165,8 +165,11 @@ def test_generate_helper_invokes_pytest_cov(self, tmp_path: Path) -> None: captured_cmd: list[str] = [] + captured_kwargs: dict[str, object] = {} + def _fake_run(cmd, **kwargs): # type: ignore[no-untyped-def] captured_cmd.extend(cmd) + captured_kwargs.update(kwargs) (tmp_path / "coverage.json").write_text(json.dumps({"files": {}})) return subprocess.CompletedProcess(cmd, 0, stdout="", stderr="") @@ -177,6 +180,9 @@ def _fake_run(cmd, **kwargs): # type: ignore[no-untyped-def] assert "coverage.json" in message assert "pytest" in captured_cmd assert any("--cov-report=json" in part for part in captured_cmd) + assert "capture_output" not in captured_kwargs + assert "stdout" not in captured_kwargs + assert "stderr" not in captured_kwargs def test_generate_helper_reports_failure_without_swallowing( self, tmp_path: Path From 98d9c06b7494ade618e717d09a37eea39dc1fa1d Mon Sep 17 00:00:00 2001 From: Jonah Gray Date: Wed, 15 Jul 2026 11:08:16 -0700 Subject: [PATCH 2/2] fix(OMN-14641): heartbeat coverage sweep generation --- scripts/ci/run_coverage_sweep_gate.py | 25 +++++++- tests/ci/test_run_coverage_sweep_gate.py | 73 +++++++++++++++++++++--- 2 files changed, 86 insertions(+), 12 deletions(-) diff --git a/scripts/ci/run_coverage_sweep_gate.py b/scripts/ci/run_coverage_sweep_gate.py index 4b45078b8a..7198d0ba8b 100644 --- a/scripts/ci/run_coverage_sweep_gate.py +++ b/scripts/ci/run_coverage_sweep_gate.py @@ -50,6 +50,7 @@ import json import subprocess import sys +import time from pathlib import Path # Add src/ to path so omnimarket imports work in CI without editable install @@ -68,6 +69,7 @@ def generate_coverage_json( test_path: str = "tests/", marker_expr: str = "not kafka", timeout_s: int = 1800, + heartbeat_s: int = 60, ) -> tuple[bool, str]: """Run the real test suite under coverage and write ``coverage.json``. @@ -99,12 +101,29 @@ def generate_coverage_json( flush=True, ) try: - proc = subprocess.run( + proc = subprocess.Popen( cmd, cwd=str(target_dir), - timeout=timeout_s, - check=False, ) + started_at = time.monotonic() + next_heartbeat_at = started_at + heartbeat_s + deadline_at = started_at + timeout_s + + while proc.poll() is None: + now = time.monotonic() + if now >= deadline_at: + proc.kill() + proc.wait() + return False, f"coverage generation timed out after {timeout_s}s" + if now >= next_heartbeat_at: + elapsed_s = int(now - started_at) + print( + "coverage-sweep-gate: coverage generation still running " + f"after {elapsed_s}s", + flush=True, + ) + next_heartbeat_at = now + heartbeat_s + time.sleep(min(1.0, max(0.1, next_heartbeat_at - now))) except subprocess.TimeoutExpired as exc: return False, f"coverage generation timed out after {timeout_s}s: {exc}" except OSError as exc: diff --git a/tests/ci/test_run_coverage_sweep_gate.py b/tests/ci/test_run_coverage_sweep_gate.py index 961d5c9717..5432f9b709 100644 --- a/tests/ci/test_run_coverage_sweep_gate.py +++ b/tests/ci/test_run_coverage_sweep_gate.py @@ -11,7 +11,6 @@ from __future__ import annotations import json -import subprocess import sys from pathlib import Path from unittest.mock import patch @@ -167,14 +166,29 @@ def test_generate_helper_invokes_pytest_cov(self, tmp_path: Path) -> None: captured_kwargs: dict[str, object] = {} - def _fake_run(cmd, **kwargs): # type: ignore[no-untyped-def] + class _FakeProcess: + returncode = 0 + + def __init__(self) -> None: + self._poll_count = 0 + + def poll(self) -> int | None: + self._poll_count += 1 + if self._poll_count == 1: + return None + (tmp_path / "coverage.json").write_text(json.dumps({"files": {}})) + return 0 + + def _fake_popen(cmd, **kwargs): # type: ignore[no-untyped-def] captured_cmd.extend(cmd) captured_kwargs.update(kwargs) - (tmp_path / "coverage.json").write_text(json.dumps({"files": {}})) - return subprocess.CompletedProcess(cmd, 0, stdout="", stderr="") + return _FakeProcess() - with patch("run_coverage_sweep_gate.subprocess.run", side_effect=_fake_run): - ok, message = generate_coverage_json(tmp_path) + with ( + patch("run_coverage_sweep_gate.subprocess.Popen", side_effect=_fake_popen), + patch("run_coverage_sweep_gate.time.sleep"), + ): + ok, message = generate_coverage_json(tmp_path, heartbeat_s=1) assert ok is True assert "coverage.json" in message @@ -189,15 +203,56 @@ def test_generate_helper_reports_failure_without_swallowing( ) -> None: from run_coverage_sweep_gate import generate_coverage_json - def _fake_run(cmd, **kwargs): # type: ignore[no-untyped-def] - return subprocess.CompletedProcess(cmd, 1, stdout="", stderr="boom") + class _FakeProcess: + returncode = 1 + + def poll(self) -> int: + return 1 - with patch("run_coverage_sweep_gate.subprocess.run", side_effect=_fake_run): + def _fake_popen(cmd, **kwargs): # type: ignore[no-untyped-def] + return _FakeProcess() + + with patch("run_coverage_sweep_gate.subprocess.Popen", side_effect=_fake_popen): ok, message = generate_coverage_json(tmp_path) assert ok is False assert "boom" in message or "coverage.json" in message + def test_generate_helper_emits_parent_heartbeat( + self, tmp_path: Path, capsys: pytest.CaptureFixture[str] + ) -> None: + from run_coverage_sweep_gate import generate_coverage_json + + class _FakeProcess: + returncode = 0 + + def __init__(self) -> None: + self._poll_count = 0 + + def poll(self) -> int | None: + self._poll_count += 1 + if self._poll_count < 3: + return None + (tmp_path / "coverage.json").write_text(json.dumps({"files": {}})) + return 0 + + times = iter([0.0, 1.0, 1.0, 2.0, 2.0]) + + with ( + patch( + "run_coverage_sweep_gate.subprocess.Popen", return_value=_FakeProcess() + ), + patch( + "run_coverage_sweep_gate.time.monotonic", + side_effect=lambda: next(times), + ), + patch("run_coverage_sweep_gate.time.sleep"), + ): + ok, _ = generate_coverage_json(tmp_path, heartbeat_s=1) + + assert ok is True + assert "coverage generation still running after 1s" in capsys.readouterr().out + def test_generation_failure_exits_two(self, tmp_path: Path) -> None: from run_coverage_sweep_gate import main