Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/git/objects/tree.py: 38%

1# Copyright (C) 2008, 2009 Michael Trier (mtrier@gmail.com) and contributors 

2# 

3# This module is part of GitPython and is released under the 

4# 3-Clause BSD License: https://opensource.org/license/bsd-3-clause/ 

5 

6__all__ = ["TreeModifier", "Tree"] 

7 

8import os 

9 

10import git.diff as git_diff 

11from git.util import IterableList, join_path, to_bin_sha 

12 

13from . import util 

14from .base import IndexObjUnion, IndexObject 

15from .blob import Blob 

16from .fun import tree_entries_from_data, tree_to_stream, _validate_tree_entry_name 

17from .submodule.base import Submodule 

18 

19# typing ------------------------------------------------- 

20 

21from typing import ( 

22 Any, 

23 Callable, 

24 Dict, 

25 Iterable, 

26 Iterator, 

27 List, 

28 Literal, 

29 Tuple, 

30 TYPE_CHECKING, 

31 Type, 

32 Union, 

33 cast, 

34) 

35 

36from git.types import PathLike 

37 

38if TYPE_CHECKING: 

39 from io import BytesIO 

40 

41 from git.repo import Repo 

42 

43TreeCacheTup = Tuple[bytes, int, str] 

44 

45TraversedTreeTup = Union[Tuple[Union["Tree", None], IndexObjUnion, Tuple["Submodule", "Submodule"]]] 

46 

47# -------------------------------------------------------- 

48 

49 

50def cmp(a: str, b: str) -> int: 

51 return (a > b) - (a < b) 

52 

53 

54class TreeModifier: 

55 """A utility class providing methods to alter the underlying cache in a list-like 

56 fashion. 

57 

58 Once all adjustments are complete, the :attr:`_cache`, which really is a reference 

59 to the cache of a tree, will be sorted. This ensures it will be in a serializable 

60 state. 

61 """ 

62 

63 __slots__ = ("_cache",) 

64 

65 def __init__(self, cache: List[TreeCacheTup]) -> None: 

66 self._cache = cache 

67 

68 def _index_by_name(self, name: str) -> int: 

69 """:return: index of an item with name, or -1 if not found""" 

70 for i, t in enumerate(self._cache): 

71 if t[2] == name: 

72 return i 

73 # END found item 

74 # END for each item in cache 

75 return -1 

76 

77 # { Interface 

78 def set_done(self) -> "TreeModifier": 

79 """Call this method once you are done modifying the tree information. 

80 

81 This may be called several times, but be aware that each call will cause a sort 

82 operation. 

83 

84 :return: 

85 self 

86 """ 

87 self._cache.sort(key=lambda x: (x[2] + "/") if x[1] == Tree.tree_id << 12 else x[2]) 

88 return self 

89 

90 # } END interface 

91 

92 # { Mutators 

93 def add(self, sha: bytes, mode: int, name: str, force: bool = False) -> "TreeModifier": 

94 """Add the given item to the tree. 

95 

96 If an item with the given name already exists, nothing will be done, but a 

97 :exc:`ValueError` will be raised if the sha and mode of the existing item do not 

98 match the one you add, unless `force` is ``True``. 

99 

100 :param sha: 

101 The 20 or 40 byte sha of the item to add. 

102 

103 :param mode: 

104 :class:`int` representing the stat-compatible mode of the item. 

105 

106 :param force: 

107 If ``True``, an item with your name and information will overwrite any 

108 existing item with the same name, no matter which information it has. 

109 

110 :return: 

111 self 

112 """ 

113 _validate_tree_entry_name(name, mode) 

114 if (mode >> 12) not in Tree._map_id_to_type: 

115 raise ValueError("Invalid object type according to mode %o" % mode) 

116 

117 sha = to_bin_sha(sha) 

118 index = self._index_by_name(name) 

119 

120 item = (sha, mode, name) 

121 

122 if index == -1: 

123 self._cache.append(item) 

124 else: 

125 if force: 

126 self._cache[index] = item 

127 else: 

128 ex_item = self._cache[index] 

129 if ex_item[0] != sha or ex_item[1] != mode: 

130 raise ValueError("Item %r existed with different properties" % name) 

131 # END handle mismatch 

132 # END handle force 

133 # END handle name exists 

134 return self 

135 

136 def add_unchecked(self, binsha: bytes, mode: int, name: str) -> None: 

137 """Add the given item to the tree. Its correctness is assumed, so it is the 

138 caller's responsibility to ensure that the input is correct. 

139 

140 For more information on the parameters, see :meth:`add`. 

141 

142 :param binsha: 

143 20 byte binary sha. 

144 """ 

145 assert isinstance(binsha, bytes) and isinstance(mode, int) and isinstance(name, str) 

146 tree_cache = (binsha, mode, name) 

147 

148 self._cache.append(tree_cache) 

149 

150 def __delitem__(self, name: str) -> None: 

151 """Delete an item with the given name if it exists.""" 

152 index = self._index_by_name(name) 

153 if index > -1: 

154 del self._cache[index] 

155 

156 # } END mutators 

157 

158 

159class Tree(IndexObject, git_diff.Diffable, util.Traversable, util.Serializable): 

160 R"""Tree objects represent an ordered list of :class:`~git.objects.blob.Blob`\s and 

161 other :class:`Tree`\s. 

162 

163 See :manpage:`gitglossary(7)` on "tree object": 

164 https://git-scm.com/docs/gitglossary#def_tree_object 

165 

166 Subscripting is supported, as with a list or dict: 

167 

168 * Access a specific blob using the ``tree["filename"]`` notation. 

169 * You may likewise access by index, like ``blob = tree[0]``. 

170 """ 

171 

172 type: Literal["tree"] = "tree" 

173 

174 __slots__ = ("_cache",) 

175 

176 # Actual integer IDs for comparison. 

177 commit_id = 0o16 # Equals stat.S_IFDIR | stat.S_IFLNK - a directory link. 

178 blob_id = 0o10 

179 symlink_id = 0o12 

180 tree_id = 0o04 

181 

182 _map_id_to_type: Dict[int, Type[IndexObjUnion]] = { 

183 commit_id: Submodule, 

184 blob_id: Blob, 

185 symlink_id: Blob, 

186 # Tree ID added once Tree is defined. 

187 } 

188 

189 def __init__( 

190 self, 

191 repo: "Repo", 

192 binsha: bytes, 

193 mode: int = tree_id << 12, 

194 path: Union[PathLike, None] = None, 

195 ): 

196 super().__init__(repo, binsha, mode, path) 

197 

198 @classmethod 

199 def _get_intermediate_items( 

200 cls, 

201 index_object: IndexObjUnion, 

202 ) -> Union[Tuple["Tree", ...], Tuple[()]]: 

203 if index_object.type == "tree": 

204 return tuple(index_object._iter_convert_to_object(index_object._cache)) 

205 return () 

206 

207 def _set_cache_(self, attr: str) -> None: 

208 if attr == "_cache": 

209 # Set the data when we need it. 

210 ostream = self.repo.odb.stream(self.binsha) 

211 self._cache: List[TreeCacheTup] = tree_entries_from_data(ostream.read()) 

212 else: 

213 super()._set_cache_(attr) 

214 # END handle attribute 

215 

216 def _iter_convert_to_object(self, iterable: Iterable[TreeCacheTup]) -> Iterator[IndexObjUnion]: 

217 """Iterable yields tuples of (binsha, mode, name), which will be converted to 

218 the respective object representation. 

219 """ 

220 for binsha, mode, name in iterable: 

221 path = join_path(self.path, name) 

222 try: 

223 yield self._map_id_to_type[mode >> 12](self.repo, binsha, mode, path) 

224 except KeyError as e: 

225 raise TypeError("Unknown mode %o found in tree data for path '%s'" % (mode, path)) from e 

226 # END for each item 

227 

228 def join(self, file: PathLike) -> IndexObjUnion: 

229 """Find the named object in this tree's contents. 

230 

231 :return: 

232 :class:`~git.objects.blob.Blob`, :class:`Tree`, or 

233 :class:`~git.objects.submodule.base.Submodule` 

234 

235 :raise KeyError: 

236 If the given file or tree does not exist in this tree. 

237 """ 

238 msg = "Blob or Tree named %r not found" 

239 file = os.fspath(file) 

240 if "/" in file: 

241 tree = self 

242 item = self 

243 tokens = file.split("/") 

244 for i, token in enumerate(tokens): 

245 item = tree[token] 

246 if item.type == "tree": 

247 tree = item 

248 else: 

249 # Safety assertion - blobs are at the end of the path. 

250 if i != len(tokens) - 1: 

251 raise KeyError(msg % file) 

252 return item 

253 # END handle item type 

254 # END for each token of split path 

255 if item == self: 

256 raise KeyError(msg % file) 

257 return item 

258 else: 

259 for info in self._cache: 

260 if info[2] == file: # [2] == name 

261 return self._map_id_to_type[info[1] >> 12]( 

262 self.repo, info[0], info[1], join_path(self.path, info[2]) 

263 ) 

264 # END for each obj 

265 raise KeyError(msg % file) 

266 # END handle long paths 

267 

268 def __truediv__(self, file: PathLike) -> IndexObjUnion: 

269 """The ``/`` operator is another syntax for joining. 

270 

271 See :meth:`join` for details. 

272 """ 

273 return self.join(file) 

274 

275 @property 

276 def trees(self) -> List["Tree"]: 

277 """:return: list(Tree, ...) List of trees directly below this tree""" 

278 return [i for i in self if i.type == "tree"] 

279 

280 @property 

281 def blobs(self) -> List[Blob]: 

282 """:return: list(Blob, ...) List of blobs directly below this tree""" 

283 return [i for i in self if i.type == "blob"] 

284 

285 @property 

286 def cache(self) -> TreeModifier: 

287 """ 

288 :return: 

289 An object allowing modification of the internal cache. This can be used to 

290 change the tree's contents. When done, make sure you call 

291 :meth:`~TreeModifier.set_done` on the tree modifier, or serialization 

292 behaviour will be incorrect. 

293 

294 :note: 

295 See :class:`TreeModifier` for more information on how to alter the cache. 

296 """ 

297 return TreeModifier(self._cache) 

298 

299 def traverse( 

300 self, 

301 predicate: Callable[[Union[IndexObjUnion, TraversedTreeTup], int], bool] = lambda i, d: True, 

302 prune: Callable[[Union[IndexObjUnion, TraversedTreeTup], int], bool] = lambda i, d: False, 

303 depth: int = -1, 

304 branch_first: bool = True, 

305 visit_once: bool = False, 

306 ignore_self: int = 1, 

307 as_edge: bool = False, 

308 ) -> Union[Iterator[IndexObjUnion], Iterator[TraversedTreeTup]]: 

309 """For documentation, see 

310 `Traversable._traverse() <git.objects.util.Traversable._traverse>`. 

311 

312 Trees are set to ``visit_once = False`` to gain more performance in the 

313 traversal. 

314 """ 

315 

316 # # To typecheck instead of using cast. 

317 # import itertools 

318 # def is_tree_traversed(inp: Tuple) -> TypeGuard[Tuple[Iterator[Union['Tree', 'Blob', 'Submodule']]]]: 

319 # return all(isinstance(x, (Blob, Tree, Submodule)) for x in inp[1]) 

320 

321 # ret = super().traverse(predicate, prune, depth, branch_first, visit_once, ignore_self) 

322 # ret_tup = itertools.tee(ret, 2) 

323 # assert is_tree_traversed(ret_tup), f"Type is {[type(x) for x in list(ret_tup[0])]}" 

324 # return ret_tup[0] 

325 

326 return cast( 

327 Union[Iterator[IndexObjUnion], Iterator[TraversedTreeTup]], 

328 super()._traverse( 

329 predicate, # type: ignore[arg-type] 

330 prune, # type: ignore[arg-type] 

331 depth, 

332 branch_first, 

333 visit_once, 

334 ignore_self, 

335 ), 

336 ) 

337 

338 def list_traverse(self, *args: Any, **kwargs: Any) -> IterableList[IndexObjUnion]: 

339 """ 

340 :return: 

341 :class:`~git.util.IterableList` with the results of the traversal as 

342 produced by :meth:`traverse` 

343 

344 Tree -> IterableList[Union[Submodule, Tree, Blob]] 

345 """ 

346 return super()._list_traverse(*args, **kwargs) 

347 

348 # List protocol 

349 

350 def __getslice__(self, i: int, j: int) -> List[IndexObjUnion]: 

351 return list(self._iter_convert_to_object(self._cache[i:j])) 

352 

353 def __iter__(self) -> Iterator[IndexObjUnion]: 

354 return self._iter_convert_to_object(self._cache) 

355 

356 def __len__(self) -> int: 

357 return len(self._cache) 

358 

359 def __getitem__(self, item: Union[str, int, slice]) -> IndexObjUnion: 

360 if isinstance(item, int): 

361 info = self._cache[item] 

362 return self._map_id_to_type[info[1] >> 12](self.repo, info[0], info[1], join_path(self.path, info[2])) 

363 

364 if isinstance(item, str): 

365 # compatibility 

366 return self.join(item) 

367 # END index is basestring 

368 

369 raise TypeError("Invalid index type: %r" % item) 

370 

371 def __contains__(self, item: Union[IndexObjUnion, PathLike]) -> bool: 

372 if isinstance(item, IndexObject): 

373 for info in self._cache: 

374 if item.binsha == info[0]: 

375 return True 

376 # END compare sha 

377 # END for each entry 

378 # END handle item is index object 

379 # compatibility 

380 

381 # Treat item as repo-relative path. 

382 else: 

383 path = self.path 

384 for info in self._cache: 

385 if item == join_path(path, info[2]): 

386 return True 

387 # END for each item 

388 return False 

389 

390 def __reversed__(self) -> Iterator[IndexObjUnion]: 

391 return reversed(self._iter_convert_to_object(self._cache)) # type: ignore[call-overload] 

392 

393 def _serialize(self, stream: "BytesIO") -> "Tree": 

394 """Serialize this tree into the stream. Assumes sorted tree data. 

395 

396 :note: 

397 We will assume our tree data to be in a sorted state. If this is not the 

398 case, serialization will not generate a correct tree representation as these 

399 are assumed to be sorted by algorithms. 

400 """ 

401 tree_to_stream(self._cache, stream.write) 

402 return self 

403 

404 def _deserialize(self, stream: "BytesIO") -> "Tree": 

405 self._cache = tree_entries_from_data(stream.read()) 

406 return self 

407 

408 

409# END tree 

410 

411# Finalize map definition. 

412Tree._map_id_to_type[Tree.tree_id] = Tree