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

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

368 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 import Head, TagReference, RemoteReference, Reference 

43 from git.refs.log import RefLogEntry 

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 commit = property( 

391 _get_commit, 

392 set_commit, # type: ignore[arg-type] 

393 doc="Query or set commits directly", 

394 ) 

395 

396 object = property( 

397 _get_object, 

398 set_object, # type: ignore[arg-type] 

399 doc="Return the object our ref currently refers to", 

400 ) 

401 

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

403 """ 

404 :return: 

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

406 

407 :raise TypeError: 

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

409 reference, but to a commit. 

410 """ 

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

412 if target_ref_path is None: 

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

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

415 

416 def set_reference( 

417 self, 

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

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

420 ) -> "SymbolicReference": 

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

422 

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

424 

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

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

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

428 

429 :param ref: 

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

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

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

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

434 

435 :param logmsg: 

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

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

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

439 

440 See also: :meth:`log_append` 

441 

442 :return: 

443 self 

444 

445 :note: 

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

447 :meth:`set_object`. 

448 """ 

449 write_value = None 

450 obj = None 

451 if isinstance(ref, SymbolicReference): 

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

453 elif isinstance(ref, Object): 

454 obj = ref 

455 write_value = ref.hexsha 

456 elif isinstance(ref, str): 

457 try: 

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

459 write_value = obj.hexsha 

460 except (BadObject, BadName) as e: 

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

462 # END end try string 

463 else: 

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

465 # END try commit attribute 

466 

467 # typecheck 

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

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

470 # END verify type 

471 

472 oldbinsha: bytes = b"" 

473 if logmsg is not None: 

474 try: 

475 oldbinsha = self.commit.binsha 

476 except ValueError: 

477 oldbinsha = Commit.NULL_BIN_SHA 

478 # END handle non-existing 

479 # END retrieve old hexsha 

480 

481 fpath = self.abspath 

482 assure_directory_exists(fpath, is_file=True) 

483 

484 lfd = LockedFD(fpath) 

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

486 try: 

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

488 lfd.commit() 

489 except BaseException: 

490 lfd.rollback() 

491 raise 

492 # Adjust the reflog 

493 if logmsg is not None: 

494 self.log_append(oldbinsha, logmsg) 

495 

496 return self 

497 

498 # Aliased reference 

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

500 reference = property( # type: ignore[assignment] 

501 _get_reference, 

502 set_reference, # type: ignore[arg-type] 

503 doc="Returns the Reference we point to", 

504 ) 

505 ref = reference 

506 

507 def is_valid(self) -> bool: 

508 """ 

509 :return: 

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

511 valid object or reference. 

512 """ 

513 try: 

514 self.object # noqa: B018 

515 except (OSError, ValueError): 

516 return False 

517 else: 

518 return True 

519 

520 @property 

521 def is_detached(self) -> bool: 

522 """ 

523 :return: 

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

525 instead to another reference. 

526 """ 

527 try: 

528 self.ref # noqa: B018 

529 return False 

530 except TypeError: 

531 return True 

532 

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

534 """ 

535 :return: 

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

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

538 

539 :note: 

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

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

542 """ 

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

544 

545 def log_append( 

546 self, 

547 oldbinsha: bytes, 

548 message: Union[str, None], 

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

550 ) -> "RefLogEntry": 

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

552 

553 :param oldbinsha: 

554 Binary sha this ref used to point to. 

555 

556 :param message: 

557 A message describing the change. 

558 

559 :param newbinsha: 

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

561 

562 :return: 

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

564 """ 

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

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

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

568 try: 

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

570 except ValueError: 

571 committer_or_reader = self.repo.config_reader() 

572 # END handle newly cloned repositories 

573 if newbinsha is None: 

574 newbinsha = self.commit.binsha 

575 

576 if message is None: 

577 message = "" 

578 

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

580 

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

582 """ 

583 :return: 

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

585 

586 :param index: 

587 Python list compatible positive or negative index. 

588 

589 :note: 

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

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

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

593 """ 

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

595 

596 @classmethod 

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

598 """ 

599 :return: 

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

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

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

603 """ 

604 if isinstance(path, SymbolicReference): 

605 path = path.path 

606 full_ref_path = path 

607 if not cls._common_path_default: 

608 return full_ref_path 

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

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

611 return full_ref_path 

612 

613 @classmethod 

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

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

616 

617 :param repo: 

618 Repository to delete the reference from. 

619 

620 :param path: 

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

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

623 Alternatively the symbolic reference to be deleted. 

624 """ 

625 full_ref_path = cls.to_full_path(path) 

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

627 if os.path.exists(abs_path): 

628 os.remove(abs_path) 

629 else: 

630 # Check packed refs. 

631 pack_file_path = cls._get_packed_refs_path(repo) 

632 try: 

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

634 new_lines = [] 

635 made_change = False 

636 dropped_last_line = False 

637 for line_bytes in reader: 

638 line = line_bytes.decode(defenc) 

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

640 line_ref = line_ref.strip() 

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

642 # the line. 

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

644 # object, we drop it as well. 

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

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

647 ): 

648 new_lines.append(line) 

649 dropped_last_line = False 

650 continue 

651 # END skip comments and lines without our path 

652 

653 # Drop this line. 

654 made_change = True 

655 dropped_last_line = True 

656 

657 # Write the new lines. 

658 if made_change: 

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

660 # in text mode and change LF to CRLF! 

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

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

663 

664 except OSError: 

665 pass # It didn't exist at all. 

666 

667 # Delete the reflog. 

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

669 if os.path.isfile(reflog_path): 

670 os.remove(reflog_path) 

671 # END remove reflog 

672 

673 @classmethod 

674 def _create( 

675 cls: Type[T_References], 

676 repo: "Repo", 

677 path: PathLike, 

678 resolve: bool, 

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

680 force: bool, 

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

682 ) -> T_References: 

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

684 

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

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

687 and a detached symbolic reference will be created instead. 

688 """ 

689 git_dir = _git_dir(repo, path) 

690 full_ref_path = cls.to_full_path(path) 

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

692 

693 # Figure out target data. 

694 target = reference 

695 if resolve: 

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

697 

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

699 target_data = str(target) 

700 if isinstance(target, SymbolicReference): 

701 target_data = str(target.path) 

702 if not resolve: 

703 target_data = "ref: " + target_data 

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

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

706 if existing_data != target_data: 

707 raise OSError( 

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

709 % (full_ref_path, existing_data, target_data) 

710 ) 

711 # END no force handling 

712 

713 ref = cls(repo, full_ref_path) 

714 ref.set_reference(target, logmsg) 

715 return ref 

716 

717 @classmethod 

718 def create( 

719 cls: Type[T_References], 

720 repo: "Repo", 

721 path: PathLike, 

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

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

724 force: bool = False, 

725 **kwargs: Any, 

726 ) -> T_References: 

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

728 

729 :param repo: 

730 Repository to create the reference in. 

731 

732 :param path: 

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

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

735 

736 :param reference: 

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

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

739 

740 :param force: 

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

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

743 

744 :param logmsg: 

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

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

747 

748 :return: 

749 Newly created symbolic reference 

750 

751 :raise OSError: 

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

753 exists. 

754 

755 :note: 

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

757 """ 

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

759 

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

761 """Rename self to a new path. 

762 

763 :param new_path: 

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

765 ``features/new_name``. 

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

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

768 

769 :param force: 

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

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

772 

773 :return: 

774 self 

775 

776 :raise OSError: 

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

778 """ 

779 new_path = self.to_full_path(new_path) 

780 if self.path == new_path: 

781 return self 

782 

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

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

785 if os.path.isfile(new_abs_path): 

786 if not force: 

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

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

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

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

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

792 if f1 != f2: 

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

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

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

796 # END not force handling 

797 os.remove(new_abs_path) 

798 # END handle existing target file 

799 

800 dname = os.path.dirname(new_abs_path) 

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

802 os.makedirs(dname) 

803 # END create directory 

804 

805 os.rename(cur_abs_path, new_abs_path) 

806 self.path = new_path 

807 

808 return self 

809 

810 @classmethod 

811 def _iter_items( 

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

813 ) -> Iterator[T_References]: 

814 if common_path is None: 

815 common_path = cls._common_path_default 

816 rela_paths = set() 

817 

818 # Walk loose refs. 

819 # Currently we do not follow links. 

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

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

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

823 if refs_id: 

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

825 # END prune non-refs folders 

826 

827 for f in files: 

828 if f == "packed-refs": 

829 continue 

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

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

832 # END for each file in root directory 

833 # END for each directory to walk 

834 

835 # Read packed refs. 

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

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

838 rela_paths.add(rela_path) 

839 # END relative path matches common path 

840 # END packed refs reading 

841 

842 # Yield paths in sorted order. 

843 for path in sorted(rela_paths): 

844 try: 

845 yield cls.from_path(repo, path) 

846 except ValueError: 

847 continue 

848 # END for each sorted relative refpath 

849 

850 @classmethod 

851 def iter_items( 

852 cls: Type[T_References], 

853 repo: "Repo", 

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

855 *args: Any, 

856 **kwargs: Any, 

857 ) -> Iterator[T_References]: 

858 """Find all refs in the repository. 

859 

860 :param repo: 

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

862 

863 :param common_path: 

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

865 Ref objects. 

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

867 suitable for the actual class are returned. 

868 

869 :return: 

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

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

872 

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

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

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

876 """ 

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

878 

879 @classmethod 

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

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

882 

883 :param path: 

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

885 

886 :note: 

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

888 Reference type. 

889 

890 :return: 

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

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

893 the given path. 

894 """ 

895 if not path: 

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

897 

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

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

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

901 

902 for ref_type in ( 

903 HEAD, 

904 Head, 

905 RemoteReference, 

906 TagReference, 

907 Reference, 

908 SymbolicReference, 

909 ): 

910 try: 

911 instance: T_References 

912 instance = ref_type(repo, path) 

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

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

915 else: 

916 return instance 

917 

918 except ValueError: 

919 pass 

920 # END exception handling 

921 # END for each type to try 

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

923 

924 def is_remote(self) -> bool: 

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

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