git_status.py - git-repo - Git at Google

 # 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
	# 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