|
| 1 | +#!/usr/bin/env python3 |
| 2 | +"""find-polluter.py — binary-search test isolation. |
| 3 | +
|
| 4 | +Finds which test in a suite contaminates another test when run before it. |
| 5 | +Uses binary search: O(log N) runs instead of O(N). |
| 6 | +
|
| 7 | +Usage (Claude Code plugin — CLAUDE_PLUGIN_ROOT is set automatically): |
| 8 | + python3 "${CLAUDE_PLUGIN_ROOT}/bin/find-polluter.py" <failing-test-id> [test-dir] |
| 9 | +
|
| 10 | +Arguments: |
| 11 | + failing-test-id pytest node ID of the test that fails due to contamination |
| 12 | + e.g. tests/test_foo.py::TestClass::test_method |
| 13 | + test-dir directory to search for candidate tests (default: tests) |
| 14 | +
|
| 15 | +Example: |
| 16 | + python3 "${CLAUDE_PLUGIN_ROOT}/bin/find-polluter.py" tests/test_model.py::test_predict tests/ |
| 17 | +
|
| 18 | +Requirements: pytest available on PATH (or via `python -m pytest`). |
| 19 | +
|
| 20 | +Exit codes: |
| 21 | + 0 polluter found and reported |
| 22 | + 1 validation failure (bad args, test passes when isolated, no candidates, |
| 23 | + or pytest unavailable) |
| 24 | +""" |
| 25 | + |
| 26 | +from __future__ import annotations |
| 27 | + |
| 28 | +import math |
| 29 | +import re |
| 30 | +import shutil |
| 31 | +import subprocess |
| 32 | +import sys |
| 33 | +import tempfile |
| 34 | +from collections.abc import Sequence |
| 35 | + |
| 36 | +ISOLATION_PASS_RE = re.compile(r"^(PASSED|1 passed)", re.MULTILINE) |
| 37 | +FAILURE_RE = re.compile(r"FAILED|ERROR") |
| 38 | + |
| 39 | + |
| 40 | +def round_estimate(total: int) -> int: |
| 41 | + """Return upper bound on binary-search rounds for ``total`` candidates. |
| 42 | +
|
| 43 | + Mirrors the bash version's ``ceil(log2(total + 1))`` formula. |
| 44 | +
|
| 45 | + Args: |
| 46 | + total: Number of candidate tests (must be non-negative). |
| 47 | +
|
| 48 | + Returns: |
| 49 | + Maximum number of binary-search rounds needed. |
| 50 | +
|
| 51 | + Examples: |
| 52 | + >>> round_estimate(0) |
| 53 | + 0 |
| 54 | + >>> round_estimate(1) |
| 55 | + 1 |
| 56 | + >>> round_estimate(2) |
| 57 | + 2 |
| 58 | + >>> round_estimate(7) |
| 59 | + 3 |
| 60 | + >>> round_estimate(8) |
| 61 | + 4 |
| 62 | + >>> round_estimate(1000) |
| 63 | + 10 |
| 64 | + """ |
| 65 | + if total < 0: |
| 66 | + raise ValueError(f"total must be non-negative, got {total}") |
| 67 | + if total == 0: |
| 68 | + return 0 |
| 69 | + return math.ceil(math.log2(total + 1)) |
| 70 | + |
| 71 | + |
| 72 | +def binary_midpoint(lo: int, hi: int) -> int: |
| 73 | + """Return the integer midpoint between ``lo`` and ``hi``. |
| 74 | +
|
| 75 | + Args: |
| 76 | + lo: Lower bound (inclusive). |
| 77 | + hi: Upper bound (exclusive). |
| 78 | +
|
| 79 | + Returns: |
| 80 | + Floor of ``(lo + hi) / 2``. |
| 81 | +
|
| 82 | + Examples: |
| 83 | + >>> binary_midpoint(0, 10) |
| 84 | + 5 |
| 85 | + >>> binary_midpoint(3, 7) |
| 86 | + 5 |
| 87 | + >>> binary_midpoint(0, 1) |
| 88 | + 0 |
| 89 | + >>> binary_midpoint(4, 5) |
| 90 | + 4 |
| 91 | + >>> binary_midpoint(0, 0) |
| 92 | + 0 |
| 93 | + """ |
| 94 | + return (lo + hi) // 2 |
| 95 | + |
| 96 | + |
| 97 | +def _resolve_pytest_cmd() -> list[str] | None: |
| 98 | + """Locate the pytest executable; fall back to ``python -m pytest``. |
| 99 | +
|
| 100 | + Returns: |
| 101 | + Argv prefix for invoking pytest, or ``None`` if pytest cannot be |
| 102 | + located. |
| 103 | + """ |
| 104 | + pytest_bin = shutil.which("pytest") |
| 105 | + if pytest_bin: |
| 106 | + return [pytest_bin] |
| 107 | + # Fallback: try the running interpreter's pytest module. |
| 108 | + probe = subprocess.run( |
| 109 | + [sys.executable, "-c", "import pytest"], |
| 110 | + capture_output=True, |
| 111 | + text=True, |
| 112 | + ) |
| 113 | + if probe.returncode == 0: |
| 114 | + return [sys.executable, "-m", "pytest"] |
| 115 | + return None |
| 116 | + |
| 117 | + |
| 118 | +def _run_pytest(pytest_cmd: Sequence[str], args: Sequence[str]) -> str: |
| 119 | + """Run pytest with ``args`` and return combined stdout+stderr text.""" |
| 120 | + result = subprocess.run( |
| 121 | + [*pytest_cmd, *args], |
| 122 | + capture_output=True, |
| 123 | + text=True, |
| 124 | + ) |
| 125 | + return (result.stdout or "") + (result.stderr or "") |
| 126 | + |
| 127 | + |
| 128 | +def passes_isolation(test_id: str, pytest_cmd: Sequence[str]) -> bool: |
| 129 | + """Return ``True`` when ``test_id`` passes when run in isolation. |
| 130 | +
|
| 131 | + Args: |
| 132 | + test_id: pytest node ID of the candidate failing test. |
| 133 | + pytest_cmd: Argv prefix used to launch pytest. |
| 134 | +
|
| 135 | + Returns: |
| 136 | + ``True`` if pytest output indicates the test passed; ``False`` |
| 137 | + otherwise. |
| 138 | + """ |
| 139 | + output = _run_pytest(pytest_cmd, [test_id, "-q", "--tb=short"]) |
| 140 | + return bool(ISOLATION_PASS_RE.search(output)) |
| 141 | + |
| 142 | + |
| 143 | +def collect_candidates( |
| 144 | + test_dir: str, |
| 145 | + failing_test: str, |
| 146 | + pytest_cmd: Sequence[str], |
| 147 | +) -> list[str]: |
| 148 | + """Collect candidate test node IDs, excluding the failing test itself. |
| 149 | +
|
| 150 | + Args: |
| 151 | + test_dir: Directory to search for candidate tests. |
| 152 | + failing_test: pytest node ID to exclude from the candidate list. |
| 153 | + pytest_cmd: Argv prefix used to launch pytest. |
| 154 | +
|
| 155 | + Returns: |
| 156 | + List of candidate node IDs (one per line of pytest collect output |
| 157 | + containing ``::``). |
| 158 | + """ |
| 159 | + result = subprocess.run( |
| 160 | + [*pytest_cmd, test_dir, "--collect-only", "-q"], |
| 161 | + capture_output=True, |
| 162 | + text=True, |
| 163 | + ) |
| 164 | + # Mirror bash: `grep "::" | grep -v failing | grep -v ^$`. pytest writes |
| 165 | + # its collection lines to stdout; ignore stderr noise. |
| 166 | + candidates: list[str] = [] |
| 167 | + for line in result.stdout.splitlines(): |
| 168 | + stripped = line.strip() |
| 169 | + if not stripped: |
| 170 | + continue |
| 171 | + if "::" not in stripped: |
| 172 | + continue |
| 173 | + if stripped == failing_test: |
| 174 | + continue |
| 175 | + candidates.append(stripped) |
| 176 | + return candidates |
| 177 | + |
| 178 | + |
| 179 | +def _contaminates( |
| 180 | + batch: Sequence[str], |
| 181 | + failing_test: str, |
| 182 | + pytest_cmd: Sequence[str], |
| 183 | +) -> bool: |
| 184 | + """Return ``True`` if running ``batch`` before ``failing_test`` fails it.""" |
| 185 | + # Pass the batch via a tempfile (one test per line). pytest accepts |
| 186 | + # ``@filename`` for "args from file"; using a tempfile keeps the OS argv |
| 187 | + # cap from biting on large suites, mirroring bash's mktemp approach. |
| 188 | + with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=True, encoding="utf-8") as batch_file: |
| 189 | + batch_file.write("\n".join(batch)) |
| 190 | + batch_file.write("\n") |
| 191 | + batch_file.flush() |
| 192 | + args = [*batch, failing_test, "-q", "--tb=no"] |
| 193 | + output = _run_pytest(pytest_cmd, args) |
| 194 | + return bool(FAILURE_RE.search(output)) |
| 195 | + |
| 196 | + |
| 197 | +def binary_search( |
| 198 | + candidates: Sequence[str], |
| 199 | + failing_test: str, |
| 200 | + pytest_cmd: Sequence[str], |
| 201 | + log: object | None = None, |
| 202 | +) -> tuple[str, int]: |
| 203 | + """Locate the polluting test via binary search. |
| 204 | +
|
| 205 | + Args: |
| 206 | + candidates: Ordered list of candidate node IDs. |
| 207 | + failing_test: pytest node ID of the failing test. |
| 208 | + pytest_cmd: Argv prefix used to launch pytest. |
| 209 | + log: Stream for progress output; defaults to the *current* value of |
| 210 | + ``sys.stdout`` at call time (so test capture and stream redirection |
| 211 | + work correctly). |
| 212 | +
|
| 213 | + Returns: |
| 214 | + Tuple of ``(polluter_node_id, rounds_used)``. Rounds is ``0`` when only |
| 215 | + a single candidate exists (no narrowing needed). |
| 216 | +
|
| 217 | + Raises: |
| 218 | + ValueError: If ``candidates`` is empty. |
| 219 | + """ |
| 220 | + if not candidates: |
| 221 | + raise ValueError("candidates must contain at least one test") |
| 222 | + |
| 223 | + stream = log if log is not None else sys.stdout |
| 224 | + lo = 0 |
| 225 | + hi = len(candidates) |
| 226 | + rounds = 0 |
| 227 | + |
| 228 | + while (hi - lo) > 1: |
| 229 | + rounds += 1 |
| 230 | + mid = binary_midpoint(lo, hi) |
| 231 | + count = mid - lo |
| 232 | + print( |
| 233 | + f" Round {rounds}: testing [{lo}–{mid}] ({count} tests)...", |
| 234 | + file=stream, |
| 235 | + ) |
| 236 | + |
| 237 | + batch = list(candidates[lo:mid]) |
| 238 | + if _contaminates(batch, failing_test, pytest_cmd): |
| 239 | + hi = mid # polluter is in [lo, mid) |
| 240 | + else: |
| 241 | + lo = mid # polluter is in [mid, hi) |
| 242 | + |
| 243 | + return candidates[lo], rounds |
| 244 | + |
| 245 | + |
| 246 | +def main(argv: Sequence[str] | None = None) -> int: |
| 247 | + """Entry point. Returns process exit code.""" |
| 248 | + args = list(argv if argv is not None else sys.argv[1:]) |
| 249 | + if not args: |
| 250 | + print( |
| 251 | + "Usage: find-polluter.py <failing-test-id> [test-dir]\n\tExample: find-polluter.py tests/test_foo.py::test_bar tests/", |
| 252 | + file=sys.stderr, |
| 253 | + ) |
| 254 | + return 1 |
| 255 | + |
| 256 | + failing_test = args[0] |
| 257 | + test_dir = args[1] if len(args) > 1 else "tests" |
| 258 | + |
| 259 | + pytest_cmd = _resolve_pytest_cmd() |
| 260 | + if pytest_cmd is None: |
| 261 | + print("✗ pytest not found on PATH or via python -m pytest", file=sys.stderr) |
| 262 | + return 1 |
| 263 | + |
| 264 | + # Step 1: verify the failing test passes in isolation. |
| 265 | + print(f"→ Checking {failing_test} in isolation...") |
| 266 | + if passes_isolation(failing_test, pytest_cmd): |
| 267 | + print("✓ Passes in isolation — test-ordering contamination confirmed") |
| 268 | + else: |
| 269 | + print( |
| 270 | + f"✗ {failing_test} fails in isolation — not a test-ordering issue\n\tFix the test itself before using this script.", |
| 271 | + file=sys.stderr, |
| 272 | + ) |
| 273 | + return 1 |
| 274 | + |
| 275 | + # Step 2: collect candidates. |
| 276 | + print(f"→ Collecting candidates from {test_dir}...") |
| 277 | + candidates = collect_candidates(test_dir, failing_test, pytest_cmd) |
| 278 | + total = len(candidates) |
| 279 | + if total == 0: |
| 280 | + print(f"✗ No candidate tests found in {test_dir}", file=sys.stderr) |
| 281 | + return 1 |
| 282 | + |
| 283 | + rounds_upper = round_estimate(total) |
| 284 | + print(f"✓ Found {total} candidates — starting binary search (up to {rounds_upper} rounds)\n") |
| 285 | + |
| 286 | + # Step 3: binary search. |
| 287 | + polluter, rounds = binary_search(candidates, failing_test, pytest_cmd) |
| 288 | + |
| 289 | + # Step 4: report. |
| 290 | + pretty_cmd = " ".join(pytest_cmd) |
| 291 | + print( |
| 292 | + f"\n✓ Polluter found after {rounds} rounds:\n\n\t{polluter}\n\n" |
| 293 | + f'Verify with:\n\t{pretty_cmd} "{polluter}" "{failing_test}" -v\n\n' |
| 294 | + f"Next steps:\n" |
| 295 | + f"\t1. Run the verify command above to confirm\n" |
| 296 | + f"\t2. Check {polluter} for shared state mutation (module-level vars, fixtures, monkeypatches)\n" |
| 297 | + f"\t3. Add proper teardown or use pytest fixtures with 'function' scope to isolate the state" |
| 298 | + ) |
| 299 | + return 0 |
| 300 | + |
| 301 | + |
| 302 | +if __name__ == "__main__": |
| 303 | + sys.exit(main()) |
0 commit comments