diff --git a/git_status.py b/git_status.py new file mode 100644 index 000000000..257031464 --- /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 000000000..72698db26 --- /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]