Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/git/objects/fun.py: 32%
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"""Functions that are supposed to be as fast as possible."""
6__all__ = [
7 "tree_to_stream",
8 "tree_entries_from_data",
9 "traverse_trees_recursive",
10 "traverse_tree_recursive",
11]
13from stat import S_ISDIR
15from git.compat import safe_decode, defenc
16from git.util import _validate_repo_path
18# typing ----------------------------------------------
20from typing import (
21 Callable,
22 List,
23 MutableSequence,
24 Sequence,
25 Tuple,
26 TYPE_CHECKING,
27 Union,
28 overload,
29)
31if TYPE_CHECKING:
32 from _typeshed import ReadableBuffer
34 from gitdb.db.base import ObjectDBR
36EntryTup = Tuple[bytes, int, str] # Same as TreeCacheTup in tree.py.
37EntryTupOrNone = Union[EntryTup, None]
39# ---------------------------------------------------
42def _validate_tree_entry_name(name: str, mode: Union[int, None] = None) -> None:
43 if "/" in name:
44 raise ValueError("Tree entry names must not contain '/' characters")
45 # A tree name is a component, not a rooted path; a colon cannot select a drive.
46 _validate_repo_path("tree/" + name, mode)
49def tree_to_stream(entries: Sequence[EntryTup], write: Callable[["ReadableBuffer"], Union[int, None]]) -> None:
50 """Write the given list of entries into a stream using its ``write`` method.
52 :param entries:
53 **Sorted** list of tuples with (binsha, mode, name).
55 :param write:
56 A ``write`` method which takes a data string.
57 """
58 ord_zero = ord("0")
59 bit_mask = 7 # 3 bits set.
61 for binsha, mode, name in entries:
62 if len(binsha) != 20:
63 raise ValueError("Tree entry object IDs must be exactly 20 bytes")
64 if mode >> 12 not in (4, 8, 10, 14):
65 raise ValueError("Invalid tree entry mode")
66 mode_str = b""
67 for i in range(6):
68 mode_str = bytes([((mode >> (i * 3)) & bit_mask) + ord_zero]) + mode_str
69 # END for each 8 octal value
71 # git slices away the first octal if it's zero.
72 if mode_str[0] == ord_zero:
73 mode_str = mode_str[1:]
74 # END save a byte
76 # Here it comes: If the name is actually unicode, the replacement below will not
77 # work as the binsha is not part of the ascii unicode encoding - hence we must
78 # convert to an UTF-8 string for it to work properly. According to my tests,
79 # this is exactly what git does, that is it just takes the input literally,
80 # which appears to be UTF-8 on linux.
81 if isinstance(name, str):
82 name_bytes = name.encode(defenc)
83 else:
84 name_bytes = name # type: ignore[unreachable] # check runtime types - is always str?
85 _validate_tree_entry_name(safe_decode(name_bytes), mode)
86 write(b"".join((mode_str, b" ", name_bytes, b"\0", binsha)))
87 # END for each item
90def tree_entries_from_data(data: bytes) -> List[EntryTup]:
91 """Read complete tree records, rejecting invalid names and truncated fields.
93 :param data:
94 Data block with tree data (as bytes).
96 :return:
97 list(tuple(binsha, mode, tree_relative_path), ...)
98 """
99 out = []
100 offset = 0
101 while offset < len(data):
102 mode_end = data.find(b" ", offset)
103 if mode_end < 0:
104 raise ValueError("Unterminated tree entry mode")
105 mode_bytes = data[offset:mode_end]
106 if not mode_bytes or mode_bytes.strip(b"01234567"):
107 raise ValueError("Invalid tree entry mode")
108 mode = int(mode_bytes, 8)
109 if mode >> 12 not in (4, 8, 10, 14):
110 raise ValueError("Invalid tree entry mode")
111 name_end = data.find(b"\0", mode_end + 1)
112 if name_end < 0 or name_end + 21 > len(data):
113 raise ValueError("Truncated tree entry")
114 name = safe_decode(bytes(data[mode_end + 1 : name_end]))
115 _validate_tree_entry_name(name, mode)
116 offset = name_end + 21
117 out.append((bytes(data[name_end + 1 : offset]), mode, name))
118 return out
121def _find_by_name(tree_data: MutableSequence[EntryTupOrNone], name: str, is_dir: bool, start_at: int) -> EntryTupOrNone:
122 """Return data entry matching the given name and tree mode or ``None``.
124 Before the item is returned, the respective data item is set None in the `tree_data`
125 list to mark it done.
126 """
128 try:
129 item = tree_data[start_at]
130 if item and item[2] == name and S_ISDIR(item[1]) == is_dir:
131 tree_data[start_at] = None
132 return item
133 except IndexError:
134 pass
135 # END exception handling
136 for index, item in enumerate(tree_data):
137 if item and item[2] == name and S_ISDIR(item[1]) == is_dir:
138 tree_data[index] = None
139 return item
140 # END if item matches
141 # END for each item
142 return None
145@overload
146def _to_full_path(item: None, path_prefix: str) -> None: ...
149@overload
150def _to_full_path(item: EntryTup, path_prefix: str) -> EntryTup: ...
153def _to_full_path(item: EntryTupOrNone, path_prefix: str) -> EntryTupOrNone:
154 """Rebuild entry with given path prefix."""
155 if not item:
156 return item
157 return (item[0], item[1], path_prefix + item[2])
160def traverse_trees_recursive(
161 odb: "ObjectDBR", tree_shas: Sequence[Union[bytes, None]], path_prefix: str
162) -> List[Tuple[EntryTupOrNone, ...]]:
163 """
164 :return:
165 List of list with entries according to the given binary tree-shas.
167 The result is encoded in a list
168 of n tuple|None per blob/commit, (n == len(tree_shas)), where:
170 * [0] == 20 byte sha
171 * [1] == mode as int
172 * [2] == path relative to working tree root
174 The entry tuple is ``None`` if the respective blob/commit did not exist in the
175 given tree.
177 :param tree_shas:
178 Iterable of shas pointing to trees. All trees must be on the same level.
179 A tree-sha may be ``None``, in which case ``None``.
181 :param path_prefix:
182 A prefix to be added to the returned paths on this level.
183 Set it ``""`` for the first iteration.
185 :note:
186 The ordering of the returned items will be partially lost.
187 """
188 trees_data: List[List[EntryTupOrNone]] = []
190 nt = len(tree_shas)
191 for tree_sha in tree_shas:
192 if tree_sha is None:
193 data: List[EntryTupOrNone] = []
194 else:
195 # Make new list for typing as list invariant.
196 data = list(tree_entries_from_data(odb.stream(tree_sha).read()))
197 # END handle muted trees
198 trees_data.append(data)
199 # END for each sha to get data for
201 out: List[Tuple[EntryTupOrNone, ...]] = []
203 # Find all matching entries and recursively process them together if the match is a
204 # tree. If the match is a non-tree item, put it into the result.
205 # Processed items will be set None.
206 for ti, tree_data in enumerate(trees_data):
207 for ii, item in enumerate(tree_data):
208 if not item:
209 continue
210 # END skip already done items
211 entries: List[EntryTupOrNone]
212 entries = [None for _ in range(nt)]
213 entries[ti] = item
214 _sha, mode, name = item
215 is_dir = S_ISDIR(mode) # Type mode bits
217 # Find this item in all other tree data items.
218 # Wrap around, but stop one before our current index, hence ti+nt, not
219 # ti+1+nt.
220 for tio in range(ti + 1, ti + nt):
221 tio = tio % nt
222 entries[tio] = _find_by_name(trees_data[tio], name, is_dir, ii)
224 # END for each other item data
225 # If we are a directory, enter recursion.
226 if is_dir:
227 out.extend(
228 traverse_trees_recursive(
229 odb,
230 [((ei and ei[0]) or None) for ei in entries],
231 path_prefix + name + "/",
232 )
233 )
234 else:
235 out.append(tuple(_to_full_path(e, path_prefix) for e in entries))
237 # END handle recursion
238 # Finally mark it done.
239 tree_data[ii] = None
240 # END for each item
242 # We are done with one tree, set all its data empty.
243 del tree_data[:]
244 # END for each tree_data chunk
245 return out
248def traverse_tree_recursive(odb: "ObjectDBR", tree_sha: bytes, path_prefix: str) -> List[EntryTup]:
249 """
250 :return:
251 List of entries of the tree pointed to by the binary `tree_sha`.
253 An entry has the following format:
255 * [0] 20 byte sha
256 * [1] mode as int
257 * [2] path relative to the repository
259 :param path_prefix:
260 Prefix to prepend to the front of all returned paths.
261 """
262 entries = []
263 data = tree_entries_from_data(odb.stream(tree_sha).read())
265 # Unpacking/packing is faster than accessing individual items.
266 for sha, mode, name in data:
267 if S_ISDIR(mode):
268 entries.extend(traverse_tree_recursive(odb, sha, path_prefix + name + "/"))
269 else:
270 entries.append((sha, mode, path_prefix + name))
271 # END for each item
273 return entries