Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/git/refs/symbolic.py: 16%

335 statements  

« prev     ^ index     » next       coverage.py v6.4.4, created at 2023-07-17 14:22 -0600

1from git.types import PathLike 

2import os 

3 

4from git.compat import defenc 

5from git.objects import Object 

6from git.objects.commit import Commit 

7from git.util import ( 

8 join_path, 

9 join_path_native, 

10 to_native_path_linux, 

11 assure_directory_exists, 

12 hex_to_bin, 

13 LockedFD, 

14) 

15from gitdb.exc import BadObject, BadName 

16 

17from .log import RefLog 

18 

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

20 

21from typing import ( 

22 Any, 

23 Iterator, 

24 List, 

25 Tuple, 

26 Type, 

27 TypeVar, 

28 Union, 

29 TYPE_CHECKING, 

30 cast, 

31) # NOQA 

32from git.types import Commit_ish, PathLike # NOQA 

33 

34if TYPE_CHECKING: 34 ↛ 35line 34 didn't jump to line 35, because the condition on line 34 was never true

35 from git.repo import Repo 

36 from git.refs import Head, TagReference, RemoteReference, Reference 

37 from .log import RefLogEntry 

38 from git.config import GitConfigParser 

39 from git.objects.commit import Actor 

40 

41 

42T_References = TypeVar("T_References", bound="SymbolicReference") 

43 

44# ------------------------------------------------------------------------------ 

45 

46 

47__all__ = ["SymbolicReference"] 

48 

49 

50def _git_dir(repo: "Repo", path: Union[PathLike, None]) -> PathLike: 

51 """Find the git dir that's appropriate for the path""" 

52 name = f"{path}" 

53 if name in ["HEAD", "ORIG_HEAD", "FETCH_HEAD", "index", "logs"]: 

54 return repo.git_dir 

55 return repo.common_dir 

56 

57 

58class SymbolicReference(object): 

59 

60 """Represents a special case of a reference such that this reference is symbolic. 

61 It does not point to a specific commit, but to another Head, which itself 

62 specifies a commit. 

63 

64 A typical example for a symbolic reference is HEAD.""" 

65 

66 __slots__ = ("repo", "path") 

67 _resolve_ref_on_create = False 

68 _points_to_commits_only = True 

69 _common_path_default = "" 

70 _remote_common_path_default = "refs/remotes" 

71 _id_attribute_ = "name" 

72 

73 def __init__(self, repo: "Repo", path: PathLike, check_path: bool = False): 

74 self.repo = repo 

75 self.path = path 

76 

77 def __str__(self) -> str: 

78 return str(self.path) 

79 

80 def __repr__(self) -> str: 

81 return '<git.%s "%s">' % (self.__class__.__name__, self.path) 

82 

83 def __eq__(self, other: object) -> bool: 

84 if hasattr(other, "path"): 

85 other = cast(SymbolicReference, other) 

86 return self.path == other.path 

87 return False 

88 

89 def __ne__(self, other: object) -> bool: 

90 return not (self == other) 

91 

92 def __hash__(self) -> int: 

93 return hash(self.path) 

94 

95 @property 

96 def name(self) -> str: 

97 """ 

98 :return: 

99 In case of symbolic references, the shortest assumable name 

100 is the path itself.""" 

101 return str(self.path) 

102 

103 @property 

104 def abspath(self) -> PathLike: 

105 return join_path_native(_git_dir(self.repo, self.path), self.path) 

106 

107 @classmethod 

108 def _get_packed_refs_path(cls, repo: "Repo") -> str: 

109 return os.path.join(repo.common_dir, "packed-refs") 

110 

111 @classmethod 

112 def _iter_packed_refs(cls, repo: "Repo") -> Iterator[Tuple[str, str]]: 

113 """Returns an iterator yielding pairs of sha1/path pairs (as strings) for the corresponding refs. 

114 :note: The packed refs file will be kept open as long as we iterate""" 

115 try: 

116 with open(cls._get_packed_refs_path(repo), "rt", encoding="UTF-8") as fp: 

117 for line in fp: 

118 line = line.strip() 

119 if not line: 

120 continue 

121 if line.startswith("#"): 

122 # "# pack-refs with: peeled fully-peeled sorted" 

123 # the git source code shows "peeled", 

124 # "fully-peeled" and "sorted" as the keywords 

125 # that can go on this line, as per comments in git file 

126 # refs/packed-backend.c 

127 # I looked at master on 2017-10-11, 

128 # commit 111ef79afe, after tag v2.15.0-rc1 

129 # from repo https://github.com/git/git.git 

130 if line.startswith("# pack-refs with:") and "peeled" not in line: 

131 raise TypeError("PackingType of packed-Refs not understood: %r" % line) 

132 # END abort if we do not understand the packing scheme 

133 continue 

134 # END parse comment 

135 

136 # skip dereferenced tag object entries - previous line was actual 

137 # tag reference for it 

138 if line[0] == "^": 

139 continue 

140 

141 yield cast(Tuple[str, str], tuple(line.split(" ", 1))) 

142 # END for each line 

143 except OSError: 

144 return None 

145 # END no packed-refs file handling 

146 # NOTE: Had try-finally block around here to close the fp, 

147 # but some python version wouldn't allow yields within that. 

148 # I believe files are closing themselves on destruction, so it is 

149 # alright. 

150 

151 @classmethod 

152 def dereference_recursive(cls, repo: "Repo", ref_path: Union[PathLike, None]) -> str: 

153 """ 

154 :return: hexsha stored in the reference at the given ref_path, recursively dereferencing all 

155 intermediate references as required 

156 :param repo: the repository containing the reference at ref_path""" 

157 

158 while True: 

159 hexsha, ref_path = cls._get_ref_info(repo, ref_path) 

160 if hexsha is not None: 

161 return hexsha 

162 # END recursive dereferencing 

163 

164 @classmethod 

165 def _get_ref_info_helper( 

166 cls, repo: "Repo", ref_path: Union[PathLike, None] 

167 ) -> Union[Tuple[str, None], Tuple[None, str]]: 

168 """Return: (str(sha), str(target_ref_path)) if available, the sha the file at 

169 rela_path points to, or None. target_ref_path is the reference we 

170 point to, or None""" 

171 tokens: Union[None, List[str], Tuple[str, str]] = None 

172 repodir = _git_dir(repo, ref_path) 

173 try: 

174 with open(os.path.join(repodir, str(ref_path)), "rt", encoding="UTF-8") as fp: 

175 value = fp.read().rstrip() 

176 # Don't only split on spaces, but on whitespace, which allows to parse lines like 

177 # 60b64ef992065e2600bfef6187a97f92398a9144 branch 'master' of git-server:/path/to/repo 

178 tokens = value.split() 

179 assert len(tokens) != 0 

180 except OSError: 

181 # Probably we are just packed, find our entry in the packed refs file 

182 # NOTE: We are not a symbolic ref if we are in a packed file, as these 

183 # are excluded explicitly 

184 for sha, path in cls._iter_packed_refs(repo): 

185 if path != ref_path: 

186 continue 

187 # sha will be used 

188 tokens = sha, path 

189 break 

190 # END for each packed ref 

191 # END handle packed refs 

192 if tokens is None: 

193 raise ValueError("Reference at %r does not exist" % ref_path) 

194 

195 # is it a reference ? 

196 if tokens[0] == "ref:": 

197 return (None, tokens[1]) 

198 

199 # its a commit 

200 if repo.re_hexsha_only.match(tokens[0]): 

201 return (tokens[0], None) 

202 

203 raise ValueError("Failed to parse reference information from %r" % ref_path) 

204 

205 @classmethod 

206 def _get_ref_info(cls, repo: "Repo", ref_path: Union[PathLike, None]) -> Union[Tuple[str, None], Tuple[None, str]]: 

207 """Return: (str(sha), str(target_ref_path)) if available, the sha the file at 

208 rela_path points to, or None. target_ref_path is the reference we 

209 point to, or None""" 

210 return cls._get_ref_info_helper(repo, ref_path) 

211 

212 def _get_object(self) -> Commit_ish: 

213 """ 

214 :return: 

215 The object our ref currently refers to. Refs can be cached, they will 

216 always point to the actual object as it gets re-created on each query""" 

217 # have to be dynamic here as we may be a tag which can point to anything 

218 # Our path will be resolved to the hexsha which will be used accordingly 

219 return Object.new_from_sha(self.repo, hex_to_bin(self.dereference_recursive(self.repo, self.path))) 

220 

221 def _get_commit(self) -> "Commit": 

222 """ 

223 :return: 

224 Commit object we point to, works for detached and non-detached 

225 SymbolicReferences. The symbolic reference will be dereferenced recursively.""" 

226 obj = self._get_object() 

227 if obj.type == "tag": 

228 obj = obj.object 

229 # END dereference tag 

230 

231 if obj.type != Commit.type: 

232 raise TypeError("Symbolic Reference pointed to object %r, commit was required" % obj) 

233 # END handle type 

234 return obj 

235 

236 def set_commit( 

237 self, 

238 commit: Union[Commit, "SymbolicReference", str], 

239 logmsg: Union[str, None] = None, 

240 ) -> "SymbolicReference": 

241 """As set_object, but restricts the type of object to be a Commit 

242 

243 :raise ValueError: If commit is not a Commit object or doesn't point to 

244 a commit 

245 :return: self""" 

246 # check the type - assume the best if it is a base-string 

247 invalid_type = False 

248 if isinstance(commit, Object): 

249 invalid_type = commit.type != Commit.type 

250 elif isinstance(commit, SymbolicReference): 

251 invalid_type = commit.object.type != Commit.type 

252 else: 

253 try: 

254 invalid_type = self.repo.rev_parse(commit).type != Commit.type 

255 except (BadObject, BadName) as e: 

256 raise ValueError("Invalid object: %s" % commit) from e 

257 # END handle exception 

258 # END verify type 

259 

260 if invalid_type: 

261 raise ValueError("Need commit, got %r" % commit) 

262 # END handle raise 

263 

264 # we leave strings to the rev-parse method below 

265 self.set_object(commit, logmsg) 

266 

267 return self 

268 

269 def set_object( 

270 self, 

271 object: Union[Commit_ish, "SymbolicReference", str], 

272 logmsg: Union[str, None] = None, 

273 ) -> "SymbolicReference": 

274 """Set the object we point to, possibly dereference our symbolic reference first. 

275 If the reference does not exist, it will be created 

276 

277 :param object: a refspec, a SymbolicReference or an Object instance. SymbolicReferences 

278 will be dereferenced beforehand to obtain the object they point to 

279 :param logmsg: If not None, the message will be used in the reflog entry to be 

280 written. Otherwise the reflog is not altered 

281 :note: plain SymbolicReferences may not actually point to objects by convention 

282 :return: self""" 

283 if isinstance(object, SymbolicReference): 

284 object = object.object # @ReservedAssignment 

285 # END resolve references 

286 

287 is_detached = True 

288 try: 

289 is_detached = self.is_detached 

290 except ValueError: 

291 pass 

292 # END handle non-existing ones 

293 

294 if is_detached: 

295 return self.set_reference(object, logmsg) 

296 

297 # set the commit on our reference 

298 return self._get_reference().set_object(object, logmsg) 

299 

300 commit = property(_get_commit, set_commit, doc="Query or set commits directly") # type: ignore 

301 object = property(_get_object, set_object, doc="Return the object our ref currently refers to") # type: ignore 

302 

303 def _get_reference(self) -> "SymbolicReference": 

304 """:return: Reference Object we point to 

305 :raise TypeError: If this symbolic reference is detached, hence it doesn't point 

306 to a reference, but to a commit""" 

307 sha, target_ref_path = self._get_ref_info(self.repo, self.path) 

308 if target_ref_path is None: 

309 raise TypeError("%s is a detached symbolic reference as it points to %r" % (self, sha)) 

310 return self.from_path(self.repo, target_ref_path) 

311 

312 def set_reference( 

313 self, 

314 ref: Union[Commit_ish, "SymbolicReference", str], 

315 logmsg: Union[str, None] = None, 

316 ) -> "SymbolicReference": 

317 """Set ourselves to the given ref. It will stay a symbol if the ref is a Reference. 

318 Otherwise an Object, given as Object instance or refspec, is assumed and if valid, 

319 will be set which effectively detaches the reference if it was a purely 

320 symbolic one. 

321 

322 :param ref: SymbolicReference instance, Object instance or refspec string 

323 Only if the ref is a SymbolicRef instance, we will point to it. Everything 

324 else is dereferenced to obtain the actual object. 

325 :param logmsg: If set to a string, the message will be used in the reflog. 

326 Otherwise, a reflog entry is not written for the changed reference. 

327 The previous commit of the entry will be the commit we point to now. 

328 

329 See also: log_append() 

330 

331 :return: self 

332 :note: This symbolic reference will not be dereferenced. For that, see 

333 ``set_object(...)``""" 

334 write_value = None 

335 obj = None 

336 if isinstance(ref, SymbolicReference): 

337 write_value = "ref: %s" % ref.path 

338 elif isinstance(ref, Object): 

339 obj = ref 

340 write_value = ref.hexsha 

341 elif isinstance(ref, str): 

342 try: 

343 obj = self.repo.rev_parse(ref + "^{}") # optionally deref tags 

344 write_value = obj.hexsha 

345 except (BadObject, BadName) as e: 

346 raise ValueError("Could not extract object from %s" % ref) from e 

347 # END end try string 

348 else: 

349 raise ValueError("Unrecognized Value: %r" % ref) 

350 # END try commit attribute 

351 

352 # typecheck 

353 if obj is not None and self._points_to_commits_only and obj.type != Commit.type: 

354 raise TypeError("Require commit, got %r" % obj) 

355 # END verify type 

356 

357 oldbinsha: bytes = b"" 

358 if logmsg is not None: 

359 try: 

360 oldbinsha = self.commit.binsha 

361 except ValueError: 

362 oldbinsha = Commit.NULL_BIN_SHA 

363 # END handle non-existing 

364 # END retrieve old hexsha 

365 

366 fpath = self.abspath 

367 assure_directory_exists(fpath, is_file=True) 

368 

369 lfd = LockedFD(fpath) 

370 fd = lfd.open(write=True, stream=True) 

371 ok = True 

372 try: 

373 fd.write(write_value.encode("utf-8") + b"\n") 

374 lfd.commit() 

375 ok = True 

376 finally: 

377 if not ok: 

378 lfd.rollback() 

379 # Adjust the reflog 

380 if logmsg is not None: 

381 self.log_append(oldbinsha, logmsg) 

382 

383 return self 

384 

385 # aliased reference 

386 reference: Union["Head", "TagReference", "RemoteReference", "Reference"] 

387 reference = property(_get_reference, set_reference, doc="Returns the Reference we point to") # type: ignore 

388 ref = reference 

389 

390 def is_valid(self) -> bool: 

391 """ 

392 :return: 

393 True if the reference is valid, hence it can be read and points to 

394 a valid object or reference.""" 

395 try: 

396 self.object 

397 except (OSError, ValueError): 

398 return False 

399 else: 

400 return True 

401 

402 @property 

403 def is_detached(self) -> bool: 

404 """ 

405 :return: 

406 True if we are a detached reference, hence we point to a specific commit 

407 instead to another reference""" 

408 try: 

409 self.ref 

410 return False 

411 except TypeError: 

412 return True 

413 

414 def log(self) -> "RefLog": 

415 """ 

416 :return: RefLog for this reference. Its last entry reflects the latest change 

417 applied to this reference 

418 

419 .. note:: As the log is parsed every time, its recommended to cache it for use 

420 instead of calling this method repeatedly. It should be considered read-only.""" 

421 return RefLog.from_file(RefLog.path(self)) 

422 

423 def log_append( 

424 self, 

425 oldbinsha: bytes, 

426 message: Union[str, None], 

427 newbinsha: Union[bytes, None] = None, 

428 ) -> "RefLogEntry": 

429 """Append a logentry to the logfile of this ref 

430 

431 :param oldbinsha: binary sha this ref used to point to 

432 :param message: A message describing the change 

433 :param newbinsha: The sha the ref points to now. If None, our current commit sha 

434 will be used 

435 :return: added RefLogEntry instance""" 

436 # NOTE: we use the committer of the currently active commit - this should be 

437 # correct to allow overriding the committer on a per-commit level. 

438 # See https://github.com/gitpython-developers/GitPython/pull/146 

439 try: 

440 committer_or_reader: Union["Actor", "GitConfigParser"] = self.commit.committer 

441 except ValueError: 

442 committer_or_reader = self.repo.config_reader() 

443 # end handle newly cloned repositories 

444 if newbinsha is None: 

445 newbinsha = self.commit.binsha 

446 

447 if message is None: 

448 message = "" 

449 

450 return RefLog.append_entry(committer_or_reader, RefLog.path(self), oldbinsha, newbinsha, message) 

451 

452 def log_entry(self, index: int) -> "RefLogEntry": 

453 """:return: RefLogEntry at the given index 

454 :param index: python list compatible positive or negative index 

455 

456 .. note:: This method must read part of the reflog during execution, hence 

457 it should be used sparringly, or only if you need just one index. 

458 In that case, it will be faster than the ``log()`` method""" 

459 return RefLog.entry_at(RefLog.path(self), index) 

460 

461 @classmethod 

462 def to_full_path(cls, path: Union[PathLike, "SymbolicReference"]) -> PathLike: 

463 """ 

464 :return: string with a full repository-relative path which can be used to initialize 

465 a Reference instance, for instance by using ``Reference.from_path``""" 

466 if isinstance(path, SymbolicReference): 

467 path = path.path 

468 full_ref_path = path 

469 if not cls._common_path_default: 

470 return full_ref_path 

471 if not str(path).startswith(cls._common_path_default + "/"): 

472 full_ref_path = "%s/%s" % (cls._common_path_default, path) 

473 return full_ref_path 

474 

475 @classmethod 

476 def delete(cls, repo: "Repo", path: PathLike) -> None: 

477 """Delete the reference at the given path 

478 

479 :param repo: 

480 Repository to delete the reference from 

481 

482 :param path: 

483 Short or full path pointing to the reference, i.e. refs/myreference 

484 or just "myreference", hence 'refs/' is implied. 

485 Alternatively the symbolic reference to be deleted""" 

486 full_ref_path = cls.to_full_path(path) 

487 abs_path = os.path.join(repo.common_dir, full_ref_path) 

488 if os.path.exists(abs_path): 

489 os.remove(abs_path) 

490 else: 

491 # check packed refs 

492 pack_file_path = cls._get_packed_refs_path(repo) 

493 try: 

494 with open(pack_file_path, "rb") as reader: 

495 new_lines = [] 

496 made_change = False 

497 dropped_last_line = False 

498 for line_bytes in reader: 

499 line = line_bytes.decode(defenc) 

500 _, _, line_ref = line.partition(" ") 

501 line_ref = line_ref.strip() 

502 # keep line if it is a comment or if the ref to delete is not 

503 # in the line 

504 # If we deleted the last line and this one is a tag-reference object, 

505 # we drop it as well 

506 if (line.startswith("#") or full_ref_path != line_ref) and ( 

507 not dropped_last_line or dropped_last_line and not line.startswith("^") 

508 ): 

509 new_lines.append(line) 

510 dropped_last_line = False 

511 continue 

512 # END skip comments and lines without our path 

513 

514 # drop this line 

515 made_change = True 

516 dropped_last_line = True 

517 

518 # write the new lines 

519 if made_change: 

520 # write-binary is required, otherwise windows will 

521 # open the file in text mode and change LF to CRLF ! 

522 with open(pack_file_path, "wb") as fd: 

523 fd.writelines(line.encode(defenc) for line in new_lines) 

524 

525 except OSError: 

526 pass # it didn't exist at all 

527 

528 # delete the reflog 

529 reflog_path = RefLog.path(cls(repo, full_ref_path)) 

530 if os.path.isfile(reflog_path): 

531 os.remove(reflog_path) 

532 # END remove reflog 

533 

534 @classmethod 

535 def _create( 

536 cls: Type[T_References], 

537 repo: "Repo", 

538 path: PathLike, 

539 resolve: bool, 

540 reference: Union["SymbolicReference", str], 

541 force: bool, 

542 logmsg: Union[str, None] = None, 

543 ) -> T_References: 

544 """internal method used to create a new symbolic reference. 

545 If resolve is False, the reference will be taken as is, creating 

546 a proper symbolic reference. Otherwise it will be resolved to the 

547 corresponding object and a detached symbolic reference will be created 

548 instead""" 

549 git_dir = _git_dir(repo, path) 

550 full_ref_path = cls.to_full_path(path) 

551 abs_ref_path = os.path.join(git_dir, full_ref_path) 

552 

553 # figure out target data 

554 target = reference 

555 if resolve: 

556 target = repo.rev_parse(str(reference)) 

557 

558 if not force and os.path.isfile(abs_ref_path): 

559 target_data = str(target) 

560 if isinstance(target, SymbolicReference): 

561 target_data = str(target.path) 

562 if not resolve: 

563 target_data = "ref: " + target_data 

564 with open(abs_ref_path, "rb") as fd: 

565 existing_data = fd.read().decode(defenc).strip() 

566 if existing_data != target_data: 

567 raise OSError( 

568 "Reference at %r does already exist, pointing to %r, requested was %r" 

569 % (full_ref_path, existing_data, target_data) 

570 ) 

571 # END no force handling 

572 

573 ref = cls(repo, full_ref_path) 

574 ref.set_reference(target, logmsg) 

575 return ref 

576 

577 @classmethod 

578 def create( 

579 cls: Type[T_References], 

580 repo: "Repo", 

581 path: PathLike, 

582 reference: Union["SymbolicReference", str] = "HEAD", 

583 logmsg: Union[str, None] = None, 

584 force: bool = False, 

585 **kwargs: Any, 

586 ) -> T_References: 

587 """Create a new symbolic reference, hence a reference pointing , to another reference. 

588 

589 :param repo: 

590 Repository to create the reference in 

591 

592 :param path: 

593 full path at which the new symbolic reference is supposed to be 

594 created at, i.e. "NEW_HEAD" or "symrefs/my_new_symref" 

595 

596 :param reference: 

597 The reference to which the new symbolic reference should point to. 

598 If it is a commit'ish, the symbolic ref will be detached. 

599 

600 :param force: 

601 if True, force creation even if a symbolic reference with that name already exists. 

602 Raise OSError otherwise 

603 

604 :param logmsg: 

605 If not None, the message to append to the reflog. Otherwise no reflog 

606 entry is written. 

607 

608 :return: Newly created symbolic Reference 

609 

610 :raise OSError: 

611 If a (Symbolic)Reference with the same name but different contents 

612 already exists. 

613 

614 :note: This does not alter the current HEAD, index or Working Tree""" 

615 return cls._create(repo, path, cls._resolve_ref_on_create, reference, force, logmsg) 

616 

617 def rename(self, new_path: PathLike, force: bool = False) -> "SymbolicReference": 

618 """Rename self to a new path 

619 

620 :param new_path: 

621 Either a simple name or a full path, i.e. new_name or features/new_name. 

622 The prefix refs/ is implied for references and will be set as needed. 

623 In case this is a symbolic ref, there is no implied prefix 

624 

625 :param force: 

626 If True, the rename will succeed even if a head with the target name 

627 already exists. It will be overwritten in that case 

628 

629 :return: self 

630 :raise OSError: In case a file at path but a different contents already exists""" 

631 new_path = self.to_full_path(new_path) 

632 if self.path == new_path: 

633 return self 

634 

635 new_abs_path = os.path.join(_git_dir(self.repo, new_path), new_path) 

636 cur_abs_path = os.path.join(_git_dir(self.repo, self.path), self.path) 

637 if os.path.isfile(new_abs_path): 

638 if not force: 

639 # if they point to the same file, its not an error 

640 with open(new_abs_path, "rb") as fd1: 

641 f1 = fd1.read().strip() 

642 with open(cur_abs_path, "rb") as fd2: 

643 f2 = fd2.read().strip() 

644 if f1 != f2: 

645 raise OSError("File at path %r already exists" % new_abs_path) 

646 # else: we could remove ourselves and use the otherone, but 

647 # but clarity we just continue as usual 

648 # END not force handling 

649 os.remove(new_abs_path) 

650 # END handle existing target file 

651 

652 dname = os.path.dirname(new_abs_path) 

653 if not os.path.isdir(dname): 

654 os.makedirs(dname) 

655 # END create directory 

656 

657 os.rename(cur_abs_path, new_abs_path) 

658 self.path = new_path 

659 

660 return self 

661 

662 @classmethod 

663 def _iter_items( 

664 cls: Type[T_References], repo: "Repo", common_path: Union[PathLike, None] = None 

665 ) -> Iterator[T_References]: 

666 if common_path is None: 

667 common_path = cls._common_path_default 

668 rela_paths = set() 

669 

670 # walk loose refs 

671 # Currently we do not follow links 

672 for root, dirs, files in os.walk(join_path_native(repo.common_dir, common_path)): 

673 if "refs" not in root.split(os.sep): # skip non-refs subfolders 

674 refs_id = [d for d in dirs if d == "refs"] 

675 if refs_id: 

676 dirs[0:] = ["refs"] 

677 # END prune non-refs folders 

678 

679 for f in files: 

680 if f == "packed-refs": 

681 continue 

682 abs_path = to_native_path_linux(join_path(root, f)) 

683 rela_paths.add(abs_path.replace(to_native_path_linux(repo.common_dir) + "/", "")) 

684 # END for each file in root directory 

685 # END for each directory to walk 

686 

687 # read packed refs 

688 for _sha, rela_path in cls._iter_packed_refs(repo): 

689 if rela_path.startswith(str(common_path)): 

690 rela_paths.add(rela_path) 

691 # END relative path matches common path 

692 # END packed refs reading 

693 

694 # return paths in sorted order 

695 for path in sorted(rela_paths): 

696 try: 

697 yield cls.from_path(repo, path) 

698 except ValueError: 

699 continue 

700 # END for each sorted relative refpath 

701 

702 @classmethod 

703 def iter_items( 

704 cls: Type[T_References], 

705 repo: "Repo", 

706 common_path: Union[PathLike, None] = None, 

707 *args: Any, 

708 **kwargs: Any, 

709 ) -> Iterator[T_References]: 

710 """Find all refs in the repository 

711 

712 :param repo: is the Repo 

713 

714 :param common_path: 

715 Optional keyword argument to the path which is to be shared by all 

716 returned Ref objects. 

717 Defaults to class specific portion if None assuring that only 

718 refs suitable for the actual class are returned. 

719 

720 :return: 

721 git.SymbolicReference[], each of them is guaranteed to be a symbolic 

722 ref which is not detached and pointing to a valid ref 

723 

724 List is lexicographically sorted 

725 The returned objects represent actual subclasses, such as Head or TagReference""" 

726 return (r for r in cls._iter_items(repo, common_path) if r.__class__ == SymbolicReference or not r.is_detached) 

727 

728 @classmethod 

729 def from_path(cls: Type[T_References], repo: "Repo", path: PathLike) -> T_References: 

730 """ 

731 :param path: full .git-directory-relative path name to the Reference to instantiate 

732 :note: use to_full_path() if you only have a partial path of a known Reference Type 

733 :return: 

734 Instance of type Reference, Head, or Tag 

735 depending on the given path""" 

736 if not path: 

737 raise ValueError("Cannot create Reference from %r" % path) 

738 

739 # Names like HEAD are inserted after the refs module is imported - we have an import dependency 

740 # cycle and don't want to import these names in-function 

741 from . import HEAD, Head, RemoteReference, TagReference, Reference 

742 

743 for ref_type in ( 

744 HEAD, 

745 Head, 

746 RemoteReference, 

747 TagReference, 

748 Reference, 

749 SymbolicReference, 

750 ): 

751 try: 

752 instance: T_References 

753 instance = ref_type(repo, path) 

754 if instance.__class__ == SymbolicReference and instance.is_detached: 

755 raise ValueError("SymbolRef was detached, we drop it") 

756 else: 

757 return instance 

758 

759 except ValueError: 

760 pass 

761 # END exception handling 

762 # END for each type to try 

763 raise ValueError("Could not find reference type suitable to handle path %r" % path) 

764 

765 def is_remote(self) -> bool: 

766 """:return: True if this symbolic reference points to a remote branch""" 

767 return str(self.path).startswith(self._remote_common_path_default + "/")