Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/git/refs/symbolic.py: 58%

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

385 statements  

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

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

3 

4__all__ = ["SymbolicReference"] 

5 

6import os 

7from pathlib import Path 

8 

9from gitdb.exc import BadName, BadObject 

10 

11from git.compat import defenc 

12from git.objects.base import Object 

13from git.objects.commit import Commit 

14from git.refs.log import RefLog 

15from git.util import ( 

16 LockedFD, 

17 assure_directory_exists, 

18 hex_to_bin, 

19 join_path, 

20 join_path_native, 

21 to_native_path_linux, 

22) 

23 

24# typing ------------------------------------------------------------------ 

25 

26from typing import ( 

27 Any, 

28 Iterator, 

29 List, 

30 TYPE_CHECKING, 

31 Tuple, 

32 Type, 

33 TypeVar, 

34 Union, 

35 cast, 

36) 

37 

38from git.types import AnyGitObject, PathLike 

39 

40if TYPE_CHECKING: 

41 from git.config import GitConfigParser 

42 from git.objects.commit import Actor 

43 from git.refs.log import RefLogEntry 

44 from git.refs.reference import Reference 

45 from git.repo import Repo 

46 

47 

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

49 

50# ------------------------------------------------------------------------------ 

51 

52 

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

54 """Find the git dir that is appropriate for the path.""" 

55 name = f"{path}" 

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

57 return repo.git_dir 

58 return repo.common_dir 

59 

60 

61class SymbolicReference: 

62 """Special case of a reference that is symbolic. 

63 

64 This does not point to a specific commit, but to another 

65 :class:`~git.refs.head.Head`, which itself specifies a commit. 

66 

67 A typical example for a symbolic reference is :class:`~git.refs.head.HEAD`. 

68 """ 

69 

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

71 

72 _resolve_ref_on_create = False 

73 _points_to_commits_only = True 

74 _common_path_default = "" 

75 _remote_common_path_default = "refs/remotes" 

76 _id_attribute_ = "name" 

77 

78 def __init__(self, repo: "Repo", path: PathLike, check_path: bool = False) -> None: 

79 self.repo = repo 

80 self.path: PathLike = path 

81 

82 def __str__(self) -> str: 

83 return os.fspath(self.path) 

84 

85 def __repr__(self) -> str: 

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

87 

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

89 if hasattr(other, "path"): 

90 other = cast(SymbolicReference, other) 

91 return self.path == other.path 

92 return False 

93 

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

95 return not (self == other) 

96 

97 def __hash__(self) -> int: 

98 return hash(self.path) 

99 

100 @property 

101 def name(self) -> str: 

102 """ 

103 :return: 

104 In case of symbolic references, the shortest assumable name is the path 

105 itself. 

106 """ 

107 return os.fspath(self.path) 

108 

109 @property 

110 def abspath(self) -> PathLike: 

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

112 

113 @classmethod 

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

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

116 

117 @classmethod 

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

119 """Return an iterator yielding pairs of sha1/path pairs (as strings) for the 

120 corresponding refs. 

121 

122 :note: 

123 The packed refs file will be kept open as long as we iterate. 

124 """ 

125 try: 

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

127 for line in fp: 

128 line = line.strip() 

129 if not line: 

130 continue 

131 if line.startswith("#"): 

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

133 # the git source code shows "peeled", 

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

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

136 # refs/packed-backend.c 

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

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

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

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

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

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

143 continue 

144 # END parse comment 

145 

146 # Skip dereferenced tag object entries - previous line was actual 

147 # tag reference for it. 

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

149 continue 

150 

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

152 # END for each line 

153 except OSError: 

154 return None 

155 # END no packed-refs file handling 

156 

157 @classmethod 

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

159 """ 

160 :return: 

161 hexsha stored in the reference at the given `ref_path`, recursively 

162 dereferencing all intermediate references as required 

163 

164 :param repo: 

165 The repository containing the reference at `ref_path`. 

166 """ 

167 

168 while True: 

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

170 if hexsha is not None: 

171 return hexsha 

172 # END recursive dereferencing 

173 

174 @staticmethod 

175 def _check_ref_name_valid(ref_path: PathLike) -> None: 

176 """Check a ref name for validity. 

177 

178 This is based on the rules described in :manpage:`git-check-ref-format(1)`. 

179 """ 

180 previous: Union[str, None] = None 

181 one_before_previous: Union[str, None] = None 

182 for c in os.fspath(ref_path): 

183 if c in " ~^:?*[\\": 

184 raise ValueError( 

185 f"Invalid reference '{ref_path}': references cannot contain spaces, tildes (~), carets (^)," 

186 f" colons (:), question marks (?), asterisks (*), open brackets ([) or backslashes (\\)" 

187 ) 

188 elif c == ".": 

189 if previous is None or previous == "/": 

190 raise ValueError( 

191 f"Invalid reference '{ref_path}': references cannot start with a period (.) or contain '/.'" 

192 ) 

193 elif previous == ".": 

194 raise ValueError(f"Invalid reference '{ref_path}': references cannot contain '..'") 

195 elif c == "/": 

196 if previous == "/": 

197 raise ValueError(f"Invalid reference '{ref_path}': references cannot contain '//'") 

198 elif previous is None: 

199 raise ValueError( 

200 f"Invalid reference '{ref_path}': references cannot start with forward slashes '/'" 

201 ) 

202 elif c == "{" and previous == "@": 

203 raise ValueError(f"Invalid reference '{ref_path}': references cannot contain '@{{'") 

204 elif ord(c) < 32 or ord(c) == 127: 

205 raise ValueError(f"Invalid reference '{ref_path}': references cannot contain ASCII control characters") 

206 

207 one_before_previous = previous 

208 previous = c 

209 

210 if previous == ".": 

211 raise ValueError(f"Invalid reference '{ref_path}': references cannot end with a period (.)") 

212 elif previous == "/": 

213 raise ValueError(f"Invalid reference '{ref_path}': references cannot end with a forward slash (/)") 

214 elif previous == "@" and one_before_previous is None: 

215 raise ValueError(f"Invalid reference '{ref_path}': references cannot be '@'") 

216 elif any(component.endswith(".lock") for component in Path(ref_path).parts): 

217 raise ValueError( 

218 f"Invalid reference '{ref_path}': references cannot have slash-separated components that end with" 

219 " '.lock'" 

220 ) 

221 

222 @classmethod 

223 def _get_ref_info_helper( 

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

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

226 """ 

227 :return: 

228 *(str(sha), str(target_ref_path))*, where: 

229 

230 * *sha* is of the file at rela_path points to if available, or ``None``. 

231 * *target_ref_path* is the reference we point to, or ``None``. 

232 """ 

233 if ref_path: 

234 cls._check_ref_name_valid(ref_path) 

235 

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

237 repodir = _git_dir(repo, ref_path) 

238 try: 

239 with open(os.path.join(repodir, ref_path), "rt", encoding="UTF-8") as fp: # type: ignore[arg-type] 

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

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

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

243 tokens = value.split() 

244 assert len(tokens) != 0 

245 except OSError: 

246 # Probably we are just packed. Find our entry in the packed refs file. 

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

248 # are excluded explicitly. 

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

250 if path != ref_path: 

251 continue 

252 # sha will be used. 

253 tokens = sha, path 

254 break 

255 # END for each packed ref 

256 # END handle packed refs 

257 if tokens is None: 

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

259 

260 # Is it a reference? 

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

262 return (None, tokens[1]) 

263 

264 # It's a commit. 

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

266 return (tokens[0], None) 

267 

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

269 

270 @classmethod 

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

272 """ 

273 :return: 

274 *(str(sha), str(target_ref_path))*, where: 

275 

276 * *sha* is of the file at rela_path points to if available, or ``None``. 

277 * *target_ref_path* is the reference we point to, or ``None``. 

278 """ 

279 return cls._get_ref_info_helper(repo, ref_path) 

280 

281 def _get_object(self) -> AnyGitObject: 

282 """ 

283 :return: 

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

285 point to the actual object as it gets re-created on each query. 

286 """ 

287 # We have to be dynamic here as we may be a tag which can point to anything. 

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

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

290 

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

292 """ 

293 :return: 

294 :class:`~git.objects.commit.Commit` object we point to. This works for 

295 detached and non-detached :class:`SymbolicReference` instances. The symbolic 

296 reference will be dereferenced recursively. 

297 """ 

298 obj = self._get_object() 

299 if obj.type == "tag": 

300 obj = obj.object 

301 # END dereference tag 

302 

303 if obj.type != Commit.type: 

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

305 # END handle type 

306 return obj 

307 

308 def set_commit( 

309 self, 

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

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

312 ) -> "SymbolicReference": 

313 """Like :meth:`set_object`, but restricts the type of object to be a 

314 :class:`~git.objects.commit.Commit`. 

315 

316 :raise ValueError: 

317 If `commit` is not a :class:`~git.objects.commit.Commit` object, nor does it 

318 point to a commit. 

319 

320 :return: 

321 self 

322 """ 

323 # Check the type - assume the best if it is a base-string. 

324 invalid_type = False 

325 if isinstance(commit, Object): 

326 invalid_type = commit.type != Commit.type 

327 elif isinstance(commit, SymbolicReference): 

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

329 else: 

330 try: 

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

332 except (BadObject, BadName) as e: 

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

334 # END handle exception 

335 # END verify type 

336 

337 if invalid_type: 

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

339 # END handle raise 

340 

341 # We leave strings to the rev-parse method below. 

342 self.set_object(commit, logmsg) 

343 

344 return self 

345 

346 def set_object( 

347 self, 

348 object: Union[AnyGitObject, "SymbolicReference", str], 

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

350 ) -> "SymbolicReference": 

351 """Set the object we point to, possibly dereference our symbolic reference 

352 first. If the reference does not exist, it will be created. 

353 

354 :param object: 

355 A refspec, a :class:`SymbolicReference` or an 

356 :class:`~git.objects.base.Object` instance. 

357 

358 * :class:`SymbolicReference` instances will be dereferenced beforehand to 

359 obtain the git object they point to. 

360 * :class:`~git.objects.base.Object` instances must represent git objects 

361 (:class:`~git.types.AnyGitObject`). 

362 

363 :param logmsg: 

364 If not ``None``, the message will be used in the reflog entry to be written. 

365 Otherwise the reflog is not altered. 

366 

367 :note: 

368 Plain :class:`SymbolicReference` instances may not actually point to objects 

369 by convention. 

370 

371 :return: 

372 self 

373 """ 

374 if isinstance(object, SymbolicReference): 

375 object = object.object # @ReservedAssignment 

376 # END resolve references 

377 

378 is_detached = True 

379 try: 

380 is_detached = self.is_detached 

381 except ValueError: 

382 pass 

383 # END handle non-existing ones 

384 

385 if is_detached: 

386 return self.set_reference(object, logmsg) 

387 

388 # set the commit on our reference 

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

390 

391 @property 

392 def commit(self) -> "Commit": 

393 """Query or set commits directly""" 

394 return self._get_commit() 

395 

396 @commit.setter 

397 def commit(self, commit: Union[Commit, "SymbolicReference", str]) -> "SymbolicReference": 

398 return self.set_commit(commit) 

399 

400 @property 

401 def object(self) -> AnyGitObject: 

402 """Return the object our ref currently refers to""" 

403 return self._get_object() 

404 

405 @object.setter 

406 def object(self, object: Union[AnyGitObject, "SymbolicReference", str]) -> "SymbolicReference": 

407 return self.set_object(object) 

408 

409 def _get_reference(self) -> "Reference": 

410 """ 

411 :return: 

412 :class:`~git.refs.reference.Reference` object we point to 

413 

414 :raise TypeError: 

415 If this symbolic reference is detached, hence it doesn't point to a 

416 reference, but to a commit. 

417 """ 

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

419 if target_ref_path is None: 

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

421 return cast("Reference", self.from_path(self.repo, target_ref_path)) 

422 

423 def set_reference( 

424 self, 

425 ref: Union[AnyGitObject, "SymbolicReference", str], 

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

427 ) -> "SymbolicReference": 

428 """Set ourselves to the given `ref`. 

429 

430 It will stay a symbol if the `ref` is a :class:`~git.refs.reference.Reference`. 

431 

432 Otherwise a git object, specified as a :class:`~git.objects.base.Object` 

433 instance or refspec, is assumed. If it is valid, this reference will be set to 

434 it, which effectively detaches the reference if it was a purely symbolic one. 

435 

436 :param ref: 

437 A :class:`SymbolicReference` instance, an :class:`~git.objects.base.Object` 

438 instance (specifically an :class:`~git.types.AnyGitObject`), or a refspec 

439 string. Only if the ref is a :class:`SymbolicReference` instance, we will 

440 point to it. Everything else is dereferenced to obtain the actual object. 

441 

442 :param logmsg: 

443 If set to a string, the message will be used in the reflog. 

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

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

446 

447 See also: :meth:`log_append` 

448 

449 :return: 

450 self 

451 

452 :note: 

453 This symbolic reference will not be dereferenced. For that, see 

454 :meth:`set_object`. 

455 """ 

456 write_value = None 

457 obj = None 

458 if isinstance(ref, SymbolicReference): 

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

460 elif isinstance(ref, Object): 

461 obj = ref 

462 write_value = ref.hexsha 

463 elif isinstance(ref, str): 

464 try: 

465 obj = self.repo.rev_parse(ref + "^{}") # Optionally dereference tags. 

466 write_value = obj.hexsha 

467 except (BadObject, BadName) as e: 

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

469 # END end try string 

470 else: 

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

472 # END try commit attribute 

473 

474 # typecheck 

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

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

477 # END verify type 

478 

479 oldbinsha: bytes = b"" 

480 if logmsg is not None: 

481 try: 

482 oldbinsha = self.commit.binsha 

483 except ValueError: 

484 oldbinsha = Commit.NULL_BIN_SHA 

485 # END handle non-existing 

486 # END retrieve old hexsha 

487 

488 fpath = self.abspath 

489 assure_directory_exists(fpath, is_file=True) 

490 

491 lfd = LockedFD(fpath) 

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

493 try: 

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

495 lfd.commit() 

496 except BaseException: 

497 lfd.rollback() 

498 raise 

499 # Adjust the reflog 

500 if logmsg is not None: 

501 self.log_append(oldbinsha, logmsg) 

502 

503 return self 

504 

505 # Aliased reference 

506 @property 

507 def reference(self) -> "Reference": 

508 return self._get_reference() 

509 

510 @reference.setter 

511 def reference(self, ref: Union[AnyGitObject, "SymbolicReference", str]) -> "SymbolicReference": 

512 return self.set_reference(ref) 

513 

514 ref = reference 

515 

516 def is_valid(self) -> bool: 

517 """ 

518 :return: 

519 ``True`` if the reference is valid, hence it can be read and points to a 

520 valid object or reference. 

521 """ 

522 try: 

523 self.object # noqa: B018 

524 except (OSError, ValueError): 

525 return False 

526 else: 

527 return True 

528 

529 @property 

530 def is_detached(self) -> bool: 

531 """ 

532 :return: 

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

534 instead to another reference. 

535 """ 

536 try: 

537 self.ref # noqa: B018 

538 return False 

539 except TypeError: 

540 return True 

541 

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

543 """ 

544 :return: 

545 :class:`~git.refs.log.RefLog` for this reference. 

546 Its last entry reflects the latest change applied to this reference. 

547 

548 :note: 

549 As the log is parsed every time, its recommended to cache it for use instead 

550 of calling this method repeatedly. It should be considered read-only. 

551 """ 

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

553 

554 def log_append( 

555 self, 

556 oldbinsha: bytes, 

557 message: Union[str, None], 

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

559 ) -> "RefLogEntry": 

560 """Append a logentry to the logfile of this ref. 

561 

562 :param oldbinsha: 

563 Binary sha this ref used to point to. 

564 

565 :param message: 

566 A message describing the change. 

567 

568 :param newbinsha: 

569 The sha the ref points to now. If None, our current commit sha will be used. 

570 

571 :return: 

572 The added :class:`~git.refs.log.RefLogEntry` instance. 

573 """ 

574 # NOTE: We use the committer of the currently active commit - this should be 

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

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

577 try: 

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

579 except ValueError: 

580 committer_or_reader = self.repo.config_reader() 

581 # END handle newly cloned repositories 

582 if newbinsha is None: 

583 newbinsha = self.commit.binsha 

584 

585 if message is None: 

586 message = "" 

587 

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

589 

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

591 """ 

592 :return: 

593 :class:`~git.refs.log.RefLogEntry` at the given index 

594 

595 :param index: 

596 Python list compatible positive or negative index. 

597 

598 :note: 

599 This method must read part of the reflog during execution, hence it should 

600 be used sparingly, or only if you need just one index. In that case, it will 

601 be faster than the :meth:`log` method. 

602 """ 

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

604 

605 @classmethod 

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

607 """ 

608 :return: 

609 String with a full repository-relative path which can be used to initialize 

610 a :class:`~git.refs.reference.Reference` instance, for instance by using 

611 :meth:`Reference.from_path <git.refs.reference.Reference.from_path>`. 

612 """ 

613 if isinstance(path, SymbolicReference): 

614 path = path.path 

615 full_ref_path = path 

616 if not cls._common_path_default: 

617 return full_ref_path 

618 if not os.fspath(path).startswith(cls._common_path_default + "/"): 

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

620 return full_ref_path 

621 

622 @classmethod 

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

624 """Delete the reference at the given path. 

625 

626 :param repo: 

627 Repository to delete the reference from. 

628 

629 :param path: 

630 Short or full path pointing to the reference, e.g. ``refs/myreference`` or 

631 just ``myreference``, hence ``refs/`` is implied. 

632 Alternatively the symbolic reference to be deleted. 

633 """ 

634 full_ref_path = cls.to_full_path(path) 

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

636 if os.path.exists(abs_path): 

637 os.remove(abs_path) 

638 else: 

639 # Check packed refs. 

640 pack_file_path = cls._get_packed_refs_path(repo) 

641 try: 

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

643 new_lines = [] 

644 made_change = False 

645 dropped_last_line = False 

646 for line_bytes in reader: 

647 line = line_bytes.decode(defenc) 

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

649 line_ref = line_ref.strip() 

650 # Keep line if it is a comment or if the ref to delete is not in 

651 # the line. 

652 # If we deleted the last line and this one is a tag-reference 

653 # object, we drop it as well. 

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

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

656 ): 

657 new_lines.append(line) 

658 dropped_last_line = False 

659 continue 

660 # END skip comments and lines without our path 

661 

662 # Drop this line. 

663 made_change = True 

664 dropped_last_line = True 

665 

666 # Write the new lines. 

667 if made_change: 

668 # Binary writing is required, otherwise Windows will open the file 

669 # in text mode and change LF to CRLF! 

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

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

672 

673 except OSError: 

674 pass # It didn't exist at all. 

675 

676 # Delete the reflog. 

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

678 if os.path.isfile(reflog_path): 

679 os.remove(reflog_path) 

680 # END remove reflog 

681 

682 @classmethod 

683 def _create( 

684 cls: Type[T_References], 

685 repo: "Repo", 

686 path: PathLike, 

687 resolve: bool, 

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

689 force: bool, 

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

691 ) -> T_References: 

692 """Internal method used to create a new symbolic reference. 

693 

694 If `resolve` is ``False``, the reference will be taken as is, creating a proper 

695 symbolic reference. Otherwise it will be resolved to the corresponding object 

696 and a detached symbolic reference will be created instead. 

697 """ 

698 git_dir = _git_dir(repo, path) 

699 full_ref_path = cls.to_full_path(path) 

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

701 

702 # Figure out target data. 

703 target = reference 

704 if resolve: 

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

706 

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

708 target_data = str(target) 

709 if isinstance(target, SymbolicReference): 

710 target_data = os.fspath(target.path) 

711 if not resolve: 

712 target_data = "ref: " + target_data 

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

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

715 if existing_data != target_data: 

716 raise OSError( 

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

718 % (full_ref_path, existing_data, target_data) 

719 ) 

720 # END no force handling 

721 

722 ref = cls(repo, full_ref_path) 

723 ref.set_reference(target, logmsg) 

724 return ref 

725 

726 @classmethod 

727 def create( 

728 cls: Type[T_References], 

729 repo: "Repo", 

730 path: PathLike, 

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

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

733 force: bool = False, 

734 **kwargs: Any, 

735 ) -> T_References: 

736 """Create a new symbolic reference: a reference pointing to another reference. 

737 

738 :param repo: 

739 Repository to create the reference in. 

740 

741 :param path: 

742 Full path at which the new symbolic reference is supposed to be created at, 

743 e.g. ``NEW_HEAD`` or ``symrefs/my_new_symref``. 

744 

745 :param reference: 

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

747 If it is a commit-ish, the symbolic ref will be detached. 

748 

749 :param force: 

750 If ``True``, force creation even if a symbolic reference with that name 

751 already exists. Raise :exc:`OSError` otherwise. 

752 

753 :param logmsg: 

754 If not ``None``, the message to append to the reflog. 

755 If ``None``, no reflog entry is written. 

756 

757 :return: 

758 Newly created symbolic reference 

759 

760 :raise OSError: 

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

762 exists. 

763 

764 :note: 

765 This does not alter the current HEAD, index or working tree. 

766 """ 

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

768 

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

770 """Rename self to a new path. 

771 

772 :param new_path: 

773 Either a simple name or a full path, e.g. ``new_name`` or 

774 ``features/new_name``. 

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

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

777 

778 :param force: 

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

780 already exists. It will be overwritten in that case. 

781 

782 :return: 

783 self 

784 

785 :raise OSError: 

786 If a file at path but with different contents already exists. 

787 """ 

788 new_path = self.to_full_path(new_path) 

789 if self.path == new_path: 

790 return self 

791 

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

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

794 if os.path.isfile(new_abs_path): 

795 if not force: 

796 # If they point to the same file, it's not an error. 

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

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

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

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

801 if f1 != f2: 

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

803 # else: We could remove ourselves and use the other one, but... 

804 # ...for clarity, we just continue as usual. 

805 # END not force handling 

806 os.remove(new_abs_path) 

807 # END handle existing target file 

808 

809 dname = os.path.dirname(new_abs_path) 

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

811 os.makedirs(dname) 

812 # END create directory 

813 

814 os.rename(cur_abs_path, new_abs_path) 

815 self.path = new_path 

816 

817 return self 

818 

819 @classmethod 

820 def _iter_items( 

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

822 ) -> Iterator[T_References]: 

823 if common_path is None: 

824 common_path = cls._common_path_default 

825 rela_paths = set() 

826 

827 # Walk loose refs. 

828 # Currently we do not follow links. 

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

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

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

832 if refs_id: 

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

834 # END prune non-refs folders 

835 

836 for f in files: 

837 if f == "packed-refs": 

838 continue 

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

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

841 # END for each file in root directory 

842 # END for each directory to walk 

843 

844 # Read packed refs. 

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

846 if rela_path.startswith(os.fspath(common_path)): 

847 rela_paths.add(rela_path) 

848 # END relative path matches common path 

849 # END packed refs reading 

850 

851 # Yield paths in sorted order. 

852 for path in sorted(rela_paths): 

853 try: 

854 yield cls.from_path(repo, path) 

855 except ValueError: 

856 continue 

857 # END for each sorted relative refpath 

858 

859 @classmethod 

860 def iter_items( 

861 cls: Type[T_References], 

862 repo: "Repo", 

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

864 *args: Any, 

865 **kwargs: Any, 

866 ) -> Iterator[T_References]: 

867 """Find all refs in the repository. 

868 

869 :param repo: 

870 The :class:`~git.repo.base.Repo`. 

871 

872 :param common_path: 

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

874 Ref objects. 

875 Defaults to class specific portion if ``None``, ensuring that only refs 

876 suitable for the actual class are returned. 

877 

878 :return: 

879 A list of :class:`SymbolicReference`, each guaranteed to be a symbolic ref 

880 which is not detached and pointing to a valid ref. 

881 

882 The list is lexicographically sorted. The returned objects are instances of 

883 concrete subclasses, such as :class:`~git.refs.head.Head` or 

884 :class:`~git.refs.tag.TagReference`. 

885 """ 

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

887 

888 @classmethod 

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

890 """Make a symbolic reference from a path. 

891 

892 :param path: 

893 Full ``.git``-directory-relative path name to the Reference to instantiate. 

894 

895 :note: 

896 Use :meth:`to_full_path` if you only have a partial path of a known 

897 Reference type. 

898 

899 :return: 

900 Instance of type :class:`~git.refs.reference.Reference`, 

901 :class:`~git.refs.head.Head`, or :class:`~git.refs.tag.Tag`, depending on 

902 the given path. 

903 """ 

904 if not path: 

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

906 

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

908 # import dependency cycle and don't want to import these names in-function. 

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

910 

911 for ref_type in ( 

912 HEAD, 

913 Head, 

914 RemoteReference, 

915 TagReference, 

916 Reference, 

917 SymbolicReference, 

918 ): 

919 try: 

920 instance = cast(T_References, ref_type(repo, path)) 

921 if instance.__class__ is SymbolicReference and instance.is_detached: 

922 raise ValueError("SymbolicRef was detached, we drop it") 

923 else: 

924 return instance 

925 

926 except ValueError: 

927 pass 

928 # END exception handling 

929 # END for each type to try 

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

931 

932 def is_remote(self) -> bool: 

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

934 return os.fspath(self.path).startswith(self._remote_common_path_default + "/")