"""Special objects for :meth:`Diffable.diff`.

NULL_TREE = enum.auto()

"""Stand-in indicating you want to compare against the empty tree in diffs.

INDEX = enum.auto()

"""Stand-in indicating you want to diff against the index.

result = bytearray()

escapes = {

ord("a"): 7,

ord("b"): 8,

ord("f"): 12,

ord("n"): 10,

ord("r"): 13,

ord("t"): 9,

ord("v"): 11,

}

i = 0

while i < len(path):

if path[i] != ord("\\") or i + 1 == len(path):

result.append(path[i])

i += 1

continue

if path[i + 1] in b"0123" and i + 3 < len(path) and all(c in b"01234567" for c in path[i + 2 : i + 4]):

result.append(int(path[i + 1 : i + 4], 8))

i += 4

continue

escaped = path[i + 1]

if escaped in escapes or escaped in b'\\"':

result.append(escapes.get(escaped, escaped))

else:

result.extend(path[i : i + 2])

i += 2

if path == b"/dev/null":

return None

if path.startswith(b'"') and path.endswith(b'"'):

path = _unquote_path(path[1:-1])

if has_ab_prefix:

assert path.startswith(b"a/") or path.startswith(b"b/")

path = path[2:]

"""Common interface for all objects that can be diffed against another object of

__slots__ = ()

repo: "Repo"

"""Repository to operate on. Must be provided by subclass or sibling class."""

NULL_TREE = NULL_TREE

"""Stand-in indicating you want to compare against the empty tree in diffs.

INDEX = INDEX

"""Stand-in indicating you want to diff against the index.

Index = INDEX

"""Stand-in indicating you want to diff against the index

def _process_diff_args(

self,

args: List[Union[PathLike, "Diffable"]],

) -> List[Union[PathLike, "Diffable"]]:

"""

return args

def diff(

self,

other: Union[DiffConstants, "Tree", "Commit", str, None] = INDEX,

paths: Union[PathLike, List[PathLike], Tuple[PathLike, ...], None] = None,

create_patch: bool = False,

allow_unsafe_options: bool = False,

**kwargs: Any,

) -> "DiffIndex[Diff]":

"""Create diffs between two items being trees, trees and index or an index and

if not allow_unsafe_options:

Git.check_unsafe_options(

options=Git._option_candidates([other], kwargs),

unsafe_options=self.repo.unsafe_git_diff_options,

clusterable_short_options="46abceflmnpqrstuvwzBCDMNRW",

)

args: List[Union[PathLike, Diffable]] = []

args.append("--abbrev=40") # We need full shas.

args.append("--full-index") # Get full index paths, not only filenames.

# Remove default '-M' arg (check for renames) if user is overriding it.

if not any(x in kwargs for x in ("find_renames", "no_renames", "M")):

args.append("-M")

if create_patch:

args.append("-p")

args.append("--no-ext-diff")

else:

args.append("--raw")

args.append("-z")

# Ensure we never see colored output.

# Fixes: https://github.com/gitpython-developers/GitPython/issues/172

args.append("--no-color")

if paths is not None and not isinstance(paths, (tuple, list)):

paths = [paths]

diff_cmd = self.repo.git.diff

if other is INDEX:

args.insert(0, "--cached")

elif other is NULL_TREE:

args.insert(0, "-r") # Recursive diff-tree.

args.insert(0, "--root")

diff_cmd = self.repo.git.diff_tree

elif other is not None:

args.insert(0, "-r") # Recursive diff-tree.

args.insert(0, other)

diff_cmd = self.repo.git.diff_tree

args.insert(0, self)

# paths is a list or tuple here, or None.

if paths:

args.append("--")

args.extend(paths)

# END paths handling

kwargs["as_process"] = True

args = self._process_diff_args(args)

if create_patch:

self.repo.git(c="diff.mnemonicPrefix=false")

proc = diff_cmd(*args, **kwargs)

diff_method = Diff._index_from_patch_format if create_patch else Diff._index_from_raw_format

index = diff_method(self.repo, proc)

proc.wait()

R"""An index for diffs, allowing a list of :class:`Diff`\s to be queried by the diff

change_type: Sequence[Literal["A", "C", "D", "R", "M", "T"]] = ("A", "C", "D", "R", "M", "T") # noqa: F821

"""Change type invariant identifying possible ways a blob can have changed:

def iter_change_type(self, change_type: Lit_change_type) -> Iterator[T_Diff]:

"""

if change_type not in self.change_type:

raise ValueError("Invalid change type: %s" % change_type)

for diffidx in self:

if diffidx.change_type == change_type:

yield diffidx

elif change_type == "A" and diffidx.new_file:

yield diffidx

elif change_type == "D" and diffidx.deleted_file:

yield diffidx

elif change_type == "C" and diffidx.copied_file:

yield diffidx

elif change_type == "R" and diffidx.renamed_file:

yield diffidx

elif change_type == "M" and diffidx.a_blob and diffidx.b_blob and diffidx.a_blob != diffidx.b_blob:

yield diffidx

"""A Diff contains diff information between two Trees.

# Precompiled regex.

re_header = re.compile(

rb"""

""",

re.VERBOSE | re.MULTILINE,

)

# These can be used for comparisons.

NULL_HEX_SHA = "0" * 40

NULL_BIN_SHA = b"\0" * 20

__slots__ = (

"a_blob",

"b_blob",

"a_mode",

"b_mode",

"a_rawpath",

"b_rawpath",

"new_file",

"deleted_file",

"copied_file",

"raw_rename_from",

"raw_rename_to",

"diff",

"change_type",

"score",

)

def __init__(

self,

repo: "Repo",

a_rawpath: Optional[bytes],

b_rawpath: Optional[bytes],

a_blob_id: Union[str, bytes, None],

b_blob_id: Union[str, bytes, None],

a_mode: Union[bytes, str, None],

b_mode: Union[bytes, str, None],

new_file: bool,

deleted_file: bool,

copied_file: bool,

raw_rename_from: Optional[bytes],

raw_rename_to: Optional[bytes],

diff: Union[str, bytes, None],

change_type: Optional[Lit_change_type],

score: Optional[int],

) -> None:

assert a_rawpath is None or isinstance(a_rawpath, bytes)

assert b_rawpath is None or isinstance(b_rawpath, bytes)

self.a_rawpath = a_rawpath

self.b_rawpath = b_rawpath

self.a_mode = mode_str_to_int(a_mode) if a_mode else None

self.b_mode = mode_str_to_int(b_mode) if b_mode else None

# Determine whether this diff references a submodule. If it does then

# we need to overwrite "repo" to the corresponding submodule's repo instead.

if repo and a_rawpath:

for submodule in repo.submodules:

if submodule.path == a_rawpath.decode(defenc, "replace"):

if submodule.module_exists():

repo = submodule.module()

break

# Gitlinks reference commits; generic index objects preserve their path and mode.

self.a_blob: Union["IndexObject", None]

if a_blob_id is None or a_blob_id == self.NULL_HEX_SHA:

self.a_blob = None

else:

self.a_blob = (IndexObject if self.a_mode == 0o160000 else Blob)(

repo, hex_to_bin(a_blob_id), mode=self.a_mode, path=self.a_path

)

self.b_blob: Union["IndexObject", None]

if b_blob_id is None or b_blob_id == self.NULL_HEX_SHA:

self.b_blob = None

else:

self.b_blob = (IndexObject if self.b_mode == 0o160000 else Blob)(

repo, hex_to_bin(b_blob_id), mode=self.b_mode, path=self.b_path

)

self.new_file: bool = new_file

self.deleted_file: bool = deleted_file

self.copied_file: bool = copied_file

# Be clear and use None instead of empty strings.

assert raw_rename_from is None or isinstance(raw_rename_from, bytes)

assert raw_rename_to is None or isinstance(raw_rename_to, bytes)

self.raw_rename_from = raw_rename_from or None

self.raw_rename_to = raw_rename_to or None

self.diff = diff

self.change_type: Union[Lit_change_type, None] = change_type

self.score = score

def __eq__(self, other: object) -> bool:

for name in self.__slots__:

if getattr(self, name) != getattr(other, name):

return False

# END for each name

return True

def __ne__(self, other: object) -> bool:

return not (self == other)

def __hash__(self) -> int:

return hash(tuple(getattr(self, n) for n in self.__slots__))

def __str__(self) -> str:

h = "%s"

if self.a_blob:

h %= self.a_blob.path

elif self.b_blob:

h %= self.b_blob.path

msg = ""

line = None

line_length = 0

for b, n in zip((self.a_blob, self.b_blob), ("lhs", "rhs")):

if b:

line = "\n%s: %o | %s" % (n, b.mode, b.hexsha)

else:

line = "\n%s: None" % n

# END if blob is not None

line_length = max(len(line), line_length)

msg += line

# END for each blob

# Add headline.

h += "\n" + "=" * line_length

if self.deleted_file:

msg += "\nfile deleted in rhs"

if self.new_file:

msg += "\nfile added in rhs"

if self.copied_file:

msg += "\nfile %r copied from %r" % (self.b_path, self.a_path)

if self.rename_from:

msg += "\nfile renamed from %r" % self.rename_from

if self.rename_to:

msg += "\nfile renamed to %r" % self.rename_to

if self.diff:

msg += "\n---"

try:

msg += self.diff.decode(defenc) if isinstance(self.diff, bytes) else self.diff

except UnicodeDecodeError:

msg += "OMITTED BINARY DATA"

# END handle encoding

msg += "\n---"

# END diff info

return h + msg

@property

def a_path(self) -> Optional[str]:

return self.a_rawpath.decode(defenc, "replace") if self.a_rawpath else None

@property

def b_path(self) -> Optional[str]:

return self.b_rawpath.decode(defenc, "replace") if self.b_rawpath else None

@property

def rename_from(self) -> Optional[str]:

return self.raw_rename_from.decode(defenc, "replace") if self.raw_rename_from else None

@property

def rename_to(self) -> Optional[str]:

return self.raw_rename_to.decode(defenc, "replace") if self.raw_rename_to else None

@property

def renamed(self) -> bool:

"""Deprecated, use :attr:`renamed_file` instead.

warnings.warn(

"Diff.renamed is deprecated, use Diff.renamed_file instead",

DeprecationWarning,

stacklevel=2,

)

return self.renamed_file

@property

def renamed_file(self) -> bool:

""":return: ``True`` if the blob of our diff has been renamed"""

return self.rename_from != self.rename_to

@classmethod

def _pick_best_path(cls, path_match: bytes, rename_match: bytes, path_fallback_match: bytes) -> Optional[bytes]:

if path_match:

return decode_path(path_match)

if rename_match:

return decode_path(rename_match, has_ab_prefix=False)

if path_fallback_match:

return decode_path(path_fallback_match)

return None

@classmethod

def _index_from_patch_format(cls, repo: "Repo", proc: Union["Popen", "Git.AutoInterrupt"]) -> DiffIndex["Diff"]:

"""Create a new :class:`DiffIndex` from the given process output which must be

# FIXME: Here SLURPING raw, need to re-phrase header-regexes linewise.

text_list: List[bytes] = []

stderr_list: List[bytes] = []

def finalize_process_with_stderr(proc: Union["Popen", "Git.AutoInterrupt"]) -> None:

finalize_process(proc, stderr=b"".join(stderr_list))

handle_process_output(

proc, text_list.append, stderr_list.append, finalize_process_with_stderr, decode_streams=False

)

# For now, we have to bake the stream.

text = b"".join(text_list)

index: "DiffIndex" = DiffIndex()

previous_header: Union[Match[bytes], None] = None

header: Union[Match[bytes], None] = None

a_path, b_path = None, None # For mypy.

a_mode, b_mode = None, None # For mypy.

for _header in cls.re_header.finditer(text):

(

a_path_fallback,

b_path_fallback,

old_mode,

new_mode,

rename_from,

rename_to,

new_file_mode,

deleted_file_mode,

copied_file_name,

a_blob_id,

b_blob_id,

b_mode,

a_path,

b_path,

) = _header.groups()

new_file, deleted_file, copied_file = (

bool(new_file_mode),

bool(deleted_file_mode),

bool(copied_file_name),

)

a_path = cls._pick_best_path(a_path, rename_from, a_path_fallback)

b_path = cls._pick_best_path(b_path, rename_to, b_path_fallback)

# Our only means to find the actual text is to see what has not been matched

# by our regex, and then retro-actively assign it to our index.

if previous_header is not None:

index[-1].diff = text[previous_header.end() : _header.start()]

# END assign actual diff

# Make sure the mode is set if the path is set. Otherwise the resulting blob

# is invalid. We just use the one mode we should have parsed.

a_mode = old_mode or deleted_file_mode or (a_path and (b_mode or new_mode or new_file_mode))

b_mode = b_mode or new_mode or new_file_mode or (b_path and a_mode)

index.append(

Diff(

repo,

a_path,

b_path,

a_blob_id and a_blob_id.decode(defenc),

b_blob_id and b_blob_id.decode(defenc),

a_mode and a_mode.decode(defenc),

b_mode and b_mode.decode(defenc),

new_file,

deleted_file,

copied_file,

rename_from,

rename_to,

None,

None,

None,

)

)

previous_header = _header

header = _header

# END for each header we parse

if index and header:

index[-1].diff = text[header.end() :]

# END assign last diff

return index

@staticmethod

def _handle_diff_line(lines_bytes: bytes, repo: "Repo", index: DiffIndex["Diff"]) -> None:

lines = lines_bytes.decode(defenc)

# Discard everything before the first colon, and the colon itself.

_, _, lines = lines.partition(":")

for line in lines.split("\x00:"):

if not line:

# The line data is empty, skip.

continue

meta, _, path = line.partition("\x00")

path = path.rstrip("\x00")

a_blob_id: Optional[str]

b_blob_id: Optional[str]

old_mode, new_mode, a_blob_id, b_blob_id, _change_type = meta.split(None, 4)

# Change type can be R100

# R: status letter

# 100: score (in case of copy and rename)

change_type: Lit_change_type = cast(Lit_change_type, _change_type[0])

score_str = "".join(_change_type[1:])

score = int(score_str) if score_str.isdigit() else None

path = path.strip("\n")

a_path = path.encode(defenc)

b_path = path.encode(defenc)

deleted_file = False

new_file = False

copied_file = False

rename_from = None

rename_to = None

# NOTE: We cannot conclude from the existence of a blob to change type,

# as diffs with the working do not have blobs yet.

if change_type == "D":

b_blob_id = None # Optional[str]

deleted_file = True

elif change_type == "A":

a_blob_id = None

new_file = True

elif change_type == "C":

copied_file = True

a_path_str, b_path_str = path.split("\x00", 1)

a_path = a_path_str.encode(defenc)

b_path = b_path_str.encode(defenc)

elif change_type == "R":

a_path_str, b_path_str = path.split("\x00", 1)

a_path = a_path_str.encode(defenc)

b_path = b_path_str.encode(defenc)

rename_from, rename_to = a_path, b_path

elif change_type == "T":

# Nothing to do.

pass

# END add/remove handling

diff = Diff(

repo,

a_path,

b_path,

a_blob_id,

b_blob_id,

old_mode,

new_mode,

new_file,

deleted_file,

copied_file,

rename_from,

rename_to,

"",

change_type,

score,

)

index.append(diff)

@classmethod

def _index_from_raw_format(cls, repo: "Repo", proc: "Popen") -> "DiffIndex[Diff]":

"""Create a new :class:`DiffIndex` from the given process output which must be

# handles

# :100644 100644 687099101... 37c5e30c8... M .gitignore

index: "DiffIndex" = DiffIndex()

stderr_list: List[bytes] = []

def finalize_process_with_stderr(proc: Union["Popen", "Git.AutoInterrupt"]) -> None:

finalize_process(proc, stderr=b"".join(stderr_list))

handle_process_output(

proc,

lambda byt: cls._handle_diff_line(byt, repo, index),

stderr_list.append,

finalize_process_with_stderr,

decode_streams=False,

)