git_status: parse one porcelain v2 worktree snapshot Add a strict, NUL-aware parser for porcelain v2 records, including ordinary changes, renames, conflicts, untracked paths, branch divergence, and stash metadata. The reader is gated on Git 2.11, opts into newer flags only when supported, and preserves arbitrary pathname bytes. Bug: 543851900 Bug: 553599402 Change-Id: Ibae3e1056fd9866f3cb6d490745d377cda1e8fef Reviewed-on: https://gerrit-review.googlesource.com/c/git-repo/+/632142 Tested-by: Gavin Mak <gavinmak@google.com> Commit-Queue: Gavin Mak <gavinmak@google.com> Reviewed-by: Brian Gan <brgan@google.com>
diff --git a/git_status.py b/git_status.py new file mode 100644 index 0000000..2570314 --- /dev/null +++ b/git_status.py
@@ -0,0 +1,257 @@ +# Copyright (C) 2026 The Android Open Source Project +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Read a worktree's state from one machine-readable git status snapshot.""" + +from collections import OrderedDict +import os +from typing import Iterator, List, Optional, TYPE_CHECKING + +from git_command import git_require +from git_command import GitCommand + + +if TYPE_CHECKING: + from project import Project + + +class StatusEntry: + """The state of one path on one side of the index.""" + + def __init__( + self, + path: str, + status: str, + src_path: Optional[str] = None, + level: Optional[str] = None, + ) -> None: + self.path = path + self.status = status + self.src_path = src_path + self.level = level + + +class StatusSnapshot: + """A consistent view of worktree, index, and branch state.""" + + def __init__(self) -> None: + self.index_changes = OrderedDict() + self.worktree_changes = OrderedDict() + self.untracked = [] + self.branch_oid = None + self.branch_head = None + self.upstream = None + self.ahead = 0 + self.behind = 0 + self.has_ahead_behind = False + self.stash_count = 0 + + @property + def current_branch(self) -> Optional[str]: + if self.branch_head in (None, "(detached)", "(unknown)"): + return None + return self.branch_head + + def is_dirty(self, consider_untracked: bool = True) -> bool: + return bool( + self.index_changes + or self.worktree_changes + or (consider_untracked and self.untracked) + ) + + +def GetStatus( + project: "Project", + gitdir: str, + untracked_files: str = "all", + branch: bool = False, + ahead_behind: bool = False, + show_stash: bool = False, +) -> StatusSnapshot: + """Return one machine-readable status snapshot for |project|.""" + if not git_require((2, 11, 0)): + raise UnsupportedStatusError("porcelain v2 requires Git 2.11") + cmd = [ + "status", + "--porcelain=v2", + "-z", + "--ignore-submodules=all", + f"--untracked-files={untracked_files}", + ] + if branch: + cmd.append("--branch") + if ahead_behind and git_require((2, 17, 0)): + cmd.append("--ahead-behind") + if git_require((2, 18, 0)): + # Match the existing staged diff's explicit rename detection even if + # status.renames is disabled in the user's config. + cmd.append("--renames") + if show_stash and git_require((2, 35, 0)): + cmd.append("--show-stash") + + p = GitCommand( + project, + cmd, + bare=False, + gitdir=gitdir, + capture_stdout=True, + capture_stdout_bytes=True, + capture_stderr=True, + verify_command=True, + ) + p.Wait() + return ParsePorcelainV2(p.stdout) + + +def _Path(value: bytes) -> str: + """Decode a Git pathname without losing undecodable bytes.""" + return os.fsdecode(value) + + +def _Status(value: int) -> str: + """Normalize Git's unchanged markers for repo's status display.""" + char = chr(value) + return "" if char == "." else char + + +def _Records(output: bytes) -> Iterator[bytes]: + if not output: + return iter(()) + if not output.endswith(b"\0"): + raise StatusParseError("porcelain v2 output is not NUL terminated") + records = output.split(b"\0") + if not records[-1]: + records.pop() + return iter(records) + + +class StatusParseError(ValueError): + """Raised when machine-readable status output is malformed.""" + + +class UnsupportedStatusError(RuntimeError): + """Raised when the Git client cannot produce porcelain v2.""" + + +def _Fields(record: bytes, count: int) -> List[bytes]: + fields = record.split(b" ", count - 1) + if len(fields) != count: + raise StatusParseError(f"malformed porcelain v2 record: {record!r}") + return fields + + +def _AddTracked( + status: StatusSnapshot, + path: str, + xy: bytes, + src_path: Optional[str] = None, + level: Optional[str] = None, +) -> None: + index_status = _Status(xy[0]) + worktree_status = _Status(xy[1]) + if index_status: + status.index_changes[path] = StatusEntry( + path, + index_status, + src_path=src_path if index_status in ("R", "C") else None, + level=level if index_status in ("R", "C") else None, + ) + if worktree_status: + status.worktree_changes[path] = StatusEntry( + path, + worktree_status, + src_path=src_path if worktree_status in ("R", "C") else None, + level=level if worktree_status in ("R", "C") else None, + ) + + +def ParsePorcelainV2(output: bytes) -> StatusSnapshot: + """Parse ``git status --porcelain=v2 -z --branch`` output.""" + status = StatusSnapshot() + records = _Records(output) + for record in records: + kind = record[:1] + if kind == b"#": + try: + key, value = record[2:].split(b" ", 1) + except ValueError as e: + raise StatusParseError( + f"malformed porcelain v2 header: {record!r}" + ) from e + if key == b"branch.oid": + value = value.decode("ascii") + status.branch_oid = None if value == "(initial)" else value + elif key == b"branch.head": + status.branch_head = _Path(value) + elif key == b"branch.upstream": + status.upstream = _Path(value) + elif key == b"branch.ab": + try: + value = value.decode("ascii") + ahead, behind = value.split() + if ahead != "+?" and behind != "-?": + status.ahead = int(ahead) + status.behind = -int(behind) + status.has_ahead_behind = True + except ValueError as e: + raise StatusParseError( + f"malformed porcelain v2 branch.ab record: {record!r}" + ) from e + elif key == b"stash": + status.stash_count = int(value.decode("ascii")) + continue + + if kind == b"1": + fields = _Fields(record, 9) + xy = fields[1] + if len(xy) != 2: + raise StatusParseError(f"invalid status pair: {xy!r}") + path = _Path(fields[8]) + _AddTracked(status, path, xy) + elif kind == b"2": + fields = _Fields(record, 10) + xy = fields[1] + if len(xy) != 2: + raise StatusParseError(f"invalid status pair: {xy!r}") + score = fields[8][1:].lstrip(b"0") or b"0" + try: + src_path = _Path(next(records)) + except StopIteration as e: + raise StatusParseError( + "rename record has no source path" + ) from e + path = _Path(fields[9]) + _AddTracked( + status, + path, + xy, + src_path=src_path, + level=score.decode("ascii"), + ) + elif kind == b"u": + fields = _Fields(record, 11) + path = _Path(fields[10]) + # The old diff-index/diff-files pair reported unmerged paths on + # both sides, regardless of porcelain's more specific XY pair. + status.index_changes[path] = StatusEntry(path, "U") + status.worktree_changes[path] = StatusEntry(path, "U") + elif kind == b"?": + status.untracked.append(_Path(record[2:])) + elif kind == b"!": + continue + else: + raise StatusParseError( + f"unknown porcelain v2 record type: {record!r}" + ) + return status
diff --git a/tests/test_git_status.py b/tests/test_git_status.py new file mode 100644 index 0000000..72698db --- /dev/null +++ b/tests/test_git_status.py
@@ -0,0 +1,210 @@ +# Copyright (C) 2026 The Android Open Source Project +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Unittests for the git_status.py module.""" + +import os +from typing import Any, List +from unittest import mock + +import pytest + +import git_status + + +def test_parse_porcelain_v2_branch_and_paths() -> None: + output = ( + b"# branch.oid " + b"1" * 40 + b"\0" + b"# branch.head topic\0" + b"# branch.upstream origin/main\0" + b"# branch.ab +2 -3\0" + b"# stash 1\0" + b"1 M. N... 100644 100644 100644 " + + b"1" * 40 + + b" " + + b"2" * 40 + + b" staged name\0" + b"1 .M N... 100644 100644 100644 " + + b"1" * 40 + + b" " + + b"2" * 40 + + b" worktree name\0" + b"2 R. N... 100644 100644 100644 " + + b"1" * 40 + + b" " + + b"2" * 40 + + b" R075 renamed\0old name\0" + b"? untracked\0" + ) + + status = git_status.ParsePorcelainV2(output) + + assert status.current_branch == "topic" + assert status.upstream == "origin/main" + assert (status.ahead, status.behind, status.stash_count) == (2, 3, 1) + assert status.index_changes["staged name"].status == "M" + assert status.worktree_changes["worktree name"].status == "M" + renamed = status.index_changes["renamed"] + assert (renamed.src_path, renamed.level) == ("old name", "75") + assert status.untracked == ["untracked"] + + +def test_parse_porcelain_v2_unmerged_and_non_utf8_path() -> None: + path = b"bad-\xff-name" + output = ( + b"u UU N... 100644 100644 100644 100644 " + + b"1" * 40 + + b" " + + b"2" * 40 + + b" " + + b"3" * 40 + + b" " + + path + + b"\0" + ) + + status = git_status.ParsePorcelainV2(output) + decoded = os.fsdecode(path) + + assert status.index_changes[decoded].status == "U" + assert status.worktree_changes[decoded].status == "U" + assert os.fsencode(status.index_changes[decoded].path) == path + + +def test_untracked_only_respects_consider_untracked() -> None: + status = git_status.ParsePorcelainV2(b"? new file\0") + + assert status.is_dirty() + assert not status.is_dirty(consider_untracked=False) + + +def test_branch_headers_preserve_non_ascii_names() -> None: + branch = "tópico" + status = git_status.ParsePorcelainV2( + b"# branch.oid " + b"1" * 40 + b"\0" + b"# branch.head " + os.fsencode(branch) + b"\0" + ) + + assert status.current_branch == branch + + +def test_quick_ahead_behind_is_recorded_as_unknown() -> None: + status = git_status.ParsePorcelainV2(b"# branch.ab +? -?\0") + + assert (status.ahead, status.behind) == (0, 0) + assert not status.has_ahead_behind + + +def test_unknown_head_is_not_a_current_branch() -> None: + status = git_status.ParsePorcelainV2(b"# branch.head (unknown)\0") + + assert status.current_branch is None + + +def test_malformed_output_is_rejected() -> None: + with pytest.raises(git_status.StatusParseError): + git_status.ParsePorcelainV2(b"2 R. truncated\0") + + +def test_malformed_branch_ab_is_rejected() -> None: + with pytest.raises(git_status.StatusParseError): + git_status.ParsePorcelainV2(b"# branch.ab not-a-valid-ab\0") + + +def test_ignored_records_are_skipped() -> None: + status = git_status.ParsePorcelainV2(b"! ignored_file\0") + assert not status.is_dirty() + assert status.untracked == [] + + +def test_get_status_uses_versioned_machine_options( + monkeypatch: pytest.MonkeyPatch, +) -> None: + commands = [] + + class FakeGitCommand: + def __init__( + self, _project: Any, cmdv: List[str], **kwargs: Any + ) -> None: + commands.append((cmdv, kwargs)) + self.stdout = b"" + + def Wait(self) -> int: + return 0 + + monkeypatch.setattr(git_status, "GitCommand", FakeGitCommand) + monkeypatch.setattr(git_status, "git_require", lambda _version: True) + + git_status.GetStatus( + mock.sentinel.project, + mock.sentinel.gitdir, + untracked_files="no", + branch=True, + ahead_behind=True, + show_stash=True, + ) + + cmd, kwargs = commands[0] + assert cmd == [ + "status", + "--porcelain=v2", + "-z", + "--ignore-submodules=all", + "--untracked-files=no", + "--branch", + "--ahead-behind", + "--renames", + "--show-stash", + ] + assert kwargs["capture_stdout_bytes"] + + +def test_get_status_rejects_git_before_2_11( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(git_status, "git_require", lambda _version: False) + + with pytest.raises(git_status.UnsupportedStatusError): + git_status.GetStatus(mock.sentinel.project, mock.sentinel.gitdir) + + +def test_get_status_omits_stash_header_before_2_35( + monkeypatch: pytest.MonkeyPatch, +) -> None: + commands = [] + + class FakeGitCommand: + def __init__( + self, _project: Any, cmdv: List[str], **_kwargs: Any + ) -> None: + commands.append(cmdv) + self.stdout = b"" + + def Wait(self) -> int: + return 0 + + monkeypatch.setattr(git_status, "GitCommand", FakeGitCommand) + monkeypatch.setattr( + git_status, + "git_require", + lambda version: version <= (2, 34, 0), + ) + + git_status.GetStatus( + mock.sentinel.project, + mock.sentinel.gitdir, + show_stash=True, + ) + + assert "--show-stash" not in commands[0]