Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/git/index/fun.py: 56%
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 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 run_commit_hook(name: str, index: "IndexFile", *args: str) -> None:
83 """Run the commit hook of the given name. Silently ignore hooks that do not exist.
85 :param name:
86 Name of hook, like ``pre-commit``.
88 :param index:
89 :class:`~git.index.base.IndexFile` instance.
91 :param args:
92 Arguments passed to hook file.
94 :raise git.exc.HookExecutionError:
95 """
96 hp = _commit_hook_path(name, index)
97 if not os.access(hp, os.X_OK):
98 return
100 env = os.environ.copy()
101 env["GIT_INDEX_FILE"] = safe_decode(os.fspath(index.path))
102 env["GIT_EDITOR"] = ":"
103 cmd = [hp]
104 try:
105 if sys.platform == "win32" and not _has_file_extension(hp):
106 # Windows only uses extensions to determine how to open files
107 # (doesn't understand shebangs). Try using bash to run the hook.
108 try:
109 bash_hp = osp.relpath(hp, index.repo.working_dir)
110 except ValueError:
111 # Different drives have no relative path on Windows. Git Bash accepts
112 # an absolute path in this form, although a relative path is preferable
113 # because it also works with the Windows Subsystem for Linux wrapper.
114 bash_hp = hp
115 cmd = ["bash.exe", Path(bash_hp).as_posix()]
117 process = safer_popen(
118 cmd + list(args),
119 env=env,
120 stdout=subprocess.PIPE,
121 stderr=subprocess.PIPE,
122 cwd=index.repo.working_dir,
123 )
124 except Exception as ex:
125 raise HookExecutionError(hp, ex) from ex
126 else:
127 stdout_list: List[str] = []
128 stderr_list: List[str] = []
129 handle_process_output(process, stdout_list.append, stderr_list.append, finalize_process)
130 stdout = "".join(stdout_list)
131 stderr = "".join(stderr_list)
132 if process.returncode != 0:
133 stdout = force_text(stdout, defenc)
134 stderr = force_text(stderr, defenc)
135 raise HookExecutionError(hp, process.returncode, stderr, stdout)
136 # END handle return code
139def stat_mode_to_index_mode(mode: int) -> int:
140 """Convert the given mode from a stat call to the corresponding index mode and
141 return it."""
142 if S_ISLNK(mode): # symlinks
143 return S_IFLNK
144 if S_ISDIR(mode) or S_IFMT(mode) == S_IFGITLINK: # submodules
145 return S_IFGITLINK
146 return S_IFREG | (mode & S_IXUSR and 0o755 or 0o644) # blobs with or without executable bit
149def write_cache(
150 entries: Sequence[Union[BaseIndexEntry, "IndexEntry"]],
151 stream: IO[bytes],
152 extension_data: Union[None, bytes] = None,
153 ShaStreamCls: Type[IndexFileSHA1Writer] = IndexFileSHA1Writer,
154) -> None:
155 """Write the cache represented by entries to a stream.
157 :param entries:
158 **Sorted** list of entries.
160 :param stream:
161 Stream to wrap into the AdapterStreamCls - it is used for final output.
163 :param ShaStreamCls:
164 Type to use when writing to the stream. It produces a sha while writing to it,
165 before the data is passed on to the wrapped stream.
167 :param extension_data:
168 Any kind of data to write as a trailer, it must begin a 4 byte identifier,
169 followed by its size (4 bytes).
170 """
171 # Wrap the stream into a compatible writer.
172 stream_sha = ShaStreamCls(stream)
174 tell = stream_sha.tell
175 write = stream_sha.write
177 # Header
178 version = 3 if any(entry.extended_flags for entry in entries) else 2
179 write(b"DIRC")
180 write(pack(">LL", version, len(entries)))
182 # Body
183 for entry in entries:
184 beginoffset = tell()
185 write(entry.ctime_bytes) # ctime
186 write(entry.mtime_bytes) # mtime
187 path_str = str(entry.path)
188 path: bytes = force_bytes(path_str, encoding=defenc)
189 plen = len(path) & CE_NAMEMASK # Path length
190 assert plen == len(path), "Path %s too long to fit into index" % entry.path
191 flags = plen | (entry.flags & CE_NAMEMASK_INV) # Clear possible previous values.
192 if entry.extended_flags:
193 flags |= CE_EXTENDED
194 write(
195 pack(
196 ">LLLLLL20sH",
197 entry.dev,
198 entry.inode,
199 entry.mode,
200 entry.uid,
201 entry.gid,
202 entry.size,
203 entry.binsha,
204 flags,
205 )
206 )
207 if entry.extended_flags:
208 write(pack(">H", entry.extended_flags))
209 write(path)
210 real_size = (tell() - beginoffset + 8) & ~7
211 write(b"\0" * ((beginoffset + real_size) - tell()))
212 # END for each entry
214 # Write previously cached extensions data.
215 if extension_data is not None:
216 stream_sha.write(extension_data)
218 # Write the sha over the content.
219 stream_sha.write_sha()
222def read_header(stream: IO[bytes]) -> Tuple[int, int]:
223 """Return tuple(version_long, num_entries) from the given stream."""
224 type_id = stream.read(4)
225 if type_id != b"DIRC":
226 raise AssertionError("Invalid index file header: %r" % type_id)
227 unpacked = cast(Tuple[int, int], unpack(">LL", stream.read(4 * 2)))
228 version, num_entries = unpacked
230 assert version in (1, 2, 3), "Unsupported git index version %i, only 1, 2, and 3 are supported" % version
231 return version, num_entries
234def entry_key(*entry: Union[BaseIndexEntry, PathLike, int]) -> Tuple[PathLike, int]:
235 """
236 :return:
237 Key suitable to be used for the
238 :attr:`index.entries <git.index.base.IndexFile.entries>` dictionary.
240 :param entry:
241 One instance of type BaseIndexEntry or the path and the stage.
242 """
244 # def is_entry_key_tup(entry_key: Tuple) -> TypeGuard[Tuple[PathLike, int]]:
245 # return isinstance(entry_key, tuple) and len(entry_key) == 2
247 if len(entry) == 1:
248 entry_first = entry[0]
249 assert isinstance(entry_first, BaseIndexEntry)
250 return (entry_first.path, entry_first.stage)
251 else:
252 # assert is_entry_key_tup(entry)
253 entry = cast(Tuple[PathLike, int], entry)
254 return entry
255 # END handle entry
258def read_cache(
259 stream: IO[bytes],
260) -> Tuple[int, Dict[Tuple[PathLike, int], "IndexEntry"], bytes, bytes]:
261 """Read a cache file from the given stream.
263 :return:
264 tuple(version, entries_dict, extension_data, content_sha)
266 * *version* is the integer version number.
267 * *entries_dict* is a dictionary which maps IndexEntry instances to a path at a
268 stage.
269 * *extension_data* is ``""`` or 4 bytes of type + 4 bytes of size + size bytes.
270 * *content_sha* is a 20 byte sha on all cache file contents.
271 """
272 version, num_entries = read_header(stream)
273 count = 0
274 entries: Dict[Tuple[PathLike, int], "IndexEntry"] = {}
276 read = stream.read
277 tell = stream.tell
278 while count < num_entries:
279 beginoffset = tell()
280 ctime = unpack(">8s", read(8))[0]
281 mtime = unpack(">8s", read(8))[0]
282 (dev, ino, mode, uid, gid, size, sha, flags) = unpack(">LLLLLL20sH", read(20 + 4 * 6 + 2))
283 extended_flags = 0
284 if flags & CE_EXTENDED:
285 extended_flags = unpack(">H", read(2))[0]
286 path_size = flags & CE_NAMEMASK
287 path = read(path_size).decode(defenc)
289 real_size = (tell() - beginoffset + 8) & ~7
290 read((beginoffset + real_size) - tell())
291 entry = IndexEntry((mode, sha, flags, path, ctime, mtime, dev, ino, uid, gid, size, extended_flags))
292 # entry_key would be the method to use, but we save the effort.
293 entries[(path, entry.stage)] = entry
294 count += 1
295 # END for each entry
297 # The footer contains extension data and a sha on the content so far.
298 # Keep the extension footer,and verify we have a sha in the end.
299 # Extension data format is:
300 # 4 bytes ID
301 # 4 bytes length of chunk
302 # Repeated 0 - N times
303 extension_data = stream.read(~0)
304 assert len(extension_data) > 19, (
305 "Index Footer was not at least a sha on content as it was only %i bytes in size" % len(extension_data)
306 )
308 content_sha = extension_data[-20:]
310 # Truncate the sha in the end as we will dynamically create it anyway.
311 extension_data = extension_data[:-20]
313 return (version, entries, extension_data, content_sha)
316def write_tree_from_cache(
317 entries: List[IndexEntry], odb: "GitCmdObjectDB", sl: slice, si: int = 0
318) -> Tuple[bytes, List["TreeCacheTup"]]:
319 R"""Create a tree from the given sorted list of entries and put the respective
320 trees into the given object database.
322 :param entries:
323 **Sorted** list of :class:`~git.index.typ.IndexEntry`\s.
325 :param odb:
326 Object database to store the trees in.
328 :param si:
329 Start index at which we should start creating subtrees.
331 :param sl:
332 Slice indicating the range we should process on the entries list.
334 :return:
335 tuple(binsha, list(tree_entry, ...))
337 A tuple of a sha and a list of tree entries being a tuple of hexsha, mode, name.
338 """
339 tree_items: List["TreeCacheTup"] = []
341 ci = sl.start
342 end = sl.stop
343 while ci < end:
344 entry = entries[ci]
345 if entry.stage != 0:
346 raise UnmergedEntriesError(entry)
347 # END abort on unmerged
348 ci += 1
349 rbound = entry.path.find("/", si)
350 if rbound == -1:
351 # It's not a tree.
352 tree_items.append((entry.binsha, entry.mode, entry.path[si:]))
353 else:
354 # Find common base range.
355 base = entry.path[si:rbound]
356 xi = ci
357 while xi < end:
358 oentry = entries[xi]
359 orbound = oentry.path.find("/", si)
360 if orbound == -1 or oentry.path[si:orbound] != base:
361 break
362 # END abort on base mismatch
363 xi += 1
364 # END find common base
366 # Enter recursion.
367 # ci - 1 as we want to count our current item as well.
368 sha, _tree_entry_list = write_tree_from_cache(entries, odb, slice(ci - 1, xi), rbound + 1)
369 tree_items.append((sha, S_IFDIR, base))
371 # Skip ahead.
372 ci = xi
373 # END handle bounds
374 # END for each entry
376 # Finally create the tree.
377 sio = BytesIO()
378 tree_to_stream(tree_items, sio.write) # Writes to stream as bytes, but doesn't change tree_items.
379 sio.seek(0)
381 istream = odb.store(IStream(str_tree_type, len(sio.getvalue()), sio))
382 return (istream.binsha, tree_items)
385def _tree_entry_to_baseindexentry(tree_entry: "TreeCacheTup", stage: int) -> BaseIndexEntry:
386 return BaseIndexEntry((tree_entry[1], tree_entry[0], stage << CE_STAGESHIFT, tree_entry[2]))
389def aggressive_tree_merge(odb: "GitCmdObjectDB", tree_shas: Sequence[bytes]) -> List[BaseIndexEntry]:
390 R"""
391 :return:
392 List of :class:`~git.index.typ.BaseIndexEntry`\s representing the aggressive
393 merge of the given trees. All valid entries are on stage 0, whereas the
394 conflicting ones are left on stage 1, 2 or 3, whereas stage 1 corresponds to the
395 common ancestor tree, 2 to our tree and 3 to 'their' tree.
397 :param tree_shas:
398 1, 2 or 3 trees as identified by their binary 20 byte shas. If 1 or two, the
399 entries will effectively correspond to the last given tree. If 3 are given, a 3
400 way merge is performed.
401 """
402 out: List[BaseIndexEntry] = []
404 # One and two way is the same for us, as we don't have to handle an existing
405 # index, instrea
406 if len(tree_shas) in (1, 2):
407 for entry in traverse_tree_recursive(odb, tree_shas[-1], ""):
408 out.append(_tree_entry_to_baseindexentry(entry, 0))
409 # END for each entry
410 return out
411 # END handle single tree
413 if len(tree_shas) > 3:
414 raise ValueError("Cannot handle %i trees at once" % len(tree_shas))
416 # Three trees.
417 for base, ours, theirs in traverse_trees_recursive(odb, tree_shas, ""):
418 if base is not None:
419 # Base version exists.
420 if ours is not None:
421 # Ours exists.
422 if theirs is not None:
423 # It exists in all branches. Ff it was changed in both
424 # its a conflict. Otherwise, we take the changed version.
425 # This should be the most common branch, so it comes first.
426 if (base[0] != ours[0] and base[0] != theirs[0] and ours[0] != theirs[0]) or (
427 base[1] != ours[1] and base[1] != theirs[1] and ours[1] != theirs[1]
428 ):
429 # Changed by both.
430 out.append(_tree_entry_to_baseindexentry(base, 1))
431 out.append(_tree_entry_to_baseindexentry(ours, 2))
432 out.append(_tree_entry_to_baseindexentry(theirs, 3))
433 elif base[0] != ours[0] or base[1] != ours[1]:
434 # Only we changed it.
435 out.append(_tree_entry_to_baseindexentry(ours, 0))
436 else:
437 # Either nobody changed it, or they did. In either
438 # case, use theirs.
439 out.append(_tree_entry_to_baseindexentry(theirs, 0))
440 # END handle modification
441 else:
442 if ours[0] != base[0] or ours[1] != base[1]:
443 # They deleted it, we changed it, conflict.
444 out.append(_tree_entry_to_baseindexentry(base, 1))
445 out.append(_tree_entry_to_baseindexentry(ours, 2))
446 # else:
447 # # We didn't change it, ignore.
448 # pass
449 # END handle our change
450 # END handle theirs
451 else:
452 if theirs is None:
453 # Deleted in both, its fine - it's out.
454 pass
455 else:
456 if theirs[0] != base[0] or theirs[1] != base[1]:
457 # Deleted in ours, changed theirs, conflict.
458 out.append(_tree_entry_to_baseindexentry(base, 1))
459 out.append(_tree_entry_to_baseindexentry(theirs, 3))
460 # END theirs changed
461 # else:
462 # # Theirs didn't change.
463 # pass
464 # END handle theirs
465 # END handle ours
466 else:
467 # All three can't be None.
468 if ours is None:
469 # Added in their branch.
470 assert theirs is not None
471 out.append(_tree_entry_to_baseindexentry(theirs, 0))
472 elif theirs is None:
473 # Added in our branch.
474 out.append(_tree_entry_to_baseindexentry(ours, 0))
475 else:
476 # Both have it, except for the base, see whether it changed.
477 if ours[0] != theirs[0] or ours[1] != theirs[1]:
478 out.append(_tree_entry_to_baseindexentry(ours, 2))
479 out.append(_tree_entry_to_baseindexentry(theirs, 3))
480 else:
481 # It was added the same in both.
482 out.append(_tree_entry_to_baseindexentry(ours, 0))
483 # END handle two items
484 # END handle heads
485 # END handle base exists
486 # END for each entries tuple
488 return out