diff --git a/scripts/lint_base_counts.py b/scripts/lint_base_counts.py index 4f66c0a69e4..2b86d4ce55b 100644 --- a/scripts/lint_base_counts.py +++ b/scripts/lint_base_counts.py @@ -81,41 +81,48 @@ def cache_key(base_point: str, fingerprints: Sequence[str]) -> str: return hashlib.sha256("|".join((base_point, *fingerprints)).encode()).hexdigest()[:16] -def _git(args: Sequence[str], cwd: Path) -> str: - proc: Final = subprocess.run(["git", *args], cwd=cwd, capture_output=True, text=True) - if proc.returncode not in (0, 1): - sys.stderr.write(proc.stderr) - raise SystemExit(f"git exited {proc.returncode}") - return proc.stdout +Git: TypeAlias = Callable[[Sequence[str]], str] -def head_sha(cwd: Path = REPO_ROOT) -> str: - return _git(["rev-parse", "HEAD"], cwd).strip() +def git_in(cwd: Path) -> Git: + def run(args: Sequence[str]) -> str: + proc: Final = subprocess.run(["git", *args], cwd=cwd, capture_output=True, text=True) + if proc.returncode not in (0, 1): + sys.stderr.write(proc.stderr) + raise SystemExit(f"git exited {proc.returncode}") + return proc.stdout + + return run -def resolve_base_point(base_ref: str, cwd: Path = REPO_ROOT) -> str: +REPO_GIT: Final = git_in(REPO_ROOT) + + +def head_sha(git: Git = REPO_GIT) -> str: + return git(["rev-parse", "HEAD"]).strip() + + +def resolve_base_point(base_ref: str, git: Git = REPO_GIT) -> str: """The snapshot commit base counts are measured at: merge-base(base_ref, HEAD), made aware of an in-progress merge. Mid-merge, HEAD is still the pre-merge tip, so its merge-base is the old branch point and every violation the base gained since then would be blamed on this change. While MERGE_HEAD exists, prefer merge-base(base_ref, MERGE_HEAD) whenever it is the newer of the two.""" - head_point: Final = _git(["merge-base", base_ref, "HEAD"], cwd).strip() + head_point: Final = git(["merge-base", base_ref, "HEAD"]).strip() if not head_point: return base_ref - merge_head: Final = _git(["rev-parse", "--verify", "--quiet", "MERGE_HEAD"], cwd).strip() + merge_head: Final = git(["rev-parse", "--verify", "--quiet", "MERGE_HEAD"]).strip() if not merge_head: return head_point - merge_point: Final = _git(["merge-base", base_ref, merge_head], cwd).strip() + merge_point: Final = git(["merge-base", base_ref, merge_head]).strip() if not merge_point: return head_point - older: Final = _git(["merge-base", head_point, merge_point], cwd).strip() + older: Final = git(["merge-base", head_point, merge_point]).strip() return merge_point if older == head_point else head_point -def default_cache_dir(cwd: Path = REPO_ROOT) -> Path: - common: Final = Path(_git(["rev-parse", "--git-common-dir"], cwd).strip()) - resolved: Final = common if common.is_absolute() else cwd / common - return resolved / CACHE_DIR_NAME +def default_cache_dir(git: Git = REPO_GIT) -> Path: + return Path(git(["rev-parse", "--path-format=absolute", "--git-common-dir"]).strip()) / CACHE_DIR_NAME def validated_counts(data: object) -> Counts | None: diff --git a/tests/unit/test_lint_base_counts.py b/tests/unit/test_lint_base_counts.py index 5abb2c8f598..84f8a0936c0 100644 --- a/tests/unit/test_lint_base_counts.py +++ b/tests/unit/test_lint_base_counts.py @@ -7,11 +7,10 @@ import fnmatch import io import json import os -import subprocess import zipfile from collections.abc import Callable, Mapping, Sequence from pathlib import Path -from typing import Final, NoReturn +from typing import Final, NamedTuple, NoReturn import pytest @@ -148,8 +147,7 @@ def test_store_evicts_only_the_oldest_entries_beyond_the_cap(tmp_path: Path) -> def test_store_never_evicts_the_entry_it_just_wrote_even_on_mtime_ties(tmp_path: Path) -> None: for index in range(counts.CACHE_KEEP_ENTRIES + 2): - path = counts.store_counts(tmp_path, _CHECKER, f"base{index}", {"reportAny": 1}) - os.utime(path, (9_999_999_999, 9_999_999_999)) + os.utime(counts.store_counts(tmp_path, _CHECKER, f"base{index}", {"reportAny": 1}), (9_999_999_999,) * 2) mine: Final = counts.store_counts(tmp_path, _CHECKER, "mine", {"reportAny": 2}) assert counts.load_cached_counts(mine) == {"reportAny": 2} assert len(list(tmp_path.glob(_CHECKER.cache_glob()))) == counts.CACHE_KEEP_ENTRIES @@ -394,69 +392,63 @@ def test_emitted_file_is_the_one_the_fetcher_looks_up(tmp_path: Path) -> None: } -def _git(cwd: Path, *args: str) -> str: - proc: Final = subprocess.run(["git", *args], cwd=cwd, capture_output=True, text=True) - assert proc.returncode == 0, proc.stderr - return proc.stdout.strip() +class _History(NamedTuple): + parents: Mapping[str, tuple[str, ...]] + refs: Mapping[str, str] + + def ancestry(self, commit: str) -> frozenset[str]: + return frozenset((commit,)).union(*(self.ancestry(parent) for parent in self.parents[commit])) + + def merge_base(self, left: str, right: str) -> str: + common: Final = self.ancestry(self.refs.get(left, left)) & self.ancestry(self.refs.get(right, right)) + return next(c for c in common if not any(c != other and c in self.ancestry(other) for other in common)) -def _commit(cwd: Path, name: str) -> str: - (cwd / name).write_text(name) - _git(cwd, "add", "-A") - _git(cwd, "commit", "-q", "-m", name) - return _git(cwd, "rev-parse", "HEAD") +def _git_over(history: _History) -> counts.Git: + def git(args: Sequence[str]) -> str: + match tuple(args): + case ("merge-base", left, right): + return f"{history.merge_base(left, right)}\n" + case ("rev-parse", "HEAD"): + return f"{history.refs['HEAD']}\n" + case ("rev-parse", "--verify", "--quiet", ref): + return f"{history.refs[ref]}\n" if ref in history.refs else "" + case ("rev-parse", "--path-format=absolute", "--git-common-dir"): + return "/repo/.git\n" + case ("rev-parse", "--path-format=absolute", "--git-dir"): + return "/repo/.git/worktrees/feature\n" + case _: + raise AssertionError(f"unexpected git call: {args}") + + return git -def _init_repo(tmp_path: Path) -> Path: - repo: Final = tmp_path / "repo" - repo.mkdir() - _git(repo, "init", "-q", "-b", "main") - _git(repo, "config", "user.email", "gate@example.com") - _git(repo, "config", "user.name", "gate") - _git(repo, "config", "commit.gpgsign", "false") - return repo +_FEATURE_OFF_MAIN: Final = _History( + parents={"shared": (), "feature": ("shared",), "drift": ("shared",)}, + refs={"main": "drift", "HEAD": "feature"}, +) -def _branched_repo(tmp_path: Path) -> tuple[Path, str, str]: - repo: Final = _init_repo(tmp_path) - branch_point: Final = _commit(repo, "shared.txt") - _git(repo, "checkout", "-q", "-b", "feature") - _commit(repo, "feature.txt") - _git(repo, "checkout", "-q", "main") - base_tip: Final = _commit(repo, "drift.txt") - _git(repo, "checkout", "-q", "feature") - return repo, branch_point, base_tip +def test_base_point_is_the_branch_point_when_no_merge_is_in_progress() -> None: + assert counts.resolve_base_point("main", _git_over(_FEATURE_OFF_MAIN)) == "shared" -def test_base_point_is_the_branch_point_when_no_merge_is_in_progress(tmp_path: Path) -> None: - repo, branch_point, _ = _branched_repo(tmp_path) - assert counts.resolve_base_point("main", cwd=repo) == branch_point +def test_base_point_mid_merge_advances_to_the_merged_in_base_tip() -> None: + merging_main: Final = _FEATURE_OFF_MAIN._replace(refs={**_FEATURE_OFF_MAIN.refs, "MERGE_HEAD": "drift"}) + assert counts.resolve_base_point("main", _git_over(merging_main)) == "drift" -def test_base_point_mid_merge_advances_to_the_merged_in_base_tip(tmp_path: Path) -> None: - repo, _, base_tip = _branched_repo(tmp_path) - _git(repo, "merge", "--no-commit", "--no-ff", "main") - assert counts.resolve_base_point("main", cwd=repo) == base_tip +def test_base_point_mid_merge_of_an_older_side_branch_keeps_the_newer_branch_point() -> None: + merging_old_side: Final = _History( + parents={"shared": (), "old": ("shared",), "drift": ("shared",), "feature": ("drift",)}, + refs={"main": "drift", "HEAD": "feature", "MERGE_HEAD": "old"}, + ) + assert counts.resolve_base_point("main", _git_over(merging_old_side)) == "drift" -def test_base_point_mid_merge_of_an_older_side_branch_keeps_the_newer_branch_point(tmp_path: Path) -> None: - repo: Final = _init_repo(tmp_path) - _commit(repo, "shared.txt") - _git(repo, "checkout", "-q", "-b", "old-side") - _commit(repo, "old.txt") - _git(repo, "checkout", "-q", "main") - newer_point: Final = _commit(repo, "drift.txt") - _git(repo, "checkout", "-q", "-b", "feature") - _commit(repo, "feature.txt") - _git(repo, "merge", "--no-commit", "--no-ff", "old-side") - assert counts.resolve_base_point("main", cwd=repo) == newer_point +def test_head_sha_is_the_checked_out_commit() -> None: + assert counts.head_sha(_git_over(_FEATURE_OFF_MAIN)) == "feature" -def test_head_sha_is_the_checked_out_commit(tmp_path: Path) -> None: - repo, _, _ = _branched_repo(tmp_path) - assert counts.head_sha(cwd=repo) == _git(repo, "rev-parse", "HEAD") - - -def test_default_cache_dir_lives_under_the_shared_git_dir(tmp_path: Path) -> None: - repo, _, _ = _branched_repo(tmp_path) - assert counts.default_cache_dir(cwd=repo) == repo / ".git" / counts.CACHE_DIR_NAME +def test_default_cache_dir_is_shared_by_every_worktree() -> None: + assert counts.default_cache_dir(_git_over(_FEATURE_OFF_MAIN)) == Path("/repo/.git") / counts.CACHE_DIR_NAME