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 # Prefer Bash associated with GitPython's selected Git installation. If
202 # that layout is not recognized, use an explicitly configured non-system
203 # PATH entry. Preserve the bare fallback for installations that previously
204 # relied on WSL or another CreateProcess-resolved Bash.
205 found_bash = _git_for_windows_bash() or _which_from_path("bash.exe")
206 try:
207 bash_hp = osp.relpath(hp, index.repo.working_dir)
208 except ValueError:
209 # Different drives have no relative path on Windows. Git Bash accepts
210 # an absolute path in this form, although a relative path is preferable
211 # because it also works with the Windows Subsystem for Linux wrapper.
212 bash_hp = hp
213 else:
214 # Bash resolves a relative path in its own view of the filesystem,
215 # which differs from Windows' across a mount point: Git Bash mounts
216 # the user's temporary directory as /tmp, so from a working tree
217 # there, ".." leads elsewhere. A path leaving the working tree is
218 # thus passed whole to a Bash found above, which accepts it; the
219 # bare fallback may be WSL's, which does not, and keeps it relative.
220 if found_bash and bash_hp.startswith(os.pardir + os.sep):
221 bash_hp = hp
222 cmd = [found_bash or "bash.exe", Path(bash_hp).as_posix()]
224 process = safer_popen(
225 cmd + list(args),
226 env=env,
227 stdout=subprocess.PIPE,
228 stderr=subprocess.PIPE,
229 cwd=index.repo.working_dir,
230 )
231 except Exception as ex:
232 raise HookExecutionError(hp, ex) from ex
233 else:
234 stdout_list: List[str] = []
235 stderr_list: List[str] = []
236 handle_process_output(process, stdout_list.append, stderr_list.append, finalize_process)
237 stdout = "".join(stdout_list)
238 stderr = "".join(stderr_list)
239 if process.returncode != 0:
240 stdout = force_text(stdout, defenc)
241 stderr = force_text(stderr, defenc)
242 raise HookExecutionError(hp, process.returncode, stderr, stdout)
243 # END handle return code
246def stat_mode_to_index_mode(mode: int) -> int:
247 """Convert the given mode from a stat call to the corresponding index mode and
248 return it."""
249 if S_ISLNK(mode): # symlinks
250 return S_IFLNK
251 if S_ISDIR(mode) or S_IFMT(mode) == S_IFGITLINK: # submodules
252 return S_IFGITLINK
253 return S_IFREG | (mode & S_IXUSR and 0o755 or 0o644) # blobs with or without executable bit
256def write_cache(
257 entries: Sequence[Union[BaseIndexEntry, "IndexEntry"]],
258 stream: IO[bytes],
259 extension_data: Union[None, bytes] = None,
260 ShaStreamCls: Type[IndexFileSHA1Writer] = IndexFileSHA1Writer,
261) -> None:
262 """Write the cache represented by entries to a stream.
264 :param entries:
265 **Sorted** list of entries.
267 :param stream:
268 Stream to wrap into the AdapterStreamCls - it is used for final output.
270 :param ShaStreamCls:
271 Type to use when writing to the stream. It produces a sha while writing to it,
272 before the data is passed on to the wrapped stream.
274 :param extension_data:
275 Any kind of data to write as a trailer, it must begin a 4 byte identifier,
276 followed by its size (4 bytes).
277 """
278 # Wrap the stream into a compatible writer.
279 stream_sha = ShaStreamCls(stream)
281 tell = stream_sha.tell
282 write = stream_sha.write
284 # Header
285 version = 3 if any(entry.extended_flags for entry in entries) else 2
286 write(b"DIRC")
287 write(pack(">LL", version, len(entries)))
289 # Body
290 for entry in entries:
291 _validate_repo_path(entry.path, entry.mode)
292 beginoffset = tell()
293 write(entry.ctime_bytes) # ctime
294 write(entry.mtime_bytes) # mtime
295 path_str = os.fspath(entry.path)
296 path: bytes = force_bytes(path_str, encoding=defenc)
297 plen = min(len(path), CE_NAMEMASK) # Longer names use a sentinel.
298 flags = plen | (entry.flags & CE_NAMEMASK_INV) # Clear possible previous values.
299 if entry.extended_flags:
300 flags |= CE_EXTENDED
301 write(
302 pack(
303 ">LLLLLL20sH",
304 entry.dev,
305 entry.inode,
306 entry.mode,
307 entry.uid,
308 entry.gid,
309 entry.size,
310 entry.binsha,
311 flags,
312 )
313 )
314 if entry.extended_flags:
315 write(pack(">H", entry.extended_flags))
316 write(path)
317 real_size = (tell() - beginoffset + 8) & ~7
318 write(b"\0" * ((beginoffset + real_size) - tell()))
319 # END for each entry
321 # Write previously cached extensions data.
322 if extension_data is not None:
323 stream_sha.write(extension_data)
325 # Write the sha over the content.
326 stream_sha.write_sha()
329def read_header(stream: IO[bytes]) -> Tuple[int, int]:
330 """Return tuple(version_long, num_entries) from the given stream."""
331 type_id = stream.read(4)
332 if type_id != b"DIRC":
333 raise AssertionError("Invalid index file header: %r" % type_id)
334 unpacked = cast(Tuple[int, int], unpack(">LL", stream.read(4 * 2)))
335 version, num_entries = unpacked
337 if version not in (1, 2, 3):
338 raise AssertionError("Unsupported git index version %i, only 1, 2, and 3 are supported" % version)
339 return version, num_entries
342def entry_key(*entry: Union[BaseIndexEntry, PathLike, int]) -> Tuple[PathLike, int]:
343 """
344 :return:
345 Key suitable to be used for the
346 :attr:`index.entries <git.index.base.IndexFile.entries>` dictionary.
348 :param entry:
349 One instance of type BaseIndexEntry or the path and the stage.
350 """
352 # def is_entry_key_tup(entry_key: Tuple) -> TypeGuard[Tuple[PathLike, int]]:
353 # return isinstance(entry_key, tuple) and len(entry_key) == 2
355 if len(entry) == 1:
356 entry_first = entry[0]
357 assert isinstance(entry_first, BaseIndexEntry)
358 return (entry_first.path, entry_first.stage)
359 else:
360 # assert is_entry_key_tup(entry)
361 entry = cast(Tuple[PathLike, int], entry)
362 return entry
363 # END handle entry
366def read_cache(
367 stream: IO[bytes],
368) -> Tuple[int, Dict[Tuple[PathLike, int], "IndexEntry"], bytes, bytes]:
369 """Read a cache file from the given stream.
371 :return:
372 tuple(version, entries_dict, extension_data, content_sha)
374 * *version* is the integer version number.
375 * *entries_dict* is a dictionary which maps IndexEntry instances to a path at a
376 stage.
377 * *extension_data* is ``""`` or 4 bytes of type + 4 bytes of size + size bytes.
378 * *content_sha* is a 20 byte sha on all cache file contents.
379 """
380 version, num_entries = read_header(stream)
381 count = 0
382 entries: Dict[Tuple[PathLike, int], "IndexEntry"] = {}
384 read = stream.read
385 tell = stream.tell
386 while count < num_entries:
387 beginoffset = tell()
388 ctime = unpack(">8s", read(8))[0]
389 mtime = unpack(">8s", read(8))[0]
390 (dev, ino, mode, uid, gid, size, sha, flags) = unpack(">LLLLLL20sH", read(20 + 4 * 6 + 2))
391 extended_flags = 0
392 if flags & CE_EXTENDED:
393 extended_flags = unpack(">H", read(2))[0]
394 path_size = flags & CE_NAMEMASK
395 path_bytes = bytearray(read(path_size))
396 if len(path_bytes) != path_size:
397 raise ValueError("Truncated index entry path")
398 terminator = read(1)
399 if path_size == CE_NAMEMASK:
400 while terminator and terminator != b"\0":
401 path_bytes.extend(terminator)
402 terminator = read(1)
403 if terminator != b"\0":
404 raise ValueError("Unterminated index entry path")
405 path = path_bytes.decode(defenc)
406 _validate_repo_path(path, mode)
408 real_size = (tell() - beginoffset + 7) & ~7
409 padding_size = beginoffset + real_size - tell()
410 if read(padding_size) != b"\0" * padding_size:
411 raise ValueError("Invalid index entry padding")
412 entry = IndexEntry((mode, sha, flags, path, ctime, mtime, dev, ino, uid, gid, size, extended_flags))
413 # entry_key would be the method to use, but we save the effort.
414 entries[(path, entry.stage)] = entry
415 count += 1
416 # END for each entry
418 # The footer contains extension data and a sha on the content so far.
419 # Keep the extension footer,and verify we have a sha in the end.
420 # Extension data format is:
421 # 4 bytes ID
422 # 4 bytes length of chunk
423 # Repeated 0 - N times
424 extension_data = stream.read(~0)
425 assert len(extension_data) > 19, (
426 "Index Footer was not at least a sha on content as it was only %i bytes in size" % len(extension_data)
427 )
429 content_sha = extension_data[-20:]
431 # Truncate the sha in the end as we will dynamically create it anyway.
432 extension_data = extension_data[:-20]
434 offset = 0
435 while offset < len(extension_data):
436 header = extension_data[offset : offset + 8]
437 if len(header) != 8:
438 raise ValueError("Truncated index extension header")
439 signature, size = unpack(">4sL", header)
440 if not b"A" <= signature[:1] <= b"Z":
441 raise ValueError("Unsupported mandatory index extension %r" % signature)
442 offset += 8 + size
443 if offset > len(extension_data):
444 raise ValueError("Truncated index extension %r" % signature)
446 return (version, entries, extension_data, content_sha)
449def write_tree_from_cache(
450 entries: List[IndexEntry], odb: "ObjectDBW", sl: slice, si: int = 0
451) -> Tuple[bytes, List["TreeCacheTup"]]:
452 R"""Create a tree from the given sorted list of entries and put the respective
453 trees into the given object database.
455 :param entries:
456 **Sorted** list of :class:`~git.index.typ.IndexEntry`\s.
458 :param odb:
459 Object database to store the trees in.
461 :param si:
462 Start index at which we should start creating subtrees.
464 :param sl:
465 Slice indicating the range we should process on the entries list.
467 :return:
468 tuple(binsha, list(tree_entry, ...))
470 A tuple of a sha and a list of tree entries being a tuple of hexsha, mode, name.
471 """
472 if si == 0:
473 for entry in entries[sl]:
474 _validate_repo_path(entry.path, entry.mode)
475 tree_items: List["TreeCacheTup"] = []
477 ci = sl.start
478 end = sl.stop
479 while ci < end:
480 entry = entries[ci]
481 if entry.stage != 0:
482 raise UnmergedEntriesError(entry)
483 # END abort on unmerged
484 ci += 1
485 rbound = entry.path.find("/", si)
486 if rbound == -1:
487 # It's not a tree.
488 tree_items.append((entry.binsha, entry.mode, entry.path[si:]))
489 else:
490 # Find common base range.
491 base = entry.path[si:rbound]
492 xi = ci
493 while xi < end:
494 oentry = entries[xi]
495 orbound = oentry.path.find("/", si)
496 if orbound == -1 or oentry.path[si:orbound] != base:
497 break
498 # END abort on base mismatch
499 xi += 1
500 # END find common base
502 # Enter recursion.
503 # ci - 1 as we want to count our current item as well.
504 sha, _tree_entry_list = write_tree_from_cache(entries, odb, slice(ci - 1, xi), rbound + 1)
505 tree_items.append((sha, S_IFDIR, base))
507 # Skip ahead.
508 ci = xi
509 # END handle bounds
510 # END for each entry
512 # Finally create the tree.
513 sio = BytesIO()
514 tree_to_stream(tree_items, sio.write) # Writes to stream as bytes, but doesn't change tree_items.
515 sio.seek(0)
517 istream = odb.store(IStream(str_tree_type, len(sio.getvalue()), sio))
518 return (istream.binsha, tree_items)
521def _tree_entry_to_baseindexentry(tree_entry: "TreeCacheTup", stage: int) -> BaseIndexEntry:
522 _validate_repo_path(tree_entry[2], tree_entry[1])
523 return BaseIndexEntry((tree_entry[1], tree_entry[0], stage << CE_STAGESHIFT, tree_entry[2]))
526def aggressive_tree_merge(odb: "ObjectDBR", tree_shas: Sequence[bytes]) -> List[BaseIndexEntry]:
527 R"""
528 :return:
529 List of :class:`~git.index.typ.BaseIndexEntry`\s representing the aggressive
530 merge of the given trees. All valid entries are on stage 0, whereas the
531 conflicting ones are left on stage 1, 2 or 3, whereas stage 1 corresponds to the
532 common ancestor tree, 2 to our tree and 3 to 'their' tree.
534 :param tree_shas:
535 1, 2 or 3 trees as identified by their binary 20 byte shas. If 1 or two, the
536 entries will effectively correspond to the last given tree. If 3 are given, a 3
537 way merge is performed.
538 """
539 out: List[BaseIndexEntry] = []
541 # One and two way is the same for us, as we don't have to handle an existing
542 # index, instrea
543 if len(tree_shas) in (1, 2):
544 for entry in traverse_tree_recursive(odb, tree_shas[-1], ""):
545 out.append(_tree_entry_to_baseindexentry(entry, 0))
546 # END for each entry
547 return out
548 # END handle single tree
550 if len(tree_shas) > 3:
551 raise ValueError("Cannot handle %i trees at once" % len(tree_shas))
553 # Three trees.
554 for base, ours, theirs in traverse_trees_recursive(odb, tree_shas, ""):
555 if base is not None:
556 # Base version exists.
557 if ours is not None:
558 # Ours exists.
559 if theirs is not None:
560 # It exists in all branches. Ff it was changed in both
561 # its a conflict. Otherwise, we take the changed version.
562 # This should be the most common branch, so it comes first.
563 if (base[0] != ours[0] and base[0] != theirs[0] and ours[0] != theirs[0]) or (
564 base[1] != ours[1] and base[1] != theirs[1] and ours[1] != theirs[1]
565 ):
566 # Changed by both.
567 out.append(_tree_entry_to_baseindexentry(base, 1))
568 out.append(_tree_entry_to_baseindexentry(ours, 2))
569 out.append(_tree_entry_to_baseindexentry(theirs, 3))
570 elif base[0] != ours[0] or base[1] != ours[1]:
571 # Only we changed it.
572 out.append(_tree_entry_to_baseindexentry(ours, 0))
573 else:
574 # Either nobody changed it, or they did. In either
575 # case, use theirs.
576 out.append(_tree_entry_to_baseindexentry(theirs, 0))
577 # END handle modification
578 else:
579 if ours[0] != base[0] or ours[1] != base[1]:
580 # They deleted it, we changed it, conflict.
581 out.append(_tree_entry_to_baseindexentry(base, 1))
582 out.append(_tree_entry_to_baseindexentry(ours, 2))
583 # else:
584 # # We didn't change it, ignore.
585 # pass
586 # END handle our change
587 # END handle theirs
588 else:
589 if theirs is None:
590 # Deleted in both, its fine - it's out.
591 pass
592 else:
593 if theirs[0] != base[0] or theirs[1] != base[1]:
594 # Deleted in ours, changed theirs, conflict.
595 out.append(_tree_entry_to_baseindexentry(base, 1))
596 out.append(_tree_entry_to_baseindexentry(theirs, 3))
597 # END theirs changed
598 # else:
599 # # Theirs didn't change.
600 # pass
601 # END handle theirs
602 # END handle ours
603 else:
604 # All three can't be None.
605 if ours is None:
606 # Added in their branch.
607 assert theirs is not None
608 out.append(_tree_entry_to_baseindexentry(theirs, 0))
609 elif theirs is None:
610 # Added in our branch.
611 out.append(_tree_entry_to_baseindexentry(ours, 0))
612 else:
613 # Both have it, except for the base, see whether it changed.
614 if ours[0] != theirs[0] or ours[1] != theirs[1]:
615 out.append(_tree_entry_to_baseindexentry(ours, 2))
616 out.append(_tree_entry_to_baseindexentry(theirs, 3))
617 else:
618 # It was added the same in both.
619 out.append(_tree_entry_to_baseindexentry(ours, 0))
620 # END handle two items
621 # END handle heads
622 # END handle base exists
623 # END for each entries tuple
625 return out