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

384 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.refs.reference import Reference 

44 from git.repo import Repo 

45 

46 

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

48 

49# ------------------------------------------------------------------------------ 

50 

51 

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

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

54 name = f"{path}" 

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

56 return repo.git_dir 

57 return repo.common_dir 

58 

59 

60class SymbolicReference: 

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

62 

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

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

65 

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

67 """ 

68 

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

70 

71 _resolve_ref_on_create = False 

72 _points_to_commits_only = True 

73 _common_path_default = "" 

74 _remote_common_path_default = "refs/remotes" 

75 _id_attribute_ = "name" 

76 

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

78 self.repo = repo 

79 self.path = path 

80 

81 def __str__(self) -> str: 

82 return str(self.path) 

83 

84 def __repr__(self) -> str: 

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

86 

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

88 if hasattr(other, "path"): 

89 other = cast(SymbolicReference, other) 

90 return self.path == other.path 

91 return False 

92 

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

94 return not (self == other) 

95 

96 def __hash__(self) -> int: 

97 return hash(self.path) 

98 

99 @property 

100 def name(self) -> str: 

101 """ 

102 :return: 

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

104 itself. 

105 """ 

106 return str(self.path) 

107 

108 @property 

109 def abspath(self) -> PathLike: 

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

111 

112 @classmethod 

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

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

115 

116 @classmethod 

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

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

119 corresponding refs. 

120 

121 :note: 

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

123 """ 

124 try: 

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

126 for line in fp: 

127 line = line.strip() 

128 if not line: 

129 continue 

130 if line.startswith("#"): 

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

132 # the git source code shows "peeled", 

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

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

135 # refs/packed-backend.c 

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

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

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

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

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

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

142 continue 

143 # END parse comment 

144 

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

146 # tag reference for it. 

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

148 continue 

149 

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

151 # END for each line 

152 except OSError: 

153 return None 

154 # END no packed-refs file handling 

155 

156 @classmethod 

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

158 """ 

159 :return: 

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

161 dereferencing all intermediate references as required 

162 

163 :param repo: 

164 The repository containing the reference at `ref_path`. 

165 """ 

166 

167 while True: 

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

169 if hexsha is not None: 

170 return hexsha 

171 # END recursive dereferencing 

172 

173 @staticmethod 

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

175 """Check a ref name for validity. 

176 

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

178 """ 

179 previous: Union[str, None] = None 

180 one_before_previous: Union[str, None] = None 

181 for c in str(ref_path): 

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

183 raise ValueError( 

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

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

186 ) 

187 elif c == ".": 

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

189 raise ValueError( 

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

191 ) 

192 elif previous == ".": 

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

194 elif c == "/": 

195 if previous == "/": 

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

197 elif previous is None: 

198 raise ValueError( 

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

200 ) 

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

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

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

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

205 

206 one_before_previous = previous 

207 previous = c 

208 

209 if previous == ".": 

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

211 elif previous == "/": 

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

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

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

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

216 raise ValueError( 

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

218 " '.lock'" 

219 ) 

220 

221 @classmethod 

222 def _get_ref_info_helper( 

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

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

225 """ 

226 :return: 

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

228 

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

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

231 """ 

232 if ref_path: 

233 cls._check_ref_name_valid(ref_path) 

234 

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

236 repodir = _git_dir(repo, ref_path) 

237 try: 

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

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

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

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

242 tokens = value.split() 

243 assert len(tokens) != 0 

244 except OSError: 

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

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

247 # are excluded explicitly. 

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

249 if path != ref_path: 

250 continue 

251 # sha will be used. 

252 tokens = sha, path 

253 break 

254 # END for each packed ref 

255 # END handle packed refs 

256 if tokens is None: 

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

258 

259 # Is it a reference? 

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

261 return (None, tokens[1]) 

262 

263 # It's a commit. 

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

265 return (tokens[0], None) 

266 

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

268 

269 @classmethod 

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

271 """ 

272 :return: 

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

274 

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

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

277 """ 

278 return cls._get_ref_info_helper(repo, ref_path) 

279 

280 def _get_object(self) -> AnyGitObject: 

281 """ 

282 :return: 

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

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

285 """ 

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

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

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

289 

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

291 """ 

292 :return: 

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

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

295 reference will be dereferenced recursively. 

296 """ 

297 obj = self._get_object() 

298 if obj.type == "tag": 

299 obj = obj.object 

300 # END dereference tag 

301 

302 if obj.type != Commit.type: 

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

304 # END handle type 

305 return obj 

306 

307 def set_commit( 

308 self, 

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

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

311 ) -> "SymbolicReference": 

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

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

314 

315 :raise ValueError: 

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

317 point to a commit. 

318 

319 :return: 

320 self 

321 """ 

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

323 invalid_type = False 

324 if isinstance(commit, Object): 

325 invalid_type = commit.type != Commit.type 

326 elif isinstance(commit, SymbolicReference): 

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

328 else: 

329 try: 

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

331 except (BadObject, BadName) as e: 

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

333 # END handle exception 

334 # END verify type 

335 

336 if invalid_type: 

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

338 # END handle raise 

339 

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

341 self.set_object(commit, logmsg) 

342 

343 return self 

344 

345 def set_object( 

346 self, 

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

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

349 ) -> "SymbolicReference": 

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

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

352 

353 :param object: 

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

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

356 

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

358 obtain the git object they point to. 

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

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

361 

362 :param logmsg: 

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

364 Otherwise the reflog is not altered. 

365 

366 :note: 

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

368 by convention. 

369 

370 :return: 

371 self 

372 """ 

373 if isinstance(object, SymbolicReference): 

374 object = object.object # @ReservedAssignment 

375 # END resolve references 

376 

377 is_detached = True 

378 try: 

379 is_detached = self.is_detached 

380 except ValueError: 

381 pass 

382 # END handle non-existing ones 

383 

384 if is_detached: 

385 return self.set_reference(object, logmsg) 

386 

387 # set the commit on our reference 

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

389 

390 @property 

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

392 """Query or set commits directly""" 

393 return self._get_commit() 

394 

395 @commit.setter 

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

397 return self.set_commit(commit) 

398 

399 @property 

400 def object(self) -> AnyGitObject: 

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

402 return self._get_object() 

403 

404 @object.setter 

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

406 return self.set_object(object) 

407 

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

409 """ 

410 :return: 

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

412 

413 :raise TypeError: 

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

415 reference, but to a commit. 

416 """ 

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

418 if target_ref_path is None: 

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

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

421 

422 def set_reference( 

423 self, 

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

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

426 ) -> "SymbolicReference": 

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

428 

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

430 

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

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

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

434 

435 :param ref: 

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

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

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

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

440 

441 :param logmsg: 

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

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

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

445 

446 See also: :meth:`log_append` 

447 

448 :return: 

449 self 

450 

451 :note: 

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

453 :meth:`set_object`. 

454 """ 

455 write_value = None 

456 obj = None 

457 if isinstance(ref, SymbolicReference): 

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

459 elif isinstance(ref, Object): 

460 obj = ref 

461 write_value = ref.hexsha 

462 elif isinstance(ref, str): 

463 try: 

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

465 write_value = obj.hexsha 

466 except (BadObject, BadName) as e: 

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

468 # END end try string 

469 else: 

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

471 # END try commit attribute 

472 

473 # typecheck 

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

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

476 # END verify type 

477 

478 oldbinsha: bytes = b"" 

479 if logmsg is not None: 

480 try: 

481 oldbinsha = self.commit.binsha 

482 except ValueError: 

483 oldbinsha = Commit.NULL_BIN_SHA 

484 # END handle non-existing 

485 # END retrieve old hexsha 

486 

487 fpath = self.abspath 

488 assure_directory_exists(fpath, is_file=True) 

489 

490 lfd = LockedFD(fpath) 

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

492 try: 

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

494 lfd.commit() 

495 except BaseException: 

496 lfd.rollback() 

497 raise 

498 # Adjust the reflog 

499 if logmsg is not None: 

500 self.log_append(oldbinsha, logmsg) 

501 

502 return self 

503 

504 # Aliased reference 

505 @property 

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

507 return self._get_reference() 

508 

509 @reference.setter 

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

511 return self.set_reference(ref) 

512 

513 ref = reference 

514 

515 def is_valid(self) -> bool: 

516 """ 

517 :return: 

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

519 valid object or reference. 

520 """ 

521 try: 

522 self.object # noqa: B018 

523 except (OSError, ValueError): 

524 return False 

525 else: 

526 return True 

527 

528 @property 

529 def is_detached(self) -> bool: 

530 """ 

531 :return: 

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

533 instead to another reference. 

534 """ 

535 try: 

536 self.ref # noqa: B018 

537 return False 

538 except TypeError: 

539 return True 

540 

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

542 """ 

543 :return: 

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

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

546 

547 :note: 

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

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

550 """ 

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

552 

553 def log_append( 

554 self, 

555 oldbinsha: bytes, 

556 message: Union[str, None], 

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

558 ) -> "RefLogEntry": 

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

560 

561 :param oldbinsha: 

562 Binary sha this ref used to point to. 

563 

564 :param message: 

565 A message describing the change. 

566 

567 :param newbinsha: 

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

569 

570 :return: 

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

572 """ 

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

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

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

576 try: 

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

578 except ValueError: 

579 committer_or_reader = self.repo.config_reader() 

580 # END handle newly cloned repositories 

581 if newbinsha is None: 

582 newbinsha = self.commit.binsha 

583 

584 if message is None: 

585 message = "" 

586 

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

588 

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

590 """ 

591 :return: 

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

593 

594 :param index: 

595 Python list compatible positive or negative index. 

596 

597 :note: 

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

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

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

601 """ 

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

603 

604 @classmethod 

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

606 """ 

607 :return: 

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

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

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

611 """ 

612 if isinstance(path, SymbolicReference): 

613 path = path.path 

614 full_ref_path = path 

615 if not cls._common_path_default: 

616 return full_ref_path 

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

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

619 return full_ref_path 

620 

621 @classmethod 

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

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

624 

625 :param repo: 

626 Repository to delete the reference from. 

627 

628 :param path: 

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

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

631 Alternatively the symbolic reference to be deleted. 

632 """ 

633 full_ref_path = cls.to_full_path(path) 

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

635 if os.path.exists(abs_path): 

636 os.remove(abs_path) 

637 else: 

638 # Check packed refs. 

639 pack_file_path = cls._get_packed_refs_path(repo) 

640 try: 

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

642 new_lines = [] 

643 made_change = False 

644 dropped_last_line = False 

645 for line_bytes in reader: 

646 line = line_bytes.decode(defenc) 

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

648 line_ref = line_ref.strip() 

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

650 # the line. 

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

652 # object, we drop it as well. 

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

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

655 ): 

656 new_lines.append(line) 

657 dropped_last_line = False 

658 continue 

659 # END skip comments and lines without our path 

660 

661 # Drop this line. 

662 made_change = True 

663 dropped_last_line = True 

664 

665 # Write the new lines. 

666 if made_change: 

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

668 # in text mode and change LF to CRLF! 

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

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

671 

672 except OSError: 

673 pass # It didn't exist at all. 

674 

675 # Delete the reflog. 

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

677 if os.path.isfile(reflog_path): 

678 os.remove(reflog_path) 

679 # END remove reflog 

680 

681 @classmethod 

682 def _create( 

683 cls: Type[T_References], 

684 repo: "Repo", 

685 path: PathLike, 

686 resolve: bool, 

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

688 force: bool, 

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

690 ) -> T_References: 

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

692 

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

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

695 and a detached symbolic reference will be created instead. 

696 """ 

697 git_dir = _git_dir(repo, path) 

698 full_ref_path = cls.to_full_path(path) 

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

700 

701 # Figure out target data. 

702 target = reference 

703 if resolve: 

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

705 

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

707 target_data = str(target) 

708 if isinstance(target, SymbolicReference): 

709 target_data = str(target.path) 

710 if not resolve: 

711 target_data = "ref: " + target_data 

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

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

714 if existing_data != target_data: 

715 raise OSError( 

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

717 % (full_ref_path, existing_data, target_data) 

718 ) 

719 # END no force handling 

720 

721 ref = cls(repo, full_ref_path) 

722 ref.set_reference(target, logmsg) 

723 return ref 

724 

725 @classmethod 

726 def create( 

727 cls: Type[T_References], 

728 repo: "Repo", 

729 path: PathLike, 

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

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

732 force: bool = False, 

733 **kwargs: Any, 

734 ) -> T_References: 

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

736 

737 :param repo: 

738 Repository to create the reference in. 

739 

740 :param path: 

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

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

743 

744 :param reference: 

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

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

747 

748 :param force: 

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

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

751 

752 :param logmsg: 

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

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

755 

756 :return: 

757 Newly created symbolic reference 

758 

759 :raise OSError: 

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

761 exists. 

762 

763 :note: 

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

765 """ 

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

767 

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

769 """Rename self to a new path. 

770 

771 :param new_path: 

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

773 ``features/new_name``. 

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

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

776 

777 :param force: 

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

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

780 

781 :return: 

782 self 

783 

784 :raise OSError: 

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

786 """ 

787 new_path = self.to_full_path(new_path) 

788 if self.path == new_path: 

789 return self 

790 

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

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

793 if os.path.isfile(new_abs_path): 

794 if not force: 

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

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

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

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

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

800 if f1 != f2: 

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

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

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

804 # END not force handling 

805 os.remove(new_abs_path) 

806 # END handle existing target file 

807 

808 dname = os.path.dirname(new_abs_path) 

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

810 os.makedirs(dname) 

811 # END create directory 

812 

813 os.rename(cur_abs_path, new_abs_path) 

814 self.path = new_path 

815 

816 return self 

817 

818 @classmethod 

819 def _iter_items( 

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

821 ) -> Iterator[T_References]: 

822 if common_path is None: 

823 common_path = cls._common_path_default 

824 rela_paths = set() 

825 

826 # Walk loose refs. 

827 # Currently we do not follow links. 

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

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

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

831 if refs_id: 

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

833 # END prune non-refs folders 

834 

835 for f in files: 

836 if f == "packed-refs": 

837 continue 

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

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

840 # END for each file in root directory 

841 # END for each directory to walk 

842 

843 # Read packed refs. 

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

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

846 rela_paths.add(rela_path) 

847 # END relative path matches common path 

848 # END packed refs reading 

849 

850 # Yield paths in sorted order. 

851 for path in sorted(rela_paths): 

852 try: 

853 yield cls.from_path(repo, path) 

854 except ValueError: 

855 continue 

856 # END for each sorted relative refpath 

857 

858 @classmethod 

859 def iter_items( 

860 cls: Type[T_References], 

861 repo: "Repo", 

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

863 *args: Any, 

864 **kwargs: Any, 

865 ) -> Iterator[T_References]: 

866 """Find all refs in the repository. 

867 

868 :param repo: 

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

870 

871 :param common_path: 

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

873 Ref objects. 

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

875 suitable for the actual class are returned. 

876 

877 :return: 

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

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

880 

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

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

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

884 """ 

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

886 

887 @classmethod 

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

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

890 

891 :param path: 

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

893 

894 :note: 

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

896 Reference type. 

897 

898 :return: 

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

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

901 the given path. 

902 """ 

903 if not path: 

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

905 

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

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

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

909 

910 for ref_type in ( 

911 HEAD, 

912 Head, 

913 RemoteReference, 

914 TagReference, 

915 Reference, 

916 SymbolicReference, 

917 ): 

918 try: 

919 instance: T_References 

920 instance = 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 str(self.path).startswith(self._remote_common_path_default + "/")