git-repo/git_status.py

# 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 git_require((2, 17, 0)):
            cmd.append(
                "--ahead-behind" if ahead_behind else "--no-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