From 477d71653a9a4291bf6362f3185a8b323d8db401 Mon Sep 17 00:00:00 2001 From: nia-sg-bot Date: Sat, 15 Aug 2026 09:04:54 +0530 Subject: [PATCH] feat(git): resolve exact two-dot and three-dot ranges --- diffgraph/git_snapshot.py | 151 ++++++++++++++++++++++++++++++++++++- tests/test_git_snapshot.py | 115 +++++++++++++++++++++++++++- 2 files changed, 261 insertions(+), 5 deletions(-) diff --git a/diffgraph/git_snapshot.py b/diffgraph/git_snapshot.py index f77676d..f92c04d 100644 --- a/diffgraph/git_snapshot.py +++ b/diffgraph/git_snapshot.py @@ -1,13 +1,14 @@ """Resolve exact Git object identities for index and working-tree changes. -This module deliberately models only the two local snapshot pairs: +This module models the two local snapshot pairs and immutable commit ranges: * staged: ``HEAD`` -> index * unstaged: index -> working tree +* two-dot: the requested base commit -> the requested head commit +* three-dot: the merge base of the requested commits -> the requested head -Commit/range resolution belongs to a separate layer. Paths returned by Git are -read using its NUL-delimited raw format, so tabs, newlines, and other unusual -filename bytes are not delimiters. +Paths returned by Git are read using its NUL-delimited raw format, so tabs, +newlines, and other unusual filename bytes are not delimiters. """ from __future__ import annotations @@ -55,6 +56,25 @@ class SnapshotResolution: warnings: Tuple[ResolutionWarning, ...] +@dataclass(frozen=True) +class CommitRangeResolution: + """Exact endpoints and entries for a two-dot or three-dot comparison. + + ``comparison_base_oid`` equals ``base_oid`` for two-dot comparisons and + the resolved merge-base object ID for three-dot comparisons. Recording + both prevents display refs from being mistaken for immutable provenance. + """ + + entries: Tuple[SnapshotEntry, ...] + warnings: Tuple[ResolutionWarning, ...] + base_ref: str + head_ref: str + base_oid: Optional[str] + head_oid: Optional[str] + comparison_base_oid: Optional[str] + three_dot: bool + + class GitSnapshotError(RuntimeError): """A Git or working-tree operation required for exact snapshots failed.""" @@ -97,6 +117,129 @@ def resolve_unstaged( return _resolve(repository, pathspecs, staged=False) +def resolve_commit_range( + repository: str, + base_ref: str, + head_ref: str, + *, + three_dot: bool = False, + pathspecs: Optional[Sequence[str]] = None, +) -> CommitRangeResolution: + """Resolve exact tree identities for a two-dot or three-dot comparison. + + Refs are resolved to commits before diffing. Two-dot compares those two + commits directly; three-dot compares their merge base with the resolved + head, matching ``git diff base...head`` semantics. Failures are returned + as structured warnings and never as fabricated changes. + """ + + warnings: List[ResolutionWarning] = [] + root = _repository_root(repository, warnings) + if root is None: + return _commit_range_result(base_ref, head_ref, three_dot, warnings=warnings) + + base_oid = _resolve_commit(root, base_ref, "invalid_base_ref", warnings) + head_oid = _resolve_commit(root, head_ref, "invalid_head_ref", warnings) + if base_oid is None or head_oid is None: + return _commit_range_result( + base_ref, head_ref, three_dot, warnings=warnings, + base_oid=base_oid, head_oid=head_oid, + ) + + comparison_base_oid = base_oid + if three_dot: + output = _run( + ["git", "merge-base", base_oid, head_oid], root, warnings, + "merge_base_failed", + ) + if output is None: + return _commit_range_result( + base_ref, head_ref, three_dot, warnings=warnings, + base_oid=base_oid, head_oid=head_oid, + ) + comparison_base_oid = os.fsdecode(output).strip() + if not _is_hex_oid(comparison_base_oid): + warnings.append(ResolutionWarning( + "malformed_merge_base", + "Git returned an invalid merge-base object ID", + )) + return _commit_range_result( + base_ref, head_ref, three_dot, warnings=warnings, + base_oid=base_oid, head_oid=head_oid, + ) + + command = [ + "git", "diff", "--raw", "-z", "--no-abbrev", "--no-ext-diff", + "--find-renames=50%", comparison_base_oid, head_oid, + ] + scoped_pathspecs = _root_relative_pathspecs(repository, root, pathspecs) + if scoped_pathspecs: + command.append("--") + command.extend(scoped_pathspecs) + output = _run(command, root, warnings, "git_diff_failed") + entries: List[SnapshotEntry] = [] + if output is not None: + for raw in _parse_raw(output, warnings): + entry = _exact_staged_entry(raw, warnings) + if entry is not None: + entries.append(entry) + entries.sort(key=_entry_sort_key) + + return _commit_range_result( + base_ref, head_ref, three_dot, entries=entries, warnings=warnings, + base_oid=base_oid, head_oid=head_oid, + comparison_base_oid=comparison_base_oid, + ) + + +def _commit_range_result( + base_ref: str, + head_ref: str, + three_dot: bool, + *, + entries: Sequence[SnapshotEntry] = (), + warnings: Sequence[ResolutionWarning] = (), + base_oid: Optional[str] = None, + head_oid: Optional[str] = None, + comparison_base_oid: Optional[str] = None, +) -> CommitRangeResolution: + return CommitRangeResolution( + tuple(entries), tuple(warnings), base_ref, head_ref, base_oid, head_oid, + comparison_base_oid, three_dot, + ) + + +def _resolve_commit( + root: str, ref: str, warning_code: str, + warnings: List[ResolutionWarning], +) -> Optional[str]: + output = _run( + [ + "git", + "rev-parse", + "--verify", + "--end-of-options", + "{}^{{commit}}".format(ref), + ], + root, warnings, warning_code, + ) + if output is None: + return None + oid = os.fsdecode(output).strip() + if not _is_hex_oid(oid): + warnings.append(ResolutionWarning( + warning_code, "Git returned an invalid commit object ID" + )) + return None + return oid + + +def _is_hex_oid(value: str) -> bool: + return bool(value) and all( + character in "0123456789abcdef" for character in value + ) + + def _resolve( repository: str, pathspecs: Optional[Sequence[str]], staged: bool ) -> SnapshotResolution: diff --git a/tests/test_git_snapshot.py b/tests/test_git_snapshot.py index 4037b02..eed0bf6 100644 --- a/tests/test_git_snapshot.py +++ b/tests/test_git_snapshot.py @@ -1,7 +1,11 @@ import os import subprocess -from diffgraph.git_snapshot import resolve_staged, resolve_unstaged +from diffgraph.git_snapshot import ( + resolve_commit_range, + resolve_staged, + resolve_unstaged, +) def git(repo, *args, input_bytes=None): @@ -288,3 +292,112 @@ def test_unstaged_symlink_hashes_raw_target_without_attributes(tmp_path): assert entry.new_oid == git( repo, "hash-object", "--stdin", input_bytes=b"replacement.py" ).decode("ascii").strip() + + +def test_two_dot_range_records_exact_endpoints_and_tree_identities(tmp_path): + repo = make_repo(tmp_path) + write(repo, "shared.txt", b"common\n") + commit_all(repo, "common") + git(repo, "branch", "base") + + write(repo, "head-only.txt", b"head\n") + commit_all(repo, "head change") + git(repo, "branch", "head") + head_oid = oid(repo, "head") + + git(repo, "switch", "base") + write(repo, "base-only.txt", b"base\n") + commit_all(repo, "base change") + base_oid = oid(repo, "base") + + result = resolve_commit_range(str(repo), "base", "head") + entries = {entry.new_path or entry.old_path: entry for entry in result.entries} + + assert result.warnings == () + assert result.base_oid == result.comparison_base_oid == base_oid + assert result.head_oid == head_oid + assert result.three_dot is False + assert set(entries) == {"base-only.txt", "head-only.txt"} + assert entries["base-only.txt"].status == "D" + assert entries["base-only.txt"].old_oid == oid(repo, "base:base-only.txt") + assert entries["base-only.txt"].new_oid is None + assert entries["head-only.txt"].status == "A" + assert entries["head-only.txt"].old_oid is None + assert entries["head-only.txt"].new_oid == oid(repo, "head:head-only.txt") + + +def test_three_dot_range_uses_merge_base_and_excludes_base_only_changes(tmp_path): + repo = make_repo(tmp_path) + write(repo, "shared.txt", b"common\n") + commit_all(repo, "common") + common_oid = oid(repo, "HEAD") + git(repo, "branch", "base") + + write(repo, "head-only.txt", b"head\n") + commit_all(repo, "head change") + git(repo, "branch", "head") + + git(repo, "switch", "base") + write(repo, "base-only.txt", b"base\n") + commit_all(repo, "base change") + + result = resolve_commit_range(str(repo), "base", "head", three_dot=True) + + assert result.warnings == () + assert result.comparison_base_oid == common_oid + assert result.base_oid == oid(repo, "base") + assert result.head_oid == oid(repo, "head") + assert result.three_dot is True + assert [entry.new_path or entry.old_path for entry in result.entries] == [ + "head-only.txt" + ] + + +def test_commit_range_pathspec_is_relative_to_calling_subdirectory(tmp_path): + repo = make_repo(tmp_path) + write(repo, "inside/a.txt", b"old a\n") + write(repo, "outside/b.txt", b"old b\n") + commit_all(repo) + git(repo, "branch", "before") + write(repo, "inside/a.txt", b"new a\n") + write(repo, "outside/b.txt", b"new b\n") + commit_all(repo, "both changed") + + result = resolve_commit_range( + str(repo / "inside"), "before", "HEAD", pathspecs=["a.txt"] + ) + + assert result.warnings == () + assert [entry.new_path for entry in result.entries] == ["inside/a.txt"] + + +def test_invalid_commit_range_ref_is_a_warning_not_a_change(tmp_path): + repo = make_repo(tmp_path) + write(repo, "tracked.txt", b"content\n") + commit_all(repo) + + result = resolve_commit_range(str(repo), "missing-ref", "HEAD") + + assert result.entries == () + assert result.base_oid is None + assert result.head_oid == oid(repo, "HEAD") + assert [warning.code for warning in result.warnings] == ["invalid_base_ref"] + + +def test_three_dot_without_merge_base_is_a_warning_not_a_change(tmp_path): + repo = make_repo(tmp_path) + write(repo, "first.txt", b"first\n") + commit_all(repo) + git(repo, "branch", "first") + git(repo, "switch", "--orphan", "second") + first_path = repo / "first.txt" + if first_path.exists(): + first_path.unlink() + write(repo, "second.txt", b"second\n") + commit_all(repo, "unrelated root") + + result = resolve_commit_range(str(repo), "first", "second", three_dot=True) + + assert result.entries == () + assert result.comparison_base_oid is None + assert [warning.code for warning in result.warnings] == ["merge_base_failed"]