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, _validate_repo_path
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 gitdb.db.base import ObjectDBR, ObjectDBW
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 _validate_repo_path(entry.path)
283 beginoffset = tell()
284 write(entry.ctime_bytes) # ctime
285 write(entry.mtime_bytes) # mtime
286 path_str = os.fspath(entry.path)
287 path: bytes = force_bytes(path_str, encoding=defenc)
288 plen = min(len(path), CE_NAMEMASK) # Longer names use a sentinel.
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 if version not in (1, 2, 3):
329 raise AssertionError("Unsupported git index version %i, only 1, 2, and 3 are supported" % version)
330 return version, num_entries
333def entry_key(*entry: Union[BaseIndexEntry, PathLike, int]) -> Tuple[PathLike, int]:
334 """
335 :return:
336 Key suitable to be used for the
337 :attr:`index.entries <git.index.base.IndexFile.entries>` dictionary.
339 :param entry:
340 One instance of type BaseIndexEntry or the path and the stage.
341 """
343 # def is_entry_key_tup(entry_key: Tuple) -> TypeGuard[Tuple[PathLike, int]]:
344 # return isinstance(entry_key, tuple) and len(entry_key) == 2
346 if len(entry) == 1:
347 entry_first = entry[0]
348 assert isinstance(entry_first, BaseIndexEntry)
349 return (entry_first.path, entry_first.stage)
350 else:
351 # assert is_entry_key_tup(entry)
352 entry = cast(Tuple[PathLike, int], entry)
353 return entry
354 # END handle entry
357def read_cache(
358 stream: IO[bytes],
359) -> Tuple[int, Dict[Tuple[PathLike, int], "IndexEntry"], bytes, bytes]:
360 """Read a cache file from the given stream.
362 :return:
363 tuple(version, entries_dict, extension_data, content_sha)
365 * *version* is the integer version number.
366 * *entries_dict* is a dictionary which maps IndexEntry instances to a path at a
367 stage.
368 * *extension_data* is ``""`` or 4 bytes of type + 4 bytes of size + size bytes.
369 * *content_sha* is a 20 byte sha on all cache file contents.
370 """
371 version, num_entries = read_header(stream)
372 count = 0
373 entries: Dict[Tuple[PathLike, int], "IndexEntry"] = {}
375 read = stream.read
376 tell = stream.tell
377 while count < num_entries:
378 beginoffset = tell()
379 ctime = unpack(">8s", read(8))[0]
380 mtime = unpack(">8s", read(8))[0]
381 (dev, ino, mode, uid, gid, size, sha, flags) = unpack(">LLLLLL20sH", read(20 + 4 * 6 + 2))
382 extended_flags = 0
383 if flags & CE_EXTENDED:
384 extended_flags = unpack(">H", read(2))[0]
385 path_size = flags & CE_NAMEMASK
386 path_bytes = bytearray(read(path_size))
387 if len(path_bytes) != path_size:
388 raise ValueError("Truncated index entry path")
389 terminator = read(1)
390 if path_size == CE_NAMEMASK:
391 while terminator and terminator != b"\0":
392 path_bytes.extend(terminator)
393 terminator = read(1)
394 if terminator != b"\0":
395 raise ValueError("Unterminated index entry path")
396 path = path_bytes.decode(defenc)
397 _validate_repo_path(path)
399 real_size = (tell() - beginoffset + 7) & ~7
400 padding_size = beginoffset + real_size - tell()
401 if read(padding_size) != b"\0" * padding_size:
402 raise ValueError("Invalid index entry padding")
403 entry = IndexEntry((mode, sha, flags, path, ctime, mtime, dev, ino, uid, gid, size, extended_flags))
404 # entry_key would be the method to use, but we save the effort.
405 entries[(path, entry.stage)] = entry
406 count += 1
407 # END for each entry
409 # The footer contains extension data and a sha on the content so far.
410 # Keep the extension footer,and verify we have a sha in the end.
411 # Extension data format is:
412 # 4 bytes ID
413 # 4 bytes length of chunk
414 # Repeated 0 - N times
415 extension_data = stream.read(~0)
416 assert len(extension_data) > 19, (
417 "Index Footer was not at least a sha on content as it was only %i bytes in size" % len(extension_data)
418 )
420 content_sha = extension_data[-20:]
422 # Truncate the sha in the end as we will dynamically create it anyway.
423 extension_data = extension_data[:-20]
425 offset = 0
426 while offset < len(extension_data):
427 header = extension_data[offset : offset + 8]
428 if len(header) != 8:
429 raise ValueError("Truncated index extension header")
430 signature, size = unpack(">4sL", header)
431 if not b"A" <= signature[:1] <= b"Z":
432 raise ValueError("Unsupported mandatory index extension %r" % signature)
433 offset += 8 + size
434 if offset > len(extension_data):
435 raise ValueError("Truncated index extension %r" % signature)
437 return (version, entries, extension_data, content_sha)
440def write_tree_from_cache(
441 entries: List[IndexEntry], odb: "ObjectDBW", sl: slice, si: int = 0
442) -> Tuple[bytes, List["TreeCacheTup"]]:
443 R"""Create a tree from the given sorted list of entries and put the respective
444 trees into the given object database.
446 :param entries:
447 **Sorted** list of :class:`~git.index.typ.IndexEntry`\s.
449 :param odb:
450 Object database to store the trees in.
452 :param si:
453 Start index at which we should start creating subtrees.
455 :param sl:
456 Slice indicating the range we should process on the entries list.
458 :return:
459 tuple(binsha, list(tree_entry, ...))
461 A tuple of a sha and a list of tree entries being a tuple of hexsha, mode, name.
462 """
463 if si == 0:
464 for entry in entries[sl]:
465 _validate_repo_path(entry.path)
466 tree_items: List["TreeCacheTup"] = []
468 ci = sl.start
469 end = sl.stop
470 while ci < end:
471 entry = entries[ci]
472 if entry.stage != 0:
473 raise UnmergedEntriesError(entry)
474 # END abort on unmerged
475 ci += 1
476 rbound = entry.path.find("/", si)
477 if rbound == -1:
478 # It's not a tree.
479 tree_items.append((entry.binsha, entry.mode, entry.path[si:]))
480 else:
481 # Find common base range.
482 base = entry.path[si:rbound]
483 xi = ci
484 while xi < end:
485 oentry = entries[xi]
486 orbound = oentry.path.find("/", si)
487 if orbound == -1 or oentry.path[si:orbound] != base:
488 break
489 # END abort on base mismatch
490 xi += 1
491 # END find common base
493 # Enter recursion.
494 # ci - 1 as we want to count our current item as well.
495 sha, _tree_entry_list = write_tree_from_cache(entries, odb, slice(ci - 1, xi), rbound + 1)
496 tree_items.append((sha, S_IFDIR, base))
498 # Skip ahead.
499 ci = xi
500 # END handle bounds
501 # END for each entry
503 # Finally create the tree.
504 sio = BytesIO()
505 tree_to_stream(tree_items, sio.write) # Writes to stream as bytes, but doesn't change tree_items.
506 sio.seek(0)
508 istream = odb.store(IStream(str_tree_type, len(sio.getvalue()), sio))
509 return (istream.binsha, tree_items)
512def _tree_entry_to_baseindexentry(tree_entry: "TreeCacheTup", stage: int) -> BaseIndexEntry:
513 _validate_repo_path(tree_entry[2])
514 return BaseIndexEntry((tree_entry[1], tree_entry[0], stage << CE_STAGESHIFT, tree_entry[2]))
517def aggressive_tree_merge(odb: "ObjectDBR", tree_shas: Sequence[bytes]) -> List[BaseIndexEntry]:
518 R"""
519 :return:
520 List of :class:`~git.index.typ.BaseIndexEntry`\s representing the aggressive
521 merge of the given trees. All valid entries are on stage 0, whereas the
522 conflicting ones are left on stage 1, 2 or 3, whereas stage 1 corresponds to the
523 common ancestor tree, 2 to our tree and 3 to 'their' tree.
525 :param tree_shas:
526 1, 2 or 3 trees as identified by their binary 20 byte shas. If 1 or two, the
527 entries will effectively correspond to the last given tree. If 3 are given, a 3
528 way merge is performed.
529 """
530 out: List[BaseIndexEntry] = []
532 # One and two way is the same for us, as we don't have to handle an existing
533 # index, instrea
534 if len(tree_shas) in (1, 2):
535 for entry in traverse_tree_recursive(odb, tree_shas[-1], ""):
536 out.append(_tree_entry_to_baseindexentry(entry, 0))
537 # END for each entry
538 return out
539 # END handle single tree
541 if len(tree_shas) > 3:
542 raise ValueError("Cannot handle %i trees at once" % len(tree_shas))
544 # Three trees.
545 for base, ours, theirs in traverse_trees_recursive(odb, tree_shas, ""):
546 if base is not None:
547 # Base version exists.
548 if ours is not None:
549 # Ours exists.
550 if theirs is not None:
551 # It exists in all branches. Ff it was changed in both
552 # its a conflict. Otherwise, we take the changed version.
553 # This should be the most common branch, so it comes first.
554 if (base[0] != ours[0] and base[0] != theirs[0] and ours[0] != theirs[0]) or (
555 base[1] != ours[1] and base[1] != theirs[1] and ours[1] != theirs[1]
556 ):
557 # Changed by both.
558 out.append(_tree_entry_to_baseindexentry(base, 1))
559 out.append(_tree_entry_to_baseindexentry(ours, 2))
560 out.append(_tree_entry_to_baseindexentry(theirs, 3))
561 elif base[0] != ours[0] or base[1] != ours[1]:
562 # Only we changed it.
563 out.append(_tree_entry_to_baseindexentry(ours, 0))
564 else:
565 # Either nobody changed it, or they did. In either
566 # case, use theirs.
567 out.append(_tree_entry_to_baseindexentry(theirs, 0))
568 # END handle modification
569 else:
570 if ours[0] != base[0] or ours[1] != base[1]:
571 # They deleted it, we changed it, conflict.
572 out.append(_tree_entry_to_baseindexentry(base, 1))
573 out.append(_tree_entry_to_baseindexentry(ours, 2))
574 # else:
575 # # We didn't change it, ignore.
576 # pass
577 # END handle our change
578 # END handle theirs
579 else:
580 if theirs is None:
581 # Deleted in both, its fine - it's out.
582 pass
583 else:
584 if theirs[0] != base[0] or theirs[1] != base[1]:
585 # Deleted in ours, changed theirs, conflict.
586 out.append(_tree_entry_to_baseindexentry(base, 1))
587 out.append(_tree_entry_to_baseindexentry(theirs, 3))
588 # END theirs changed
589 # else:
590 # # Theirs didn't change.
591 # pass
592 # END handle theirs
593 # END handle ours
594 else:
595 # All three can't be None.
596 if ours is None:
597 # Added in their branch.
598 assert theirs is not None
599 out.append(_tree_entry_to_baseindexentry(theirs, 0))
600 elif theirs is None:
601 # Added in our branch.
602 out.append(_tree_entry_to_baseindexentry(ours, 0))
603 else:
604 # Both have it, except for the base, see whether it changed.
605 if ours[0] != theirs[0] or ours[1] != theirs[1]:
606 out.append(_tree_entry_to_baseindexentry(ours, 2))
607 out.append(_tree_entry_to_baseindexentry(theirs, 3))
608 else:
609 # It was added the same in both.
610 out.append(_tree_entry_to_baseindexentry(ours, 0))
611 # END handle two items
612 # END handle heads
613 # END handle base exists
614 # END for each entries tuple
616 return out