Cache split_tests pytest --collect-only results between CI runs (#171780)

Co-authored-by: Robert Resch <robert@resch.dev>

script/split_tests.py

@@ -2,9 +2,14 @@
 """Helper script to split test into n buckets."""
 
 import argparse
+from collections.abc import Iterator
 from concurrent.futures import ProcessPoolExecutor
-from dataclasses import dataclass, field
+from contextlib import suppress
+from dataclasses import dataclass, field, replace
+import hashlib
+import json
 from math import ceil
+from operator import attrgetter, itemgetter
 import os
 from pathlib import Path
 import subprocess
@@ -15,13 +20,21 @@ from typing import Final
 # place to subdivide to keep each pytest invocation roughly equal in size.
 _FAN_OUT_DIRS: Final = frozenset({"components"})
 
+# Cache file format version; bump on any incompatible schema change so old
+# caches are ignored rather than misread.
+_CACHE_VERSION: Final = 1
+
+# Fall back from file-level to directory-level pytest collection when
+# misses make up more than this fraction of the tree; past that point
+# the per-file argv overhead pytest pays outweighs the cost of letting
+# it re-walk dirs and re-collect the hits.
+_DIR_LEVEL_MISS_RATIO: Final = 0.3
+
 
 class Bucket:
     """Class to hold bucket."""
 
-    def __init__(
-        self,
-    ):
+    def __init__(self) -> None:
         """Initialize bucket."""
         self.total_tests = 0
         self._paths: list[str] = []
@@ -47,43 +60,55 @@ class BucketHolder:
         self._buckets: list[Bucket] = [Bucket() for _ in range(bucket_count)]
 
     def split_tests(self, test_folder: TestFolder) -> None:
-        """Split tests into buckets."""
+        """Place atomic units via best-fit; oversized ones go to the smallest bucket."""
         digits = len(str(test_folder.total_tests))
-        sorted_tests = sorted(
-            test_folder.get_all_flatten(), reverse=True, key=lambda x: x.total_tests
-        )
-        for tests in sorted_tests:
-            if tests.added_to_bucket:
-                # Already added to bucket
-                continue
+        by_load = attrgetter("total_tests")
+        units = sorted(self._atomic_units(test_folder), key=itemgetter(0), reverse=True)
+        for size, items in units:
+            fits = [
+                b
+                for b in self._buckets
+                if b.total_tests + size <= self._tests_per_bucket
+            ]
+            bucket = max(fits, key=by_load) if fits else min(self._buckets, key=by_load)
+            for item in items:
+                tag = " (same bucket)" if item is not items[0] else ""
+                print(f"{item.total_tests:>{digits}} tests in {item.path}{tag}")
+                bucket.add(item)
 
-            print(f"{tests.total_tests:>{digits}} tests in {tests.path}")
-            smallest_bucket = min(self._buckets, key=lambda x: x.total_tests)
-            is_file = isinstance(tests, TestFile)
-            if (
-                smallest_bucket.total_tests + tests.total_tests < self._tests_per_bucket
-            ) or is_file:
-                smallest_bucket.add(tests)
-                # Ensure all files from the same folder are in the same bucket
-                # to ensure that syrupy correctly identifies unused snapshots
-                if is_file:
-                    for other_test in tests.parent.children.values():
-                        if other_test is tests or isinstance(other_test, TestFolder):
-                            continue
-                        print(
-                            f"{other_test.total_tests:>{digits}}"
-                            f" tests in {other_test.path}"
-                            " (same bucket)"
-                        )
-                        smallest_bucket.add(other_test)
-
-        # verify that all tests are added to a bucket
         if not test_folder.added_to_bucket:
             raise ValueError("Not all tests are added to a bucket")
 
-    def create_ouput_file(self) -> None:
+    def _atomic_units(
+        self, folder: TestFolder
+    ) -> Iterator[tuple[int, list[TestFolder | TestFile]]]:
+        """Yield ``(size, items)`` placement units.
+
+        A folder that fits is one unit; otherwise same-dir files form
+        a unit only when the folder has syrupy snapshots, else each
+        file stands alone.  Sub-folders recurse independently.
+        """
+        if folder.total_tests <= self._tests_per_bucket:
+            yield folder.total_tests, [folder]
+            return
+
+        sibling_files = [c for c in folder.children.values() if isinstance(c, TestFile)]
+        if sibling_files:
+            if _has_snapshots(folder.path):
+                yield (
+                    sum(f.total_tests for f in sibling_files),
+                    list(sibling_files),
+                )
+            else:
+                for file in sibling_files:
+                    yield file.total_tests, [file]
+        for child in folder.children.values():
+            if isinstance(child, TestFolder):
+                yield from self._atomic_units(child)
+
+    def create_output_file(self) -> None:
         """Create output file."""
-        with Path("pytest_buckets.txt").open("w") as file:
+        with Path("pytest_buckets.txt").open("w", encoding="utf-8") as file:
             for idx, bucket in enumerate(self._buckets):
                 print(f"Bucket {idx + 1} has {bucket.total_tests} tests")
                 file.write(bucket.get_paths_line())
@@ -170,6 +195,15 @@ class TestFolder:
         return result
 
 
+def _has_snapshots(folder_path: Path) -> bool:
+    """Return True when ``folder_path/snapshots`` holds ``.ambr`` files.
+
+    Same-dir tests must share a pytest run so syrupy can spot unused
+    snapshots; without snapshots that constraint doesn't apply.
+    """
+    return any((folder_path / "snapshots").glob("*.ambr"))
+
+
 def _collect_batch(paths: list[Path]) -> tuple[str, str, int]:
     """Run pytest --collect-only on a batch of paths."""
     result = subprocess.run(
@@ -216,44 +250,339 @@ def _enumerate_batch_paths(path: Path) -> list[Path]:
     return paths
 
 
-def collect_tests(path: Path) -> TestFolder:
-    """Collect all tests."""
-    batch_paths = _enumerate_batch_paths(path)
-    if not batch_paths:
-        print(f"No eligible test paths found under {path}")
-        sys.exit(1)
-    workers = min(len(batch_paths), os.cpu_count() or 1) or 1
-    # Round-robin chunking keeps batches roughly balanced when path
-    # ordering correlates with test size.
-    batches = [batch_paths[i::workers] for i in range(workers)]
+def _hash_file(path: Path) -> str:
+    """Return a short content hash for ``path``."""
+    return hashlib.sha256(path.read_bytes()).hexdigest()[:16]
 
+
+def _walk_test_tree(root: Path) -> tuple[list[Path], list[Path]]:
+    """Walk ``root`` once and return (test files, fixture files).
+
+    Fixtures are every non-``test_*.py`` ``.py``: conftests and helpers
+    like ``common.py`` that drive parametrize imports.  Uses ``os.walk``
+    (~2x faster than ``Path.rglob`` on this tree) and prunes ``.``/``_``
+    subdirs.
+    """
+    test_files: list[Path] = []
+    fixtures: list[Path] = []
+    for dirpath, dirnames, filenames in os.walk(root):
+        dirnames[:] = [d for d in dirnames if not d.startswith((".", "_"))]
+        base = Path(dirpath)
+        for name in filenames:
+            if not name.endswith(".py"):
+                continue
+            if name.startswith("test_"):
+                test_files.append(base / name)
+            else:
+                fixtures.append(base / name)
+    test_files.sort()
+    fixtures.sort()
+    return test_files, fixtures
+
+
+_PROJECT_ROOT_MARKER: Final = "pyproject.toml"
+
+
+def _find_ancestor_fixtures(root: Path) -> list[Path]:
+    """Return non-``test_*.py`` Python files above ``root``, up to the project root.
+
+    Includes conftests and helper modules (eg ``common.py``); subtree
+    runs need both so shared ancestor helpers like
+    ``tests/components/common.py`` still invalidate descendants.
+    Stops at the first ancestor containing ``pyproject.toml`` so we
+    don't read unrelated ``.py`` files outside the repo or trip on
+    dirs we can't list.
+    """
+    fixtures: list[Path] = []
+    current = root.resolve().parent
+    while True:
+        with suppress(OSError):
+            fixtures.extend(
+                entry
+                for entry in current.glob("*.py")
+                if not entry.name.startswith("test_")
+            )
+        if (current / _PROJECT_ROOT_MARKER).exists():
+            break
+        if current == current.parent:
+            break
+        current = current.parent
+    return fixtures
+
+
+def _build_fixtures_by_dir(
+    root: Path, descendants: list[Path]
+) -> dict[Path, list[Path]]:
+    """Bucket descendants plus ancestor fixtures by resolved parent dir."""
+    by_dir: dict[Path, list[Path]] = {}
+    for fixture in (*_find_ancestor_fixtures(root), *descendants):
+        by_dir.setdefault(fixture.parent.resolve(), []).append(fixture)
+    return by_dir
+
+
+def _file_fixture_hash(
+    test_file: Path,
+    root: Path,
+    fixtures_by_dir: dict[Path, list[Path]],
+    blob_cache: dict[Path, bytes] | None = None,
+    dir_cache: dict[Path, str] | None = None,
+) -> str:
+    """Hash every ``.py`` fixture on the test file's ancestor path.
+
+    Catches conftests and helper modules (``common.py`` etc.) at any
+    level so parametrize imports from shared helpers invalidate
+    descendants, while sibling subtrees stay warm.  Pass shared
+    ``blob_cache``/``dir_cache`` dicts to memoize across many files.
+    """
+    test_dir = test_file.parent.resolve()
+    if dir_cache is not None and (cached := dir_cache.get(test_dir)) is not None:
+        return cached
+    relevant: list[Path] = []
+    current = test_dir
+    while True:
+        relevant.extend(fixtures_by_dir.get(current, ()))
+        parent = current.parent
+        if parent == current:
+            break
+        current = parent
+    relevant.sort()
+    digest = hashlib.sha256()
+    for fixture in relevant:
+        blob = blob_cache.get(fixture) if blob_cache is not None else None
+        if blob is None:
+            # relpath keeps the hash machine-stable across ancestor paths.
+            blob = (
+                os.path.relpath(fixture, root).encode()
+                + b"\0"
+                + fixture.read_bytes()
+                + b"\0"
+            )
+            if blob_cache is not None:
+                blob_cache[fixture] = blob
+        digest.update(blob)
+    result = digest.hexdigest()[:16]
+    if dir_cache is not None:
+        dir_cache[test_dir] = result
+    return result
+
+
+@dataclass
+class _CacheEntry:
+    """Cached test count plus its scope hash for a single file."""
+
+    hash: str
+    fixture_hash: str
+    count: int
+
+
+@dataclass
+class _Cache:
+    """Mapping of test file path → cached entry."""
+
+    entries: dict[str, _CacheEntry]
+
+    @classmethod
+    def load(cls, path: Path) -> _Cache:
+        """Load cache; any drift (missing, bad, version, malformed) returns empty."""
+        try:
+            raw = json.loads(path.read_bytes())
+        except OSError, ValueError:
+            raw = None
+        if not (
+            isinstance(raw, dict)
+            and raw.get("version") == _CACHE_VERSION
+            and isinstance(raw.get("files"), dict)
+        ):
+            return cls(entries={})
+        entries: dict[str, _CacheEntry] = {}
+        for key, value in raw["files"].items():
+            if not isinstance(value, dict):
+                continue
+            hash_value = value.get("hash")
+            fixture_hash = value.get("fixture_hash")
+            count = value.get("count")
+            if (
+                not isinstance(hash_value, str)
+                or not isinstance(fixture_hash, str)
+                or not isinstance(count, int)
+                or count < 0
+            ):
+                continue
+            entries[key] = _CacheEntry(
+                hash=hash_value, fixture_hash=fixture_hash, count=count
+            )
+        return cls(entries=entries)
+
+    def save(self, path: Path) -> None:
+        """Write the cache to ``path``, creating parent dirs as needed."""
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(
+            json.dumps(
+                {
+                    "version": _CACHE_VERSION,
+                    "files": {
+                        key: {
+                            "hash": entry.hash,
+                            "fixture_hash": entry.fixture_hash,
+                            "count": entry.count,
+                        }
+                        for key, entry in sorted(self.entries.items())
+                    },
+                },
+                indent=2,
+                ensure_ascii=False,
+            )
+            + "\n",
+            encoding="utf-8",
+        )
+
+
+def _resolve_entries(
+    test_files: list[Path],
+    cache: _Cache,
+    root: Path,
+    fixtures_by_dir: dict[Path, list[Path]],
+) -> tuple[dict[Path, _CacheEntry], list[Path]]:
+    """Build an entry for every file; return ``(entries, misses)``.
+
+    Hits reuse the stored entry; misses get fresh hashes with a
+    count=0 placeholder for the caller to fill in after pytest runs.
+    Shared caches memoize fixture blobs and per-dir hashes so each
+    fixture file is read once and each unique dir hashed once.
+    """
+    blob_cache: dict[Path, bytes] = {}
+    dir_cache: dict[Path, str] = {}
+    entries: dict[Path, _CacheEntry] = {}
+    misses: list[Path] = []
+    for file in test_files:
+        file_hash = _hash_file(file)
+        fixture_hash = _file_fixture_hash(
+            file, root, fixtures_by_dir, blob_cache, dir_cache
+        )
+        cached = cache.entries.get(str(file.relative_to(root)))
+        if (
+            cached is not None
+            and cached.hash == file_hash
+            and cached.fixture_hash == fixture_hash
+        ):
+            entries[file] = cached
+        else:
+            entries[file] = _CacheEntry(
+                hash=file_hash, fixture_hash=fixture_hash, count=0
+            )
+            misses.append(file)
+    return entries, misses
+
+
+def _run_collect_batches(paths: list[Path]) -> list[tuple[str, str, int]]:
+    """Run pytest --collect-only across ``paths`` using a process pool."""
+    workers = min(len(paths), os.cpu_count() or 1) or 1
+    batches = [paths[i::workers] for i in range(workers)]
     if workers == 1:
-        results = [_collect_batch(batches[0])]
-    else:
-        with ProcessPoolExecutor(max_workers=workers) as executor:
-            results = list(executor.map(_collect_batch, batches))
+        return [_collect_batch(batches[0])]
+    with ProcessPoolExecutor(max_workers=workers) as executor:
+        return list(executor.map(_collect_batch, batches))
 
-    folder = TestFolder(path)
-    for stdout, stderr, returncode in results:
+
+def _parse_collect_output(stdout: str) -> dict[Path, int]:
+    """Parse ``pytest --collect-only -qq`` output into ``{path: count}``."""
+    counts: dict[Path, int] = {}
+    for line in stdout.splitlines():
+        if not line.strip():
+            continue
+        file_path, _, total_tests = line.partition(": ")
+        if not file_path or not total_tests:
+            raise ValueError(f"Unexpected line: {line}")
+        counts[Path(file_path)] = int(total_tests)
+    return counts
+
+
+def _run_pytest_collect(paths: list[Path]) -> dict[Path, int]:
+    """Run pytest --collect-only across ``paths`` and parse the output."""
+    counts: dict[Path, int] = {}
+    for stdout, stderr, returncode in _run_collect_batches(paths):
         if returncode != 0:
             print("Failed to collect tests:")
             print(stderr)
             print(stdout)
             sys.exit(1)
-        for line in stdout.splitlines():
-            if not line.strip():
-                continue
-            file_path, _, total_tests = line.partition(": ")
-            if not file_path or not total_tests:
-                print(f"Unexpected line: {line}")
-                sys.exit(1)
+        # Surface stderr from successful runs too; pytest puts deprecation
+        # and import warnings here that would otherwise vanish.
+        if stderr.strip():
+            sys.stderr.write(stderr)
+        try:
+            counts.update(_parse_collect_output(stdout))
+        except ValueError as err:
+            print(err)
+            sys.exit(1)
+    return counts
 
-            file = TestFile(int(total_tests), Path(file_path))
-            folder.add_test_file(file)
 
+def _build_folder(root: Path, counts: dict[Path, int]) -> TestFolder:
+    """Build a ``TestFolder`` from ``{path: count}``; zero-count files are skipped."""
+    folder = TestFolder(root)
+    for file_path, count in counts.items():
+        if count:
+            folder.add_test_file(TestFile(count, file_path))
     return folder
 
 
+def _exit_if_empty(paths: list[Path], root: Path) -> None:
+    """Exit with a clear message when no eligible test paths were found."""
+    if not paths:
+        print(f"No eligible test paths found under {root}")
+        sys.exit(1)
+
+
+def _collect_tests_uncached(path: Path) -> TestFolder:
+    """Hand pytest the top-level dirs; the pre-cache path when ``--cache`` is unset."""
+    batch_paths = _enumerate_batch_paths(path)
+    _exit_if_empty(batch_paths, path)
+    return _build_folder(path, _run_pytest_collect(batch_paths))
+
+
+def _collect_tests_cached(path: Path, cache_path: Path) -> TestFolder:
+    """Collect tests using an on-disk cache for incremental updates."""
+    all_test_files, fixtures = _walk_test_tree(path)
+    _exit_if_empty(all_test_files, path)
+
+    fixtures_by_dir = _build_fixtures_by_dir(path, fixtures)
+    cache = _Cache.load(cache_path)
+    entries, misses = _resolve_entries(all_test_files, cache, path, fixtures_by_dir)
+    hits = len(all_test_files) - len(misses)
+    print(f"Cache: {hits} hits / {len(misses)} misses / {len(all_test_files)} total")
+
+    if misses:
+        # Past _DIR_LEVEL_MISS_RATIO the per-file argv overhead beats
+        # re-walking the dirs, so fall back to dir-level collection.
+        if not hits or len(misses) > len(all_test_files) * _DIR_LEVEL_MISS_RATIO:
+            collect_paths = _enumerate_batch_paths(path)
+        else:
+            collect_paths = misses
+        new_counts = _run_pytest_collect(collect_paths)
+        # Files pytest returned no count for stay at 0; cached so they
+        # aren't re-collected next run.
+        for file in misses:
+            entries[file] = replace(entries[file], count=new_counts.get(file, 0))
+
+    _Cache(entries={str(f.relative_to(path)): e for f, e in entries.items()}).save(
+        cache_path
+    )
+    return _build_folder(path, {f: e.count for f, e in entries.items()})
+
+
+def collect_tests(path: Path, cache_path: Path | None = None) -> TestFolder:
+    """Collect all tests, using an on-disk cache when ``cache_path`` is set."""
+    if path.is_file():
+        # TestFolder requires a directory root; a single file has no
+        # parent to anchor against and CI never invokes us this way.
+        print(f"Expected a directory, got file: {path}")
+        sys.exit(1)
+    if cache_path is None:
+        return _collect_tests_uncached(path)
+    return _collect_tests_cached(path, cache_path)
+
+
 def main() -> None:
     """Execute script."""
     parser = argparse.ArgumentParser(description="Split tests into n buckets.")
@@ -276,11 +605,17 @@ def main() -> None:
         help="Path to the test files to split into buckets",
         type=Path,
     )
+    parser.add_argument(
+        "--cache",
+        help="Path to a JSON file used to cache per-file test counts",
+        type=Path,
+        default=None,
+    )
 
     arguments = parser.parse_args()
 
     print("Collecting tests...")
-    tests = collect_tests(arguments.path)
+    tests = collect_tests(arguments.path, arguments.cache)
     tests_per_bucket = ceil(tests.total_tests / arguments.bucket_count)
 
     bucket_holder = BucketHolder(tests_per_bucket, arguments.bucket_count)
@@ -290,7 +625,7 @@ def main() -> None:
     print(f"Total tests: {tests.total_tests}")
     print(f"Estimated tests per bucket: {tests_per_bucket}")
 
-    bucket_holder.create_ouput_file()
+    bucket_holder.create_output_file()
 
 
 if __name__ == "__main__":