Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/git/index/fun.py: 45%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# This module is part of GitPython and is released under the
2# 3-Clause BSD License: https://opensource.org/license/bsd-3-clause/
4"""Standalone functions to accompany the index implementation and make it more
5versatile."""
7__all__ = [
8 "write_cache",
9 "read_cache",
10 "write_tree_from_cache",
11 "entry_key",
12 "stat_mode_to_index_mode",
13 "S_IFGITLINK",
14 "run_commit_hook",
15 "hook_path",
16]
18from io import BytesIO
19import os
20import os.path as osp
21from pathlib import Path
22from stat import S_IFDIR, S_IFLNK, S_IFMT, S_IFREG, S_ISDIR, S_ISLNK, S_IXUSR
23import subprocess
24import sys
26from gitdb.base import IStream
27from gitdb.typ import str_tree_type
29from git.cmd import Git, handle_process_output, safer_popen
30from git.compat import defenc, force_bytes, force_text, safe_decode
31from git.exc import HookExecutionError, UnmergedEntriesError
32from git.objects.fun import (
33 traverse_tree_recursive,
34 traverse_trees_recursive,
35 tree_to_stream,
36)
37from git.util import IndexFileSHA1Writer, finalize_process
39from .typ import CE_EXTENDED, BaseIndexEntry, IndexEntry, CE_NAMEMASK, CE_STAGESHIFT
40from .util import pack, unpack
42# typing -----------------------------------------------------------------------------
44from typing import Dict, IO, List, Sequence, TYPE_CHECKING, Tuple, Type, Union, cast
46from git.types import PathLike
48if TYPE_CHECKING:
49 from git.db import GitCmdObjectDB
50 from git.objects.tree import TreeCacheTup
52 from .base import IndexFile
54# ------------------------------------------------------------------------------------
56S_IFGITLINK = S_IFLNK | S_IFDIR
57"""Flags for a submodule."""
59CE_NAMEMASK_INV = ~CE_NAMEMASK
62def hook_path(name: str, git_dir: PathLike) -> str:
63 """:return: path to the given named hook in the given git repository directory"""
64 return osp.join(git_dir, "hooks", name)
67def _commit_hook_path(name: str, index: "IndexFile") -> str:
68 """:return: path to the named commit hook, respecting Git's core.hooksPath."""
69 with index.repo.config_reader() as config:
70 hooks_dir = config.get("core", "hooksPath", fallback="")
72 if not hooks_dir:
73 return hook_path(name, index.repo.git_dir)
75 return osp.abspath(osp.join(index.repo.working_dir, osp.expanduser(hooks_dir), name))
78def _has_file_extension(path: str) -> str:
79 return osp.splitext(path)[1]
82def _is_in_windows_system_root(path: str) -> bool:
83 """Return whether ``path`` is inside the Windows installation directory."""
84 system_root = os.environ.get("SystemRoot")
85 if not system_root:
86 return False
88 system_root = osp.normcase(osp.realpath(system_root))
89 path = osp.normcase(osp.realpath(path))
90 try:
91 return osp.commonpath((system_root, path)) == system_root
92 except ValueError:
93 # Paths on different drives have no common path on Windows.
94 return False
97def _which_from_path(command: str) -> Union[str, None]:
98 """Resolve ``command`` from PATH, excluding the Windows installation."""
99 for directory in os.get_exec_path():
100 # Unlike POSIX, Windows does not define an empty PATH entry as the current
101 # directory. Skip it rather than letting abspath() turn it into one.
102 if not directory:
103 continue
104 directory = osp.abspath(directory)
105 candidate = osp.join(directory, command)
106 # SystemRoot contains the WSL launcher stubs. They are valid executables but
107 # not suitable for running a Windows Git hook: the hook path and environment
108 # were prepared for Git for Windows, and WSL may have no distribution at all.
109 if _is_in_windows_system_root(candidate):
110 continue
111 if osp.isfile(candidate) and os.access(candidate, os.X_OK):
112 return candidate
113 return None
116_GIT_FOR_WINDOWS_PREFIXES = ("mingw64", "mingw32", "clangarm64", "clang64", "clang32", "ucrt64")
119def _git_for_windows_root() -> Union[str, None]:
120 """Infer a standard Git for Windows root from GitPython's selected executable."""
121 git_executable = os.fspath(Git.GIT_PYTHON_GIT_EXECUTABLE or Git.git_exec_name)
122 if osp.dirname(git_executable):
123 # CreateProcess resolves a relative executable path containing a directory
124 # from the parent process cwd, even when Popen supplies a different child cwd.
125 git_executable = osp.abspath(git_executable)
126 else:
127 # GitPython deliberately retains a bare executable name so later PATH changes
128 # affect Git commands. Resolve it with the same PATH snapshot used for Bash.
129 names = (git_executable,) if _has_file_extension(git_executable) else (git_executable, f"{git_executable}.exe")
130 for name in names:
131 resolved = _which_from_path(name)
132 if resolved is not None:
133 git_executable = resolved
134 break
135 else:
136 git_executable = ""
137 if not git_executable:
138 return None
139 if osp.basename(git_executable).lower() not in ("git", "git.exe"):
140 return None
142 executable_dir = osp.dirname(git_executable)
143 directory_name = osp.basename(executable_dir).lower()
144 if directory_name == "cmd":
145 # The normal system-wide PATH entry is <git-root>/cmd.
146 return osp.dirname(executable_dir)
147 if directory_name == "bin":
148 prefix = osp.dirname(executable_dir)
149 if osp.basename(prefix).lower() in _GIT_FOR_WINDOWS_PREFIXES:
150 # Git Bash commonly exposes <git-root>/<platform>/bin/git.exe.
151 return osp.dirname(prefix)
152 if osp.basename(prefix).lower() != "usr":
153 # An explicitly configured Git may be the root-level bin/git.exe. Do
154 # not make the same inference from usr/bin: unlike the recognized
155 # platform prefixes, "usr" has no reliably bounded parent layout.
156 return prefix
157 return None
160def _git_for_windows_bash() -> Union[str, None]:
161 """Return Bash from the Git for Windows installation selected by GitPython."""
162 git_root = _git_for_windows_root()
163 if git_root is None:
164 return None
166 # Match gix-path's precedence: prefer the lightweight bin shim, then the
167 # underlying usr/bin executable. Both belong to the same installation as Git.
168 for relative_path in ("bin/bash.exe", "usr/bin/bash.exe"):
169 candidate = osp.join(git_root, *relative_path.split("/"))
170 if osp.isfile(candidate) and os.access(candidate, os.X_OK):
171 return candidate
172 return None
175def run_commit_hook(name: str, index: "IndexFile", *args: str) -> None:
176 """Run the commit hook of the given name. Silently ignore hooks that do not exist.
178 :param name:
179 Name of hook, like ``pre-commit``.
181 :param index:
182 :class:`~git.index.base.IndexFile` instance.
184 :param args:
185 Arguments passed to hook file.
187 :raise git.exc.HookExecutionError:
188 """
189 hp = _commit_hook_path(name, index)
190 if not os.access(hp, os.X_OK):
191 return
193 env = os.environ.copy()
194 env["GIT_INDEX_FILE"] = safe_decode(os.fspath(index.path))
195 env["GIT_EDITOR"] = ":"
196 cmd = [hp]
197 try:
198 if sys.platform == "win32" and not _has_file_extension(hp):
199 # Windows only uses extensions to determine how to open files
200 # (doesn't understand shebangs). Try using bash to run the hook.
201 try:
202 bash_hp = osp.relpath(hp, index.repo.working_dir)
203 except ValueError:
204 # Different drives have no relative path on Windows. Git Bash accepts
205 # an absolute path in this form, although a relative path is preferable
206 # because it also works with the Windows Subsystem for Linux wrapper.
207 bash_hp = hp
208 # Prefer Bash associated with GitPython's selected Git installation. If
209 # that layout is not recognized, use an explicitly configured non-system
210 # PATH entry. Preserve the bare fallback for installations that previously
211 # relied on WSL or another CreateProcess-resolved Bash.
212 bash_executable = _git_for_windows_bash() or _which_from_path("bash.exe") or "bash.exe"
213 cmd = [bash_executable, Path(bash_hp).as_posix()]
215 process = safer_popen(
216 cmd + list(args),
217 env=env,
218 stdout=subprocess.PIPE,
219 stderr=subprocess.PIPE,
220 cwd=index.repo.working_dir,
221 )
222 except Exception as ex:
223 raise HookExecutionError(hp, ex) from ex
224 else:
225 stdout_list: List[str] = []
226 stderr_list: List[str] = []
227 handle_process_output(process, stdout_list.append, stderr_list.append, finalize_process)
228 stdout = "".join(stdout_list)
229 stderr = "".join(stderr_list)
230 if process.returncode != 0:
231 stdout = force_text(stdout, defenc)
232 stderr = force_text(stderr, defenc)
233 raise HookExecutionError(hp, process.returncode, stderr, stdout)
234 # END handle return code
237def stat_mode_to_index_mode(mode: int) -> int:
238 """Convert the given mode from a stat call to the corresponding index mode and
239 return it."""
240 if S_ISLNK(mode): # symlinks
241 return S_IFLNK
242 if S_ISDIR(mode) or S_IFMT(mode) == S_IFGITLINK: # submodules
243 return S_IFGITLINK
244 return S_IFREG | (mode & S_IXUSR and 0o755 or 0o644) # blobs with or without executable bit
247def write_cache(
248 entries: Sequence[Union[BaseIndexEntry, "IndexEntry"]],
249 stream: IO[bytes],
250 extension_data: Union[None, bytes] = None,
251 ShaStreamCls: Type[IndexFileSHA1Writer] = IndexFileSHA1Writer,
252) -> None:
253 """Write the cache represented by entries to a stream.
255 :param entries:
256 **Sorted** list of entries.
258 :param stream:
259 Stream to wrap into the AdapterStreamCls - it is used for final output.
261 :param ShaStreamCls:
262 Type to use when writing to the stream. It produces a sha while writing to it,
263 before the data is passed on to the wrapped stream.
265 :param extension_data:
266 Any kind of data to write as a trailer, it must begin a 4 byte identifier,
267 followed by its size (4 bytes).
268 """
269 # Wrap the stream into a compatible writer.
270 stream_sha = ShaStreamCls(stream)
272 tell = stream_sha.tell
273 write = stream_sha.write
275 # Header
276 version = 3 if any(entry.extended_flags for entry in entries) else 2
277 write(b"DIRC")
278 write(pack(">LL", version, len(entries)))
280 # Body
281 for entry in entries:
282 beginoffset = tell()
283 write(entry.ctime_bytes) # ctime
284 write(entry.mtime_bytes) # mtime
285 path_str = str(entry.path)
286 path: bytes = force_bytes(path_str, encoding=defenc)
287 plen = len(path) & CE_NAMEMASK # Path length
288 assert plen == len(path), "Path %s too long to fit into index" % entry.path
289 flags = plen | (entry.flags & CE_NAMEMASK_INV) # Clear possible previous values.
290 if entry.extended_flags:
291 flags |= CE_EXTENDED
292 write(
293 pack(
294 ">LLLLLL20sH",
295 entry.dev,
296 entry.inode,
297 entry.mode,
298 entry.uid,
299 entry.gid,
300 entry.size,
301 entry.binsha,
302 flags,
303 )
304 )
305 if entry.extended_flags:
306 write(pack(">H", entry.extended_flags))
307 write(path)
308 real_size = (tell() - beginoffset + 8) & ~7
309 write(b"\0" * ((beginoffset + real_size) - tell()))
310 # END for each entry
312 # Write previously cached extensions data.
313 if extension_data is not None:
314 stream_sha.write(extension_data)
316 # Write the sha over the content.
317 stream_sha.write_sha()
320def read_header(stream: IO[bytes]) -> Tuple[int, int]:
321 """Return tuple(version_long, num_entries) from the given stream."""
322 type_id = stream.read(4)
323 if type_id != b"DIRC":
324 raise AssertionError("Invalid index file header: %r" % type_id)
325 unpacked = cast(Tuple[int, int], unpack(">LL", stream.read(4 * 2)))
326 version, num_entries = unpacked
328 assert version in (1, 2, 3), "Unsupported git index version %i, only 1, 2, and 3 are supported" % version
329 return version, num_entries
332def entry_key(*entry: Union[BaseIndexEntry, PathLike, int]) -> Tuple[PathLike, int]:
333 """
334 :return:
335 Key suitable to be used for the
336 :attr:`index.entries <git.index.base.IndexFile.entries>` dictionary.
338 :param entry:
339 One instance of type BaseIndexEntry or the path and the stage.
340 """
342 # def is_entry_key_tup(entry_key: Tuple) -> TypeGuard[Tuple[PathLike, int]]:
343 # return isinstance(entry_key, tuple) and len(entry_key) == 2
345 if len(entry) == 1:
346 entry_first = entry[0]
347 assert isinstance(entry_first, BaseIndexEntry)
348 return (entry_first.path, entry_first.stage)
349 else:
350 # assert is_entry_key_tup(entry)
351 entry = cast(Tuple[PathLike, int], entry)
352 return entry
353 # END handle entry
356def read_cache(
357 stream: IO[bytes],
358) -> Tuple[int, Dict[Tuple[PathLike, int], "IndexEntry"], bytes, bytes]:
359 """Read a cache file from the given stream.
361 :return:
362 tuple(version, entries_dict, extension_data, content_sha)
364 * *version* is the integer version number.
365 * *entries_dict* is a dictionary which maps IndexEntry instances to a path at a
366 stage.
367 * *extension_data* is ``""`` or 4 bytes of type + 4 bytes of size + size bytes.
368 * *content_sha* is a 20 byte sha on all cache file contents.
369 """
370 version, num_entries = read_header(stream)
371 count = 0
372 entries: Dict[Tuple[PathLike, int], "IndexEntry"] = {}
374 read = stream.read
375 tell = stream.tell
376 while count < num_entries:
377 beginoffset = tell()
378 ctime = unpack(">8s", read(8))[0]
379 mtime = unpack(">8s", read(8))[0]
380 (dev, ino, mode, uid, gid, size, sha, flags) = unpack(">LLLLLL20sH", read(20 + 4 * 6 + 2))
381 extended_flags = 0
382 if flags & CE_EXTENDED:
383 extended_flags = unpack(">H", read(2))[0]
384 path_size = flags & CE_NAMEMASK
385 path = read(path_size).decode(defenc)
387 real_size = (tell() - beginoffset + 8) & ~7
388 read((beginoffset + real_size) - tell())
389 entry = IndexEntry((mode, sha, flags, path, ctime, mtime, dev, ino, uid, gid, size, extended_flags))
390 # entry_key would be the method to use, but we save the effort.
391 entries[(path, entry.stage)] = entry
392 count += 1
393 # END for each entry
395 # The footer contains extension data and a sha on the content so far.
396 # Keep the extension footer,and verify we have a sha in the end.
397 # Extension data format is:
398 # 4 bytes ID
399 # 4 bytes length of chunk
400 # Repeated 0 - N times
401 extension_data = stream.read(~0)
402 assert len(extension_data) > 19, (
403 "Index Footer was not at least a sha on content as it was only %i bytes in size" % len(extension_data)
404 )
406 content_sha = extension_data[-20:]
408 # Truncate the sha in the end as we will dynamically create it anyway.
409 extension_data = extension_data[:-20]
411 return (version, entries, extension_data, content_sha)
414def write_tree_from_cache(
415 entries: List[IndexEntry], odb: "GitCmdObjectDB", sl: slice, si: int = 0
416) -> Tuple[bytes, List["TreeCacheTup"]]:
417 R"""Create a tree from the given sorted list of entries and put the respective
418 trees into the given object database.
420 :param entries:
421 **Sorted** list of :class:`~git.index.typ.IndexEntry`\s.
423 :param odb:
424 Object database to store the trees in.
426 :param si:
427 Start index at which we should start creating subtrees.
429 :param sl:
430 Slice indicating the range we should process on the entries list.
432 :return:
433 tuple(binsha, list(tree_entry, ...))
435 A tuple of a sha and a list of tree entries being a tuple of hexsha, mode, name.
436 """
437 tree_items: List["TreeCacheTup"] = []
439 ci = sl.start
440 end = sl.stop
441 while ci < end:
442 entry = entries[ci]
443 if entry.stage != 0:
444 raise UnmergedEntriesError(entry)
445 # END abort on unmerged
446 ci += 1
447 rbound = entry.path.find("/", si)
448 if rbound == -1:
449 # It's not a tree.
450 tree_items.append((entry.binsha, entry.mode, entry.path[si:]))
451 else:
452 # Find common base range.
453 base = entry.path[si:rbound]
454 xi = ci
455 while xi < end:
456 oentry = entries[xi]
457 orbound = oentry.path.find("/", si)
458 if orbound == -1 or oentry.path[si:orbound] != base:
459 break
460 # END abort on base mismatch
461 xi += 1
462 # END find common base
464 # Enter recursion.
465 # ci - 1 as we want to count our current item as well.
466 sha, _tree_entry_list = write_tree_from_cache(entries, odb, slice(ci - 1, xi), rbound + 1)
467 tree_items.append((sha, S_IFDIR, base))
469 # Skip ahead.
470 ci = xi
471 # END handle bounds
472 # END for each entry
474 # Finally create the tree.
475 sio = BytesIO()
476 tree_to_stream(tree_items, sio.write) # Writes to stream as bytes, but doesn't change tree_items.
477 sio.seek(0)
479 istream = odb.store(IStream(str_tree_type, len(sio.getvalue()), sio))
480 return (istream.binsha, tree_items)
483def _tree_entry_to_baseindexentry(tree_entry: "TreeCacheTup", stage: int) -> BaseIndexEntry:
484 return BaseIndexEntry((tree_entry[1], tree_entry[0], stage << CE_STAGESHIFT, tree_entry[2]))
487def aggressive_tree_merge(odb: "GitCmdObjectDB", tree_shas: Sequence[bytes]) -> List[BaseIndexEntry]:
488 R"""
489 :return:
490 List of :class:`~git.index.typ.BaseIndexEntry`\s representing the aggressive
491 merge of the given trees. All valid entries are on stage 0, whereas the
492 conflicting ones are left on stage 1, 2 or 3, whereas stage 1 corresponds to the
493 common ancestor tree, 2 to our tree and 3 to 'their' tree.
495 :param tree_shas:
496 1, 2 or 3 trees as identified by their binary 20 byte shas. If 1 or two, the
497 entries will effectively correspond to the last given tree. If 3 are given, a 3
498 way merge is performed.
499 """
500 out: List[BaseIndexEntry] = []
502 # One and two way is the same for us, as we don't have to handle an existing
503 # index, instrea
504 if len(tree_shas) in (1, 2):
505 for entry in traverse_tree_recursive(odb, tree_shas[-1], ""):
506 out.append(_tree_entry_to_baseindexentry(entry, 0))
507 # END for each entry
508 return out
509 # END handle single tree
511 if len(tree_shas) > 3:
512 raise ValueError("Cannot handle %i trees at once" % len(tree_shas))
514 # Three trees.
515 for base, ours, theirs in traverse_trees_recursive(odb, tree_shas, ""):
516 if base is not None:
517 # Base version exists.
518 if ours is not None:
519 # Ours exists.
520 if theirs is not None:
521 # It exists in all branches. Ff it was changed in both
522 # its a conflict. Otherwise, we take the changed version.
523 # This should be the most common branch, so it comes first.
524 if (base[0] != ours[0] and base[0] != theirs[0] and ours[0] != theirs[0]) or (
525 base[1] != ours[1] and base[1] != theirs[1] and ours[1] != theirs[1]
526 ):
527 # Changed by both.
528 out.append(_tree_entry_to_baseindexentry(base, 1))
529 out.append(_tree_entry_to_baseindexentry(ours, 2))
530 out.append(_tree_entry_to_baseindexentry(theirs, 3))
531 elif base[0] != ours[0] or base[1] != ours[1]:
532 # Only we changed it.
533 out.append(_tree_entry_to_baseindexentry(ours, 0))
534 else:
535 # Either nobody changed it, or they did. In either
536 # case, use theirs.
537 out.append(_tree_entry_to_baseindexentry(theirs, 0))
538 # END handle modification
539 else:
540 if ours[0] != base[0] or ours[1] != base[1]:
541 # They deleted it, we changed it, conflict.
542 out.append(_tree_entry_to_baseindexentry(base, 1))
543 out.append(_tree_entry_to_baseindexentry(ours, 2))
544 # else:
545 # # We didn't change it, ignore.
546 # pass
547 # END handle our change
548 # END handle theirs
549 else:
550 if theirs is None:
551 # Deleted in both, its fine - it's out.
552 pass
553 else:
554 if theirs[0] != base[0] or theirs[1] != base[1]:
555 # Deleted in ours, changed theirs, conflict.
556 out.append(_tree_entry_to_baseindexentry(base, 1))
557 out.append(_tree_entry_to_baseindexentry(theirs, 3))
558 # END theirs changed
559 # else:
560 # # Theirs didn't change.
561 # pass
562 # END handle theirs
563 # END handle ours
564 else:
565 # All three can't be None.
566 if ours is None:
567 # Added in their branch.
568 assert theirs is not None
569 out.append(_tree_entry_to_baseindexentry(theirs, 0))
570 elif theirs is None:
571 # Added in our branch.
572 out.append(_tree_entry_to_baseindexentry(ours, 0))
573 else:
574 # Both have it, except for the base, see whether it changed.
575 if ours[0] != theirs[0] or ours[1] != theirs[1]:
576 out.append(_tree_entry_to_baseindexentry(ours, 2))
577 out.append(_tree_entry_to_baseindexentry(theirs, 3))
578 else:
579 # It was added the same in both.
580 out.append(_tree_entry_to_baseindexentry(ours, 0))
581 # END handle two items
582 # END handle heads
583 # END handle base exists
584 # END for each entries tuple
586 return out