symbolic.py 34 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934
  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. __all__ = ["SymbolicReference"]
  4. import os
  5. from pathlib import Path
  6. from gitdb.exc import BadName, BadObject
  7. from git.compat import defenc
  8. from git.objects.base import Object
  9. from git.objects.commit import Commit
  10. from git.refs.log import RefLog
  11. from git.util import (
  12. LockedFD,
  13. assure_directory_exists,
  14. hex_to_bin,
  15. join_path,
  16. join_path_native,
  17. to_native_path_linux,
  18. )
  19. # typing ------------------------------------------------------------------
  20. from typing import (
  21. Any,
  22. Iterator,
  23. List,
  24. TYPE_CHECKING,
  25. Tuple,
  26. Type,
  27. TypeVar,
  28. Union,
  29. cast,
  30. )
  31. from git.types import AnyGitObject, PathLike
  32. if TYPE_CHECKING:
  33. from git.config import GitConfigParser
  34. from git.objects.commit import Actor
  35. from git.refs.log import RefLogEntry
  36. from git.refs.reference import Reference
  37. from git.repo import Repo
  38. T_References = TypeVar("T_References", bound="SymbolicReference")
  39. # ------------------------------------------------------------------------------
  40. def _git_dir(repo: "Repo", path: Union[PathLike, None]) -> PathLike:
  41. """Find the git dir that is appropriate for the path."""
  42. name = f"{path}"
  43. if name in ["HEAD", "ORIG_HEAD", "FETCH_HEAD", "index", "logs"]:
  44. return repo.git_dir
  45. return repo.common_dir
  46. class SymbolicReference:
  47. """Special case of a reference that is symbolic.
  48. This does not point to a specific commit, but to another
  49. :class:`~git.refs.head.Head`, which itself specifies a commit.
  50. A typical example for a symbolic reference is :class:`~git.refs.head.HEAD`.
  51. """
  52. __slots__ = ("repo", "path")
  53. _resolve_ref_on_create = False
  54. _points_to_commits_only = True
  55. _common_path_default = ""
  56. _remote_common_path_default = "refs/remotes"
  57. _id_attribute_ = "name"
  58. def __init__(self, repo: "Repo", path: PathLike, check_path: bool = False) -> None:
  59. self.repo = repo
  60. self.path: PathLike = path
  61. def __str__(self) -> str:
  62. return os.fspath(self.path)
  63. def __repr__(self) -> str:
  64. return '<git.%s "%s">' % (self.__class__.__name__, self.path)
  65. def __eq__(self, other: object) -> bool:
  66. if hasattr(other, "path"):
  67. other = cast(SymbolicReference, other)
  68. return self.path == other.path
  69. return False
  70. def __ne__(self, other: object) -> bool:
  71. return not (self == other)
  72. def __hash__(self) -> int:
  73. return hash(self.path)
  74. @property
  75. def name(self) -> str:
  76. """
  77. :return:
  78. In case of symbolic references, the shortest assumable name is the path
  79. itself.
  80. """
  81. return os.fspath(self.path)
  82. @property
  83. def abspath(self) -> PathLike:
  84. return join_path_native(_git_dir(self.repo, self.path), self.path)
  85. @classmethod
  86. def _get_packed_refs_path(cls, repo: "Repo") -> str:
  87. return os.path.join(repo.common_dir, "packed-refs")
  88. @classmethod
  89. def _iter_packed_refs(cls, repo: "Repo") -> Iterator[Tuple[str, str]]:
  90. """Return an iterator yielding pairs of sha1/path pairs (as strings) for the
  91. corresponding refs.
  92. :note:
  93. The packed refs file will be kept open as long as we iterate.
  94. """
  95. try:
  96. with open(cls._get_packed_refs_path(repo), "rt", encoding="UTF-8") as fp:
  97. for line in fp:
  98. line = line.strip()
  99. if not line:
  100. continue
  101. if line.startswith("#"):
  102. # "# pack-refs with: peeled fully-peeled sorted"
  103. # the git source code shows "peeled",
  104. # "fully-peeled" and "sorted" as the keywords
  105. # that can go on this line, as per comments in git file
  106. # refs/packed-backend.c
  107. # I looked at master on 2017-10-11,
  108. # commit 111ef79afe, after tag v2.15.0-rc1
  109. # from repo https://github.com/git/git.git
  110. if line.startswith("# pack-refs with:") and "peeled" not in line:
  111. raise TypeError("PackingType of packed-Refs not understood: %r" % line)
  112. # END abort if we do not understand the packing scheme
  113. continue
  114. # END parse comment
  115. # Skip dereferenced tag object entries - previous line was actual
  116. # tag reference for it.
  117. if line[0] == "^":
  118. continue
  119. yield cast(Tuple[str, str], tuple(line.split(" ", 1)))
  120. # END for each line
  121. except OSError:
  122. return None
  123. # END no packed-refs file handling
  124. @classmethod
  125. def dereference_recursive(cls, repo: "Repo", ref_path: Union[PathLike, None]) -> str:
  126. """
  127. :return:
  128. hexsha stored in the reference at the given `ref_path`, recursively
  129. dereferencing all intermediate references as required
  130. :param repo:
  131. The repository containing the reference at `ref_path`.
  132. """
  133. while True:
  134. hexsha, ref_path = cls._get_ref_info(repo, ref_path)
  135. if hexsha is not None:
  136. return hexsha
  137. # END recursive dereferencing
  138. @staticmethod
  139. def _check_ref_name_valid(ref_path: PathLike) -> None:
  140. """Check a ref name for validity.
  141. This is based on the rules described in :manpage:`git-check-ref-format(1)`.
  142. """
  143. previous: Union[str, None] = None
  144. one_before_previous: Union[str, None] = None
  145. for c in os.fspath(ref_path):
  146. if c in " ~^:?*[\\":
  147. raise ValueError(
  148. f"Invalid reference '{ref_path}': references cannot contain spaces, tildes (~), carets (^),"
  149. f" colons (:), question marks (?), asterisks (*), open brackets ([) or backslashes (\\)"
  150. )
  151. elif c == ".":
  152. if previous is None or previous == "/":
  153. raise ValueError(
  154. f"Invalid reference '{ref_path}': references cannot start with a period (.) or contain '/.'"
  155. )
  156. elif previous == ".":
  157. raise ValueError(f"Invalid reference '{ref_path}': references cannot contain '..'")
  158. elif c == "/":
  159. if previous == "/":
  160. raise ValueError(f"Invalid reference '{ref_path}': references cannot contain '//'")
  161. elif previous is None:
  162. raise ValueError(
  163. f"Invalid reference '{ref_path}': references cannot start with forward slashes '/'"
  164. )
  165. elif c == "{" and previous == "@":
  166. raise ValueError(f"Invalid reference '{ref_path}': references cannot contain '@{{'")
  167. elif ord(c) < 32 or ord(c) == 127:
  168. raise ValueError(f"Invalid reference '{ref_path}': references cannot contain ASCII control characters")
  169. one_before_previous = previous
  170. previous = c
  171. if previous == ".":
  172. raise ValueError(f"Invalid reference '{ref_path}': references cannot end with a period (.)")
  173. elif previous == "/":
  174. raise ValueError(f"Invalid reference '{ref_path}': references cannot end with a forward slash (/)")
  175. elif previous == "@" and one_before_previous is None:
  176. raise ValueError(f"Invalid reference '{ref_path}': references cannot be '@'")
  177. elif any(component.endswith(".lock") for component in Path(ref_path).parts):
  178. raise ValueError(
  179. f"Invalid reference '{ref_path}': references cannot have slash-separated components that end with"
  180. " '.lock'"
  181. )
  182. @classmethod
  183. def _get_ref_info_helper(
  184. cls, repo: "Repo", ref_path: Union[PathLike, None]
  185. ) -> Union[Tuple[str, None], Tuple[None, str]]:
  186. """
  187. :return:
  188. *(str(sha), str(target_ref_path))*, where:
  189. * *sha* is of the file at rela_path points to if available, or ``None``.
  190. * *target_ref_path* is the reference we point to, or ``None``.
  191. """
  192. if ref_path:
  193. cls._check_ref_name_valid(ref_path)
  194. tokens: Union[None, List[str], Tuple[str, str]] = None
  195. repodir = _git_dir(repo, ref_path)
  196. try:
  197. with open(os.path.join(repodir, ref_path), "rt", encoding="UTF-8") as fp: # type: ignore[arg-type]
  198. value = fp.read().rstrip()
  199. # Don't only split on spaces, but on whitespace, which allows to parse lines like:
  200. # 60b64ef992065e2600bfef6187a97f92398a9144 branch 'master' of git-server:/path/to/repo
  201. tokens = value.split()
  202. assert len(tokens) != 0
  203. except OSError:
  204. # Probably we are just packed. Find our entry in the packed refs file.
  205. # NOTE: We are not a symbolic ref if we are in a packed file, as these
  206. # are excluded explicitly.
  207. for sha, path in cls._iter_packed_refs(repo):
  208. if path != ref_path:
  209. continue
  210. # sha will be used.
  211. tokens = sha, path
  212. break
  213. # END for each packed ref
  214. # END handle packed refs
  215. if tokens is None:
  216. raise ValueError("Reference at %r does not exist" % ref_path)
  217. # Is it a reference?
  218. if tokens[0] == "ref:":
  219. return (None, tokens[1])
  220. # It's a commit.
  221. if repo.re_hexsha_only.match(tokens[0]):
  222. return (tokens[0], None)
  223. raise ValueError("Failed to parse reference information from %r" % ref_path)
  224. @classmethod
  225. def _get_ref_info(cls, repo: "Repo", ref_path: Union[PathLike, None]) -> Union[Tuple[str, None], Tuple[None, str]]:
  226. """
  227. :return:
  228. *(str(sha), str(target_ref_path))*, where:
  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. return cls._get_ref_info_helper(repo, ref_path)
  233. def _get_object(self) -> AnyGitObject:
  234. """
  235. :return:
  236. The object our ref currently refers to. Refs can be cached, they will always
  237. point to the actual object as it gets re-created on each query.
  238. """
  239. # We have to be dynamic here as we may be a tag which can point to anything.
  240. # Our path will be resolved to the hexsha which will be used accordingly.
  241. return Object.new_from_sha(self.repo, hex_to_bin(self.dereference_recursive(self.repo, self.path)))
  242. def _get_commit(self) -> "Commit":
  243. """
  244. :return:
  245. :class:`~git.objects.commit.Commit` object we point to. This works for
  246. detached and non-detached :class:`SymbolicReference` instances. The symbolic
  247. reference will be dereferenced recursively.
  248. """
  249. obj = self._get_object()
  250. if obj.type == "tag":
  251. obj = obj.object
  252. # END dereference tag
  253. if obj.type != Commit.type:
  254. raise TypeError("Symbolic Reference pointed to object %r, commit was required" % obj)
  255. # END handle type
  256. return obj
  257. def set_commit(
  258. self,
  259. commit: Union[Commit, "SymbolicReference", str],
  260. logmsg: Union[str, None] = None,
  261. ) -> "SymbolicReference":
  262. """Like :meth:`set_object`, but restricts the type of object to be a
  263. :class:`~git.objects.commit.Commit`.
  264. :raise ValueError:
  265. If `commit` is not a :class:`~git.objects.commit.Commit` object, nor does it
  266. point to a commit.
  267. :return:
  268. self
  269. """
  270. # Check the type - assume the best if it is a base-string.
  271. invalid_type = False
  272. if isinstance(commit, Object):
  273. invalid_type = commit.type != Commit.type
  274. elif isinstance(commit, SymbolicReference):
  275. invalid_type = commit.object.type != Commit.type
  276. else:
  277. try:
  278. invalid_type = self.repo.rev_parse(commit).type != Commit.type
  279. except (BadObject, BadName) as e:
  280. raise ValueError("Invalid object: %s" % commit) from e
  281. # END handle exception
  282. # END verify type
  283. if invalid_type:
  284. raise ValueError("Need commit, got %r" % commit)
  285. # END handle raise
  286. # We leave strings to the rev-parse method below.
  287. self.set_object(commit, logmsg)
  288. return self
  289. def set_object(
  290. self,
  291. object: Union[AnyGitObject, "SymbolicReference", str],
  292. logmsg: Union[str, None] = None,
  293. ) -> "SymbolicReference":
  294. """Set the object we point to, possibly dereference our symbolic reference
  295. first. If the reference does not exist, it will be created.
  296. :param object:
  297. A refspec, a :class:`SymbolicReference` or an
  298. :class:`~git.objects.base.Object` instance.
  299. * :class:`SymbolicReference` instances will be dereferenced beforehand to
  300. obtain the git object they point to.
  301. * :class:`~git.objects.base.Object` instances must represent git objects
  302. (:class:`~git.types.AnyGitObject`).
  303. :param logmsg:
  304. If not ``None``, the message will be used in the reflog entry to be written.
  305. Otherwise the reflog is not altered.
  306. :note:
  307. Plain :class:`SymbolicReference` instances may not actually point to objects
  308. by convention.
  309. :return:
  310. self
  311. """
  312. if isinstance(object, SymbolicReference):
  313. object = object.object # @ReservedAssignment
  314. # END resolve references
  315. is_detached = True
  316. try:
  317. is_detached = self.is_detached
  318. except ValueError:
  319. pass
  320. # END handle non-existing ones
  321. if is_detached:
  322. return self.set_reference(object, logmsg)
  323. # set the commit on our reference
  324. return self._get_reference().set_object(object, logmsg)
  325. @property
  326. def commit(self) -> "Commit":
  327. """Query or set commits directly"""
  328. return self._get_commit()
  329. @commit.setter
  330. def commit(self, commit: Union[Commit, "SymbolicReference", str]) -> "SymbolicReference":
  331. return self.set_commit(commit)
  332. @property
  333. def object(self) -> AnyGitObject:
  334. """Return the object our ref currently refers to"""
  335. return self._get_object()
  336. @object.setter
  337. def object(self, object: Union[AnyGitObject, "SymbolicReference", str]) -> "SymbolicReference":
  338. return self.set_object(object)
  339. def _get_reference(self) -> "Reference":
  340. """
  341. :return:
  342. :class:`~git.refs.reference.Reference` object we point to
  343. :raise TypeError:
  344. If this symbolic reference is detached, hence it doesn't point to a
  345. reference, but to a commit.
  346. """
  347. sha, target_ref_path = self._get_ref_info(self.repo, self.path)
  348. if target_ref_path is None:
  349. raise TypeError("%s is a detached symbolic reference as it points to %r" % (self, sha))
  350. return cast("Reference", self.from_path(self.repo, target_ref_path))
  351. def set_reference(
  352. self,
  353. ref: Union[AnyGitObject, "SymbolicReference", str],
  354. logmsg: Union[str, None] = None,
  355. ) -> "SymbolicReference":
  356. """Set ourselves to the given `ref`.
  357. It will stay a symbol if the `ref` is a :class:`~git.refs.reference.Reference`.
  358. Otherwise a git object, specified as a :class:`~git.objects.base.Object`
  359. instance or refspec, is assumed. If it is valid, this reference will be set to
  360. it, which effectively detaches the reference if it was a purely symbolic one.
  361. :param ref:
  362. A :class:`SymbolicReference` instance, an :class:`~git.objects.base.Object`
  363. instance (specifically an :class:`~git.types.AnyGitObject`), or a refspec
  364. string. Only if the ref is a :class:`SymbolicReference` instance, we will
  365. point to it. Everything else is dereferenced to obtain the actual object.
  366. :param logmsg:
  367. If set to a string, the message will be used in the reflog.
  368. Otherwise, a reflog entry is not written for the changed reference.
  369. The previous commit of the entry will be the commit we point to now.
  370. See also: :meth:`log_append`
  371. :return:
  372. self
  373. :note:
  374. This symbolic reference will not be dereferenced. For that, see
  375. :meth:`set_object`.
  376. """
  377. write_value = None
  378. obj = None
  379. if isinstance(ref, SymbolicReference):
  380. write_value = "ref: %s" % ref.path
  381. elif isinstance(ref, Object):
  382. obj = ref
  383. write_value = ref.hexsha
  384. elif isinstance(ref, str):
  385. try:
  386. obj = self.repo.rev_parse(ref + "^{}") # Optionally dereference tags.
  387. write_value = obj.hexsha
  388. except (BadObject, BadName) as e:
  389. raise ValueError("Could not extract object from %s" % ref) from e
  390. # END end try string
  391. else:
  392. raise ValueError("Unrecognized Value: %r" % ref)
  393. # END try commit attribute
  394. # typecheck
  395. if obj is not None and self._points_to_commits_only and obj.type != Commit.type:
  396. raise TypeError("Require commit, got %r" % obj)
  397. # END verify type
  398. oldbinsha: bytes = b""
  399. if logmsg is not None:
  400. try:
  401. oldbinsha = self.commit.binsha
  402. except ValueError:
  403. oldbinsha = Commit.NULL_BIN_SHA
  404. # END handle non-existing
  405. # END retrieve old hexsha
  406. fpath = self.abspath
  407. assure_directory_exists(fpath, is_file=True)
  408. lfd = LockedFD(fpath)
  409. fd = lfd.open(write=True, stream=True)
  410. try:
  411. fd.write(write_value.encode("utf-8") + b"\n")
  412. lfd.commit()
  413. except BaseException:
  414. lfd.rollback()
  415. raise
  416. # Adjust the reflog
  417. if logmsg is not None:
  418. self.log_append(oldbinsha, logmsg)
  419. return self
  420. # Aliased reference
  421. @property
  422. def reference(self) -> "Reference":
  423. return self._get_reference()
  424. @reference.setter
  425. def reference(self, ref: Union[AnyGitObject, "SymbolicReference", str]) -> "SymbolicReference":
  426. return self.set_reference(ref)
  427. ref = reference
  428. def is_valid(self) -> bool:
  429. """
  430. :return:
  431. ``True`` if the reference is valid, hence it can be read and points to a
  432. valid object or reference.
  433. """
  434. try:
  435. self.object # noqa: B018
  436. except (OSError, ValueError):
  437. return False
  438. else:
  439. return True
  440. @property
  441. def is_detached(self) -> bool:
  442. """
  443. :return:
  444. ``True`` if we are a detached reference, hence we point to a specific commit
  445. instead to another reference.
  446. """
  447. try:
  448. self.ref # noqa: B018
  449. return False
  450. except TypeError:
  451. return True
  452. def log(self) -> "RefLog":
  453. """
  454. :return:
  455. :class:`~git.refs.log.RefLog` for this reference.
  456. Its last entry reflects the latest change applied to this reference.
  457. :note:
  458. As the log is parsed every time, its recommended to cache it for use instead
  459. of calling this method repeatedly. It should be considered read-only.
  460. """
  461. return RefLog.from_file(RefLog.path(self))
  462. def log_append(
  463. self,
  464. oldbinsha: bytes,
  465. message: Union[str, None],
  466. newbinsha: Union[bytes, None] = None,
  467. ) -> "RefLogEntry":
  468. """Append a logentry to the logfile of this ref.
  469. :param oldbinsha:
  470. Binary sha this ref used to point to.
  471. :param message:
  472. A message describing the change.
  473. :param newbinsha:
  474. The sha the ref points to now. If None, our current commit sha will be used.
  475. :return:
  476. The added :class:`~git.refs.log.RefLogEntry` instance.
  477. """
  478. # NOTE: We use the committer of the currently active commit - this should be
  479. # correct to allow overriding the committer on a per-commit level.
  480. # See https://github.com/gitpython-developers/GitPython/pull/146.
  481. try:
  482. committer_or_reader: Union["Actor", "GitConfigParser"] = self.commit.committer
  483. except ValueError:
  484. committer_or_reader = self.repo.config_reader()
  485. # END handle newly cloned repositories
  486. if newbinsha is None:
  487. newbinsha = self.commit.binsha
  488. if message is None:
  489. message = ""
  490. return RefLog.append_entry(committer_or_reader, RefLog.path(self), oldbinsha, newbinsha, message)
  491. def log_entry(self, index: int) -> "RefLogEntry":
  492. """
  493. :return:
  494. :class:`~git.refs.log.RefLogEntry` at the given index
  495. :param index:
  496. Python list compatible positive or negative index.
  497. :note:
  498. This method must read part of the reflog during execution, hence it should
  499. be used sparingly, or only if you need just one index. In that case, it will
  500. be faster than the :meth:`log` method.
  501. """
  502. return RefLog.entry_at(RefLog.path(self), index)
  503. @classmethod
  504. def to_full_path(cls, path: Union[PathLike, "SymbolicReference"]) -> PathLike:
  505. """
  506. :return:
  507. String with a full repository-relative path which can be used to initialize
  508. a :class:`~git.refs.reference.Reference` instance, for instance by using
  509. :meth:`Reference.from_path <git.refs.reference.Reference.from_path>`.
  510. """
  511. if isinstance(path, SymbolicReference):
  512. path = path.path
  513. full_ref_path = path
  514. if not cls._common_path_default:
  515. return full_ref_path
  516. if not os.fspath(path).startswith(cls._common_path_default + "/"):
  517. full_ref_path = "%s/%s" % (cls._common_path_default, path)
  518. return full_ref_path
  519. @classmethod
  520. def delete(cls, repo: "Repo", path: PathLike) -> None:
  521. """Delete the reference at the given path.
  522. :param repo:
  523. Repository to delete the reference from.
  524. :param path:
  525. Short or full path pointing to the reference, e.g. ``refs/myreference`` or
  526. just ``myreference``, hence ``refs/`` is implied.
  527. Alternatively the symbolic reference to be deleted.
  528. """
  529. full_ref_path = cls.to_full_path(path)
  530. abs_path = os.path.join(repo.common_dir, full_ref_path)
  531. if os.path.exists(abs_path):
  532. os.remove(abs_path)
  533. else:
  534. # Check packed refs.
  535. pack_file_path = cls._get_packed_refs_path(repo)
  536. try:
  537. with open(pack_file_path, "rb") as reader:
  538. new_lines = []
  539. made_change = False
  540. dropped_last_line = False
  541. for line_bytes in reader:
  542. line = line_bytes.decode(defenc)
  543. _, _, line_ref = line.partition(" ")
  544. line_ref = line_ref.strip()
  545. # Keep line if it is a comment or if the ref to delete is not in
  546. # the line.
  547. # If we deleted the last line and this one is a tag-reference
  548. # object, we drop it as well.
  549. if (line.startswith("#") or full_ref_path != line_ref) and (
  550. not dropped_last_line or dropped_last_line and not line.startswith("^")
  551. ):
  552. new_lines.append(line)
  553. dropped_last_line = False
  554. continue
  555. # END skip comments and lines without our path
  556. # Drop this line.
  557. made_change = True
  558. dropped_last_line = True
  559. # Write the new lines.
  560. if made_change:
  561. # Binary writing is required, otherwise Windows will open the file
  562. # in text mode and change LF to CRLF!
  563. with open(pack_file_path, "wb") as fd:
  564. fd.writelines(line.encode(defenc) for line in new_lines)
  565. except OSError:
  566. pass # It didn't exist at all.
  567. # Delete the reflog.
  568. reflog_path = RefLog.path(cls(repo, full_ref_path))
  569. if os.path.isfile(reflog_path):
  570. os.remove(reflog_path)
  571. # END remove reflog
  572. @classmethod
  573. def _create(
  574. cls: Type[T_References],
  575. repo: "Repo",
  576. path: PathLike,
  577. resolve: bool,
  578. reference: Union["SymbolicReference", str],
  579. force: bool,
  580. logmsg: Union[str, None] = None,
  581. ) -> T_References:
  582. """Internal method used to create a new symbolic reference.
  583. If `resolve` is ``False``, the reference will be taken as is, creating a proper
  584. symbolic reference. Otherwise it will be resolved to the corresponding object
  585. and a detached symbolic reference will be created instead.
  586. """
  587. git_dir = _git_dir(repo, path)
  588. full_ref_path = cls.to_full_path(path)
  589. abs_ref_path = os.path.join(git_dir, full_ref_path)
  590. # Figure out target data.
  591. target = reference
  592. if resolve:
  593. target = repo.rev_parse(str(reference))
  594. if not force and os.path.isfile(abs_ref_path):
  595. target_data = str(target)
  596. if isinstance(target, SymbolicReference):
  597. target_data = os.fspath(target.path)
  598. if not resolve:
  599. target_data = "ref: " + target_data
  600. with open(abs_ref_path, "rb") as fd:
  601. existing_data = fd.read().decode(defenc).strip()
  602. if existing_data != target_data:
  603. raise OSError(
  604. "Reference at %r does already exist, pointing to %r, requested was %r"
  605. % (full_ref_path, existing_data, target_data)
  606. )
  607. # END no force handling
  608. ref = cls(repo, full_ref_path)
  609. ref.set_reference(target, logmsg)
  610. return ref
  611. @classmethod
  612. def create(
  613. cls: Type[T_References],
  614. repo: "Repo",
  615. path: PathLike,
  616. reference: Union["SymbolicReference", str] = "HEAD",
  617. logmsg: Union[str, None] = None,
  618. force: bool = False,
  619. **kwargs: Any,
  620. ) -> T_References:
  621. """Create a new symbolic reference: a reference pointing to another reference.
  622. :param repo:
  623. Repository to create the reference in.
  624. :param path:
  625. Full path at which the new symbolic reference is supposed to be created at,
  626. e.g. ``NEW_HEAD`` or ``symrefs/my_new_symref``.
  627. :param reference:
  628. The reference which the new symbolic reference should point to.
  629. If it is a commit-ish, the symbolic ref will be detached.
  630. :param force:
  631. If ``True``, force creation even if a symbolic reference with that name
  632. already exists. Raise :exc:`OSError` otherwise.
  633. :param logmsg:
  634. If not ``None``, the message to append to the reflog.
  635. If ``None``, no reflog entry is written.
  636. :return:
  637. Newly created symbolic reference
  638. :raise OSError:
  639. If a (Symbolic)Reference with the same name but different contents already
  640. exists.
  641. :note:
  642. This does not alter the current HEAD, index or working tree.
  643. """
  644. return cls._create(repo, path, cls._resolve_ref_on_create, reference, force, logmsg)
  645. def rename(self, new_path: PathLike, force: bool = False) -> "SymbolicReference":
  646. """Rename self to a new path.
  647. :param new_path:
  648. Either a simple name or a full path, e.g. ``new_name`` or
  649. ``features/new_name``.
  650. The prefix ``refs/`` is implied for references and will be set as needed.
  651. In case this is a symbolic ref, there is no implied prefix.
  652. :param force:
  653. If ``True``, the rename will succeed even if a head with the target name
  654. already exists. It will be overwritten in that case.
  655. :return:
  656. self
  657. :raise OSError:
  658. If a file at path but with different contents already exists.
  659. """
  660. new_path = self.to_full_path(new_path)
  661. if self.path == new_path:
  662. return self
  663. new_abs_path = os.path.join(_git_dir(self.repo, new_path), new_path)
  664. cur_abs_path = os.path.join(_git_dir(self.repo, self.path), self.path)
  665. if os.path.isfile(new_abs_path):
  666. if not force:
  667. # If they point to the same file, it's not an error.
  668. with open(new_abs_path, "rb") as fd1:
  669. f1 = fd1.read().strip()
  670. with open(cur_abs_path, "rb") as fd2:
  671. f2 = fd2.read().strip()
  672. if f1 != f2:
  673. raise OSError("File at path %r already exists" % new_abs_path)
  674. # else: We could remove ourselves and use the other one, but...
  675. # ...for clarity, we just continue as usual.
  676. # END not force handling
  677. os.remove(new_abs_path)
  678. # END handle existing target file
  679. dname = os.path.dirname(new_abs_path)
  680. if not os.path.isdir(dname):
  681. os.makedirs(dname)
  682. # END create directory
  683. os.rename(cur_abs_path, new_abs_path)
  684. self.path = new_path
  685. return self
  686. @classmethod
  687. def _iter_items(
  688. cls: Type[T_References], repo: "Repo", common_path: Union[PathLike, None] = None
  689. ) -> Iterator[T_References]:
  690. if common_path is None:
  691. common_path = cls._common_path_default
  692. rela_paths = set()
  693. # Walk loose refs.
  694. # Currently we do not follow links.
  695. for root, dirs, files in os.walk(join_path_native(repo.common_dir, common_path)):
  696. if "refs" not in root.split(os.sep): # Skip non-refs subfolders.
  697. refs_id = [d for d in dirs if d == "refs"]
  698. if refs_id:
  699. dirs[0:] = ["refs"]
  700. # END prune non-refs folders
  701. for f in files:
  702. if f == "packed-refs":
  703. continue
  704. abs_path = to_native_path_linux(join_path(root, f))
  705. rela_paths.add(abs_path.replace(to_native_path_linux(repo.common_dir) + "/", ""))
  706. # END for each file in root directory
  707. # END for each directory to walk
  708. # Read packed refs.
  709. for _sha, rela_path in cls._iter_packed_refs(repo):
  710. if rela_path.startswith(os.fspath(common_path)):
  711. rela_paths.add(rela_path)
  712. # END relative path matches common path
  713. # END packed refs reading
  714. # Yield paths in sorted order.
  715. for path in sorted(rela_paths):
  716. try:
  717. yield cls.from_path(repo, path)
  718. except ValueError:
  719. continue
  720. # END for each sorted relative refpath
  721. @classmethod
  722. def iter_items(
  723. cls: Type[T_References],
  724. repo: "Repo",
  725. common_path: Union[PathLike, None] = None,
  726. *args: Any,
  727. **kwargs: Any,
  728. ) -> Iterator[T_References]:
  729. """Find all refs in the repository.
  730. :param repo:
  731. The :class:`~git.repo.base.Repo`.
  732. :param common_path:
  733. Optional keyword argument to the path which is to be shared by all returned
  734. Ref objects.
  735. Defaults to class specific portion if ``None``, ensuring that only refs
  736. suitable for the actual class are returned.
  737. :return:
  738. A list of :class:`SymbolicReference`, each guaranteed to be a symbolic ref
  739. which is not detached and pointing to a valid ref.
  740. The list is lexicographically sorted. The returned objects are instances of
  741. concrete subclasses, such as :class:`~git.refs.head.Head` or
  742. :class:`~git.refs.tag.TagReference`.
  743. """
  744. return (r for r in cls._iter_items(repo, common_path) if r.__class__ is SymbolicReference or not r.is_detached)
  745. @classmethod
  746. def from_path(cls: Type[T_References], repo: "Repo", path: PathLike) -> T_References:
  747. """Make a symbolic reference from a path.
  748. :param path:
  749. Full ``.git``-directory-relative path name to the Reference to instantiate.
  750. :note:
  751. Use :meth:`to_full_path` if you only have a partial path of a known
  752. Reference type.
  753. :return:
  754. Instance of type :class:`~git.refs.reference.Reference`,
  755. :class:`~git.refs.head.Head`, or :class:`~git.refs.tag.Tag`, depending on
  756. the given path.
  757. """
  758. if not path:
  759. raise ValueError("Cannot create Reference from %r" % path)
  760. # Names like HEAD are inserted after the refs module is imported - we have an
  761. # import dependency cycle and don't want to import these names in-function.
  762. from . import HEAD, Head, RemoteReference, TagReference, Reference
  763. for ref_type in (
  764. HEAD,
  765. Head,
  766. RemoteReference,
  767. TagReference,
  768. Reference,
  769. SymbolicReference,
  770. ):
  771. try:
  772. instance = cast(T_References, ref_type(repo, path))
  773. if instance.__class__ is SymbolicReference and instance.is_detached:
  774. raise ValueError("SymbolicRef was detached, we drop it")
  775. else:
  776. return instance
  777. except ValueError:
  778. pass
  779. # END exception handling
  780. # END for each type to try
  781. raise ValueError("Could not find reference type suitable to handle path %r" % path)
  782. def is_remote(self) -> bool:
  783. """:return: True if this symbolic reference points to a remote branch"""
  784. return os.fspath(self.path).startswith(self._remote_common_path_default + "/")