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

383 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 

7 

8from gitdb.exc import BadName, BadObject 

9 

10from git.compat import defenc 

11from git.objects.base import Object 

12from git.objects.commit import Commit 

13from git.refs.log import RefLog 

14from git.util import ( 

15 LockedFD, 

16 assure_directory_exists, 

17 hex_to_bin, 

18 join_path, 

19 join_path_native, 

20 to_native_path_linux, 

21) 

22 

23# typing ------------------------------------------------------------------ 

24 

25from typing import ( 

26 Any, 

27 Iterator, 

28 List, 

29 TYPE_CHECKING, 

30 Tuple, 

31 Type, 

32 TypeVar, 

33 Union, 

34 cast, 

35) 

36 

37from git.types import AnyGitObject, PathLike 

38 

39if TYPE_CHECKING: 

40 from git.config import GitConfigParser 

41 from git.objects.commit import Actor 

42 from git.refs.log import RefLogEntry 

43 from git.repo import Repo 

44 

45 

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

47 

48# ------------------------------------------------------------------------------ 

49 

50 

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

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

53 name = f"{path}" 

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

55 return repo.git_dir 

56 return repo.common_dir 

57 

58 

59class SymbolicReference: 

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

61 

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

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

64 

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

66 """ 

67 

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

69 

70 _resolve_ref_on_create = False 

71 _points_to_commits_only = True 

72 _common_path_default = "" 

73 _remote_common_path_default = "refs/remotes" 

74 _id_attribute_ = "name" 

75 

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

77 self.repo = repo 

78 self.path = path 

79 

80 def __str__(self) -> str: 

81 return str(self.path) 

82 

83 def __repr__(self) -> str: 

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

85 

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

87 if hasattr(other, "path"): 

88 other = cast(SymbolicReference, other) 

89 return self.path == other.path 

90 return False 

91 

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

93 return not (self == other) 

94 

95 def __hash__(self) -> int: 

96 return hash(self.path) 

97 

98 @property 

99 def name(self) -> str: 

100 """ 

101 :return: 

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

103 itself. 

104 """ 

105 return str(self.path) 

106 

107 @property 

108 def abspath(self) -> PathLike: 

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

110 

111 @classmethod 

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

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

114 

115 @classmethod 

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

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

118 corresponding refs. 

119 

120 :note: 

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

122 """ 

123 try: 

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

125 for line in fp: 

126 line = line.strip() 

127 if not line: 

128 continue 

129 if line.startswith("#"): 

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

131 # the git source code shows "peeled", 

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

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

134 # refs/packed-backend.c 

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

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

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

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

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

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

141 continue 

142 # END parse comment 

143 

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

145 # tag reference for it. 

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

147 continue 

148 

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

150 # END for each line 

151 except OSError: 

152 return None 

153 # END no packed-refs file handling 

154 

155 @classmethod 

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

157 """ 

158 :return: 

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

160 dereferencing all intermediate references as required 

161 

162 :param repo: 

163 The repository containing the reference at `ref_path`. 

164 """ 

165 

166 while True: 

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

168 if hexsha is not None: 

169 return hexsha 

170 # END recursive dereferencing 

171 

172 @staticmethod 

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

174 """Check a ref name for validity. 

175 

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

177 """ 

178 previous: Union[str, None] = None 

179 one_before_previous: Union[str, None] = None 

180 for c in str(ref_path): 

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

182 raise ValueError( 

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

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

185 ) 

186 elif c == ".": 

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

188 raise ValueError( 

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

190 ) 

191 elif previous == ".": 

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

193 elif c == "/": 

194 if previous == "/": 

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

196 elif previous is None: 

197 raise ValueError( 

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

199 ) 

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

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

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

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

204 

205 one_before_previous = previous 

206 previous = c 

207 

208 if previous == ".": 

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

210 elif previous == "/": 

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

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

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

214 elif any(component.endswith(".lock") for component in str(ref_path).split("/")): 

215 raise ValueError( 

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

217 " '.lock'" 

218 ) 

219 

220 @classmethod 

221 def _get_ref_info_helper( 

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

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

224 """ 

225 :return: 

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

227 

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

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

230 """ 

231 if ref_path: 

232 cls._check_ref_name_valid(ref_path) 

233 

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

235 repodir = _git_dir(repo, ref_path) 

236 try: 

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

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

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

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

241 tokens = value.split() 

242 assert len(tokens) != 0 

243 except OSError: 

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

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

246 # are excluded explicitly. 

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

248 if path != ref_path: 

249 continue 

250 # sha will be used. 

251 tokens = sha, path 

252 break 

253 # END for each packed ref 

254 # END handle packed refs 

255 if tokens is None: 

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

257 

258 # Is it a reference? 

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

260 return (None, tokens[1]) 

261 

262 # It's a commit. 

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

264 return (tokens[0], None) 

265 

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

267 

268 @classmethod 

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

270 """ 

271 :return: 

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

273 

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

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

276 """ 

277 return cls._get_ref_info_helper(repo, ref_path) 

278 

279 def _get_object(self) -> AnyGitObject: 

280 """ 

281 :return: 

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

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

284 """ 

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

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

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

288 

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

290 """ 

291 :return: 

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

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

294 reference will be dereferenced recursively. 

295 """ 

296 obj = self._get_object() 

297 if obj.type == "tag": 

298 obj = obj.object 

299 # END dereference tag 

300 

301 if obj.type != Commit.type: 

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

303 # END handle type 

304 return obj 

305 

306 def set_commit( 

307 self, 

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

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

310 ) -> "SymbolicReference": 

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

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

313 

314 :raise ValueError: 

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

316 point to a commit. 

317 

318 :return: 

319 self 

320 """ 

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

322 invalid_type = False 

323 if isinstance(commit, Object): 

324 invalid_type = commit.type != Commit.type 

325 elif isinstance(commit, SymbolicReference): 

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

327 else: 

328 try: 

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

330 except (BadObject, BadName) as e: 

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

332 # END handle exception 

333 # END verify type 

334 

335 if invalid_type: 

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

337 # END handle raise 

338 

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

340 self.set_object(commit, logmsg) 

341 

342 return self 

343 

344 def set_object( 

345 self, 

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

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

348 ) -> "SymbolicReference": 

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

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

351 

352 :param object: 

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

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

355 

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

357 obtain the git object they point to. 

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

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

360 

361 :param logmsg: 

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

363 Otherwise the reflog is not altered. 

364 

365 :note: 

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

367 by convention. 

368 

369 :return: 

370 self 

371 """ 

372 if isinstance(object, SymbolicReference): 

373 object = object.object # @ReservedAssignment 

374 # END resolve references 

375 

376 is_detached = True 

377 try: 

378 is_detached = self.is_detached 

379 except ValueError: 

380 pass 

381 # END handle non-existing ones 

382 

383 if is_detached: 

384 return self.set_reference(object, logmsg) 

385 

386 # set the commit on our reference 

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

388 

389 @property 

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

391 """Query or set commits directly""" 

392 return self._get_commit() 

393 

394 @commit.setter 

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

396 return self.set_commit(commit) 

397 

398 @property 

399 def object(self) -> AnyGitObject: 

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

401 return self._get_object() 

402 

403 @object.setter 

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

405 return self.set_object(object) 

406 

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

408 """ 

409 :return: 

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

411 

412 :raise TypeError: 

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

414 reference, but to a commit. 

415 """ 

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

417 if target_ref_path is None: 

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

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

420 

421 def set_reference( 

422 self, 

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

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

425 ) -> "SymbolicReference": 

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

427 

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

429 

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

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

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

433 

434 :param ref: 

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

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

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

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

439 

440 :param logmsg: 

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

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

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

444 

445 See also: :meth:`log_append` 

446 

447 :return: 

448 self 

449 

450 :note: 

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

452 :meth:`set_object`. 

453 """ 

454 write_value = None 

455 obj = None 

456 if isinstance(ref, SymbolicReference): 

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

458 elif isinstance(ref, Object): 

459 obj = ref 

460 write_value = ref.hexsha 

461 elif isinstance(ref, str): 

462 try: 

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

464 write_value = obj.hexsha 

465 except (BadObject, BadName) as e: 

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

467 # END end try string 

468 else: 

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

470 # END try commit attribute 

471 

472 # typecheck 

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

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

475 # END verify type 

476 

477 oldbinsha: bytes = b"" 

478 if logmsg is not None: 

479 try: 

480 oldbinsha = self.commit.binsha 

481 except ValueError: 

482 oldbinsha = Commit.NULL_BIN_SHA 

483 # END handle non-existing 

484 # END retrieve old hexsha 

485 

486 fpath = self.abspath 

487 assure_directory_exists(fpath, is_file=True) 

488 

489 lfd = LockedFD(fpath) 

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

491 try: 

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

493 lfd.commit() 

494 except BaseException: 

495 lfd.rollback() 

496 raise 

497 # Adjust the reflog 

498 if logmsg is not None: 

499 self.log_append(oldbinsha, logmsg) 

500 

501 return self 

502 

503 # Aliased reference 

504 @property 

505 def reference(self) -> "SymbolicReference": 

506 return self._get_reference() 

507 

508 @reference.setter 

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

510 return self.set_reference(ref) 

511 

512 ref = reference 

513 

514 def is_valid(self) -> bool: 

515 """ 

516 :return: 

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

518 valid object or reference. 

519 """ 

520 try: 

521 self.object # noqa: B018 

522 except (OSError, ValueError): 

523 return False 

524 else: 

525 return True 

526 

527 @property 

528 def is_detached(self) -> bool: 

529 """ 

530 :return: 

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

532 instead to another reference. 

533 """ 

534 try: 

535 self.ref # noqa: B018 

536 return False 

537 except TypeError: 

538 return True 

539 

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

541 """ 

542 :return: 

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

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

545 

546 :note: 

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

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

549 """ 

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

551 

552 def log_append( 

553 self, 

554 oldbinsha: bytes, 

555 message: Union[str, None], 

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

557 ) -> "RefLogEntry": 

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

559 

560 :param oldbinsha: 

561 Binary sha this ref used to point to. 

562 

563 :param message: 

564 A message describing the change. 

565 

566 :param newbinsha: 

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

568 

569 :return: 

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

571 """ 

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

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

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

575 try: 

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

577 except ValueError: 

578 committer_or_reader = self.repo.config_reader() 

579 # END handle newly cloned repositories 

580 if newbinsha is None: 

581 newbinsha = self.commit.binsha 

582 

583 if message is None: 

584 message = "" 

585 

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

587 

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

589 """ 

590 :return: 

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

592 

593 :param index: 

594 Python list compatible positive or negative index. 

595 

596 :note: 

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

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

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

600 """ 

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

602 

603 @classmethod 

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

605 """ 

606 :return: 

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

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

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

610 """ 

611 if isinstance(path, SymbolicReference): 

612 path = path.path 

613 full_ref_path = path 

614 if not cls._common_path_default: 

615 return full_ref_path 

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

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

618 return full_ref_path 

619 

620 @classmethod 

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

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

623 

624 :param repo: 

625 Repository to delete the reference from. 

626 

627 :param path: 

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

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

630 Alternatively the symbolic reference to be deleted. 

631 """ 

632 full_ref_path = cls.to_full_path(path) 

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

634 if os.path.exists(abs_path): 

635 os.remove(abs_path) 

636 else: 

637 # Check packed refs. 

638 pack_file_path = cls._get_packed_refs_path(repo) 

639 try: 

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

641 new_lines = [] 

642 made_change = False 

643 dropped_last_line = False 

644 for line_bytes in reader: 

645 line = line_bytes.decode(defenc) 

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

647 line_ref = line_ref.strip() 

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

649 # the line. 

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

651 # object, we drop it as well. 

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

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

654 ): 

655 new_lines.append(line) 

656 dropped_last_line = False 

657 continue 

658 # END skip comments and lines without our path 

659 

660 # Drop this line. 

661 made_change = True 

662 dropped_last_line = True 

663 

664 # Write the new lines. 

665 if made_change: 

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

667 # in text mode and change LF to CRLF! 

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

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

670 

671 except OSError: 

672 pass # It didn't exist at all. 

673 

674 # Delete the reflog. 

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

676 if os.path.isfile(reflog_path): 

677 os.remove(reflog_path) 

678 # END remove reflog 

679 

680 @classmethod 

681 def _create( 

682 cls: Type[T_References], 

683 repo: "Repo", 

684 path: PathLike, 

685 resolve: bool, 

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

687 force: bool, 

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

689 ) -> T_References: 

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

691 

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

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

694 and a detached symbolic reference will be created instead. 

695 """ 

696 git_dir = _git_dir(repo, path) 

697 full_ref_path = cls.to_full_path(path) 

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

699 

700 # Figure out target data. 

701 target = reference 

702 if resolve: 

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

704 

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

706 target_data = str(target) 

707 if isinstance(target, SymbolicReference): 

708 target_data = str(target.path) 

709 if not resolve: 

710 target_data = "ref: " + target_data 

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

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

713 if existing_data != target_data: 

714 raise OSError( 

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

716 % (full_ref_path, existing_data, target_data) 

717 ) 

718 # END no force handling 

719 

720 ref = cls(repo, full_ref_path) 

721 ref.set_reference(target, logmsg) 

722 return ref 

723 

724 @classmethod 

725 def create( 

726 cls: Type[T_References], 

727 repo: "Repo", 

728 path: PathLike, 

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

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

731 force: bool = False, 

732 **kwargs: Any, 

733 ) -> T_References: 

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

735 

736 :param repo: 

737 Repository to create the reference in. 

738 

739 :param path: 

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

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

742 

743 :param reference: 

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

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

746 

747 :param force: 

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

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

750 

751 :param logmsg: 

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

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

754 

755 :return: 

756 Newly created symbolic reference 

757 

758 :raise OSError: 

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

760 exists. 

761 

762 :note: 

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

764 """ 

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

766 

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

768 """Rename self to a new path. 

769 

770 :param new_path: 

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

772 ``features/new_name``. 

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

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

775 

776 :param force: 

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

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

779 

780 :return: 

781 self 

782 

783 :raise OSError: 

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

785 """ 

786 new_path = self.to_full_path(new_path) 

787 if self.path == new_path: 

788 return self 

789 

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

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

792 if os.path.isfile(new_abs_path): 

793 if not force: 

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

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

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

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

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

799 if f1 != f2: 

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

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

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

803 # END not force handling 

804 os.remove(new_abs_path) 

805 # END handle existing target file 

806 

807 dname = os.path.dirname(new_abs_path) 

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

809 os.makedirs(dname) 

810 # END create directory 

811 

812 os.rename(cur_abs_path, new_abs_path) 

813 self.path = new_path 

814 

815 return self 

816 

817 @classmethod 

818 def _iter_items( 

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

820 ) -> Iterator[T_References]: 

821 if common_path is None: 

822 common_path = cls._common_path_default 

823 rela_paths = set() 

824 

825 # Walk loose refs. 

826 # Currently we do not follow links. 

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

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

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

830 if refs_id: 

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

832 # END prune non-refs folders 

833 

834 for f in files: 

835 if f == "packed-refs": 

836 continue 

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

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

839 # END for each file in root directory 

840 # END for each directory to walk 

841 

842 # Read packed refs. 

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

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

845 rela_paths.add(rela_path) 

846 # END relative path matches common path 

847 # END packed refs reading 

848 

849 # Yield paths in sorted order. 

850 for path in sorted(rela_paths): 

851 try: 

852 yield cls.from_path(repo, path) 

853 except ValueError: 

854 continue 

855 # END for each sorted relative refpath 

856 

857 @classmethod 

858 def iter_items( 

859 cls: Type[T_References], 

860 repo: "Repo", 

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

862 *args: Any, 

863 **kwargs: Any, 

864 ) -> Iterator[T_References]: 

865 """Find all refs in the repository. 

866 

867 :param repo: 

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

869 

870 :param common_path: 

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

872 Ref objects. 

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

874 suitable for the actual class are returned. 

875 

876 :return: 

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

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

879 

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

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

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

883 """ 

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

885 

886 @classmethod 

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

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

889 

890 :param path: 

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

892 

893 :note: 

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

895 Reference type. 

896 

897 :return: 

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

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

900 the given path. 

901 """ 

902 if not path: 

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

904 

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

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

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

908 

909 for ref_type in ( 

910 HEAD, 

911 Head, 

912 RemoteReference, 

913 TagReference, 

914 Reference, 

915 SymbolicReference, 

916 ): 

917 try: 

918 instance: T_References 

919 instance = ref_type(repo, path) 

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

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

922 else: 

923 return instance 

924 

925 except ValueError: 

926 pass 

927 # END exception handling 

928 # END for each type to try 

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

930 

931 def is_remote(self) -> bool: 

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

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