Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/git/remote.py: 21%
435 statements
« prev ^ index » next coverage.py v6.4.4, created at 2023-07-17 14:22 -0600
1# remote.py
2# Copyright (C) 2008, 2009 Michael Trier (mtrier@gmail.com) and contributors
3#
4# This module is part of GitPython and is released under
5# the BSD License: http://www.opensource.org/licenses/bsd-license.php
7# Module implementing a remote object allowing easy access to git remotes
8import logging
9import re
11from git.cmd import handle_process_output, Git
12from git.compat import defenc, force_text
13from git.exc import GitCommandError
14from git.util import (
15 LazyMixin,
16 IterableObj,
17 IterableList,
18 RemoteProgress,
19 CallableRemoteProgress,
20)
21from git.util import (
22 join_path,
23)
25from git.config import (
26 GitConfigParser,
27 SectionConstraint,
28 cp,
29)
30from git.refs import Head, Reference, RemoteReference, SymbolicReference, TagReference
32# typing-------------------------------------------------------
34from typing import (
35 Any,
36 Callable,
37 Dict,
38 Iterator,
39 List,
40 NoReturn,
41 Optional,
42 Sequence,
43 TYPE_CHECKING,
44 Type,
45 Union,
46 cast,
47 overload,
48)
50from git.types import PathLike, Literal, Commit_ish
52if TYPE_CHECKING: 52 ↛ 53line 52 didn't jump to line 53, because the condition on line 52 was never true
53 from git.repo.base import Repo
54 from git.objects.submodule.base import UpdateProgress
56 # from git.objects.commit import Commit
57 # from git.objects import Blob, Tree, TagObject
59flagKeyLiteral = Literal[" ", "!", "+", "-", "*", "=", "t", "?"]
61# def is_flagKeyLiteral(inp: str) -> TypeGuard[flagKeyLiteral]:
62# return inp in [' ', '!', '+', '-', '=', '*', 't', '?']
65# -------------------------------------------------------------
68log = logging.getLogger("git.remote")
69log.addHandler(logging.NullHandler())
72__all__ = ("RemoteProgress", "PushInfo", "FetchInfo", "Remote")
74# { Utilities
77def add_progress(
78 kwargs: Any,
79 git: Git,
80 progress: Union[RemoteProgress, "UpdateProgress", Callable[..., RemoteProgress], None],
81) -> Any:
82 """Add the --progress flag to the given kwargs dict if supported by the
83 git command. If the actual progress in the given progress instance is not
84 given, we do not request any progress
85 :return: possibly altered kwargs"""
86 if progress is not None:
87 v = git.version_info[:2]
88 if v >= (1, 7):
89 kwargs["progress"] = True
90 # END handle --progress
91 # END handle progress
92 return kwargs
95# } END utilities
98@overload
99def to_progress_instance(progress: None) -> RemoteProgress:
100 ...
103@overload
104def to_progress_instance(progress: Callable[..., Any]) -> CallableRemoteProgress:
105 ...
108@overload
109def to_progress_instance(progress: RemoteProgress) -> RemoteProgress:
110 ...
113def to_progress_instance(
114 progress: Union[Callable[..., Any], RemoteProgress, None]
115) -> Union[RemoteProgress, CallableRemoteProgress]:
116 """Given the 'progress' return a suitable object derived from
117 RemoteProgress().
118 """
119 # new API only needs progress as a function
120 if callable(progress):
121 return CallableRemoteProgress(progress)
123 # where None is passed create a parser that eats the progress
124 elif progress is None:
125 return RemoteProgress()
127 # assume its the old API with an instance of RemoteProgress.
128 return progress
131class PushInfo(IterableObj, object):
132 """
133 Carries information about the result of a push operation of a single head::
135 info = remote.push()[0]
136 info.flags # bitflags providing more information about the result
137 info.local_ref # Reference pointing to the local reference that was pushed
138 # It is None if the ref was deleted.
139 info.remote_ref_string # path to the remote reference located on the remote side
140 info.remote_ref # Remote Reference on the local side corresponding to
141 # the remote_ref_string. It can be a TagReference as well.
142 info.old_commit # commit at which the remote_ref was standing before we pushed
143 # it to local_ref.commit. Will be None if an error was indicated
144 info.summary # summary line providing human readable english text about the push
145 """
147 __slots__ = (
148 "local_ref",
149 "remote_ref_string",
150 "flags",
151 "_old_commit_sha",
152 "_remote",
153 "summary",
154 )
155 _id_attribute_ = "pushinfo"
157 (
158 NEW_TAG,
159 NEW_HEAD,
160 NO_MATCH,
161 REJECTED,
162 REMOTE_REJECTED,
163 REMOTE_FAILURE,
164 DELETED,
165 FORCED_UPDATE,
166 FAST_FORWARD,
167 UP_TO_DATE,
168 ERROR,
169 ) = [1 << x for x in range(11)]
171 _flag_map = {
172 "X": NO_MATCH,
173 "-": DELETED,
174 "*": 0,
175 "+": FORCED_UPDATE,
176 " ": FAST_FORWARD,
177 "=": UP_TO_DATE,
178 "!": ERROR,
179 }
181 def __init__(
182 self,
183 flags: int,
184 local_ref: Union[SymbolicReference, None],
185 remote_ref_string: str,
186 remote: "Remote",
187 old_commit: Optional[str] = None,
188 summary: str = "",
189 ) -> None:
190 """Initialize a new instance
191 local_ref: HEAD | Head | RemoteReference | TagReference | Reference | SymbolicReference | None"""
192 self.flags = flags
193 self.local_ref = local_ref
194 self.remote_ref_string = remote_ref_string
195 self._remote = remote
196 self._old_commit_sha = old_commit
197 self.summary = summary
199 @property
200 def old_commit(self) -> Union[str, SymbolicReference, Commit_ish, None]:
201 return self._old_commit_sha and self._remote.repo.commit(self._old_commit_sha) or None
203 @property
204 def remote_ref(self) -> Union[RemoteReference, TagReference]:
205 """
206 :return:
207 Remote Reference or TagReference in the local repository corresponding
208 to the remote_ref_string kept in this instance."""
209 # translate heads to a local remote, tags stay as they are
210 if self.remote_ref_string.startswith("refs/tags"):
211 return TagReference(self._remote.repo, self.remote_ref_string)
212 elif self.remote_ref_string.startswith("refs/heads"):
213 remote_ref = Reference(self._remote.repo, self.remote_ref_string)
214 return RemoteReference(
215 self._remote.repo,
216 "refs/remotes/%s/%s" % (str(self._remote), remote_ref.name),
217 )
218 else:
219 raise ValueError("Could not handle remote ref: %r" % self.remote_ref_string)
220 # END
222 @classmethod
223 def _from_line(cls, remote: "Remote", line: str) -> "PushInfo":
224 """Create a new PushInfo instance as parsed from line which is expected to be like
225 refs/heads/master:refs/heads/master 05d2687..1d0568e as bytes"""
226 control_character, from_to, summary = line.split("\t", 3)
227 flags = 0
229 # control character handling
230 try:
231 flags |= cls._flag_map[control_character]
232 except KeyError as e:
233 raise ValueError("Control character %r unknown as parsed from line %r" % (control_character, line)) from e
234 # END handle control character
236 # from_to handling
237 from_ref_string, to_ref_string = from_to.split(":")
238 if flags & cls.DELETED:
239 from_ref: Union[SymbolicReference, None] = None
240 else:
241 if from_ref_string == "(delete)":
242 from_ref = None
243 else:
244 from_ref = Reference.from_path(remote.repo, from_ref_string)
246 # commit handling, could be message or commit info
247 old_commit: Optional[str] = None
248 if summary.startswith("["):
249 if "[rejected]" in summary:
250 flags |= cls.REJECTED
251 elif "[remote rejected]" in summary:
252 flags |= cls.REMOTE_REJECTED
253 elif "[remote failure]" in summary:
254 flags |= cls.REMOTE_FAILURE
255 elif "[no match]" in summary:
256 flags |= cls.ERROR
257 elif "[new tag]" in summary:
258 flags |= cls.NEW_TAG
259 elif "[new branch]" in summary:
260 flags |= cls.NEW_HEAD
261 # uptodate encoded in control character
262 else:
263 # fast-forward or forced update - was encoded in control character,
264 # but we parse the old and new commit
265 split_token = "..."
266 if control_character == " ":
267 split_token = ".."
268 old_sha, _new_sha = summary.split(" ")[0].split(split_token)
269 # have to use constructor here as the sha usually is abbreviated
270 old_commit = old_sha
271 # END message handling
273 return PushInfo(flags, from_ref, to_ref_string, remote, old_commit, summary)
275 @classmethod
276 def iter_items(cls, repo: "Repo", *args: Any, **kwargs: Any) -> NoReturn: # -> Iterator['PushInfo']:
277 raise NotImplementedError
280class PushInfoList(IterableList[PushInfo]):
281 def __new__(cls) -> "PushInfoList":
282 return cast(PushInfoList, IterableList.__new__(cls, "push_infos"))
284 def __init__(self) -> None:
285 super().__init__("push_infos")
286 self.error: Optional[Exception] = None
288 def raise_if_error(self) -> None:
289 """
290 Raise an exception if any ref failed to push.
291 """
292 if self.error:
293 raise self.error
296class FetchInfo(IterableObj, object):
298 """
299 Carries information about the results of a fetch operation of a single head::
301 info = remote.fetch()[0]
302 info.ref # Symbolic Reference or RemoteReference to the changed
303 # remote head or FETCH_HEAD
304 info.flags # additional flags to be & with enumeration members,
305 # i.e. info.flags & info.REJECTED
306 # is 0 if ref is SymbolicReference
307 info.note # additional notes given by git-fetch intended for the user
308 info.old_commit # if info.flags & info.FORCED_UPDATE|info.FAST_FORWARD,
309 # field is set to the previous location of ref, otherwise None
310 info.remote_ref_path # The path from which we fetched on the remote. It's the remote's version of our info.ref
311 """
313 __slots__ = ("ref", "old_commit", "flags", "note", "remote_ref_path")
314 _id_attribute_ = "fetchinfo"
316 (
317 NEW_TAG,
318 NEW_HEAD,
319 HEAD_UPTODATE,
320 TAG_UPDATE,
321 REJECTED,
322 FORCED_UPDATE,
323 FAST_FORWARD,
324 ERROR,
325 ) = [1 << x for x in range(8)]
327 _re_fetch_result = re.compile(r"^\s*(.) (\[[\w\s\.$@]+\]|[\w\.$@]+)\s+(.+) -> ([^\s]+)( \(.*\)?$)?")
329 _flag_map: Dict[flagKeyLiteral, int] = {
330 "!": ERROR,
331 "+": FORCED_UPDATE,
332 "*": 0,
333 "=": HEAD_UPTODATE,
334 " ": FAST_FORWARD,
335 "-": TAG_UPDATE,
336 }
338 @classmethod
339 def refresh(cls) -> Literal[True]:
340 """This gets called by the refresh function (see the top level
341 __init__).
342 """
343 # clear the old values in _flag_map
344 try:
345 del cls._flag_map["t"]
346 except KeyError:
347 pass
349 try:
350 del cls._flag_map["-"]
351 except KeyError:
352 pass
354 # set the value given the git version
355 if Git().version_info[:2] >= (2, 10): 355 ↛ 358line 355 didn't jump to line 358, because the condition on line 355 was never false
356 cls._flag_map["t"] = cls.TAG_UPDATE
357 else:
358 cls._flag_map["-"] = cls.TAG_UPDATE
360 return True
362 def __init__(
363 self,
364 ref: SymbolicReference,
365 flags: int,
366 note: str = "",
367 old_commit: Union[Commit_ish, None] = None,
368 remote_ref_path: Optional[PathLike] = None,
369 ) -> None:
370 """
371 Initialize a new instance
372 """
373 self.ref = ref
374 self.flags = flags
375 self.note = note
376 self.old_commit = old_commit
377 self.remote_ref_path = remote_ref_path
379 def __str__(self) -> str:
380 return self.name
382 @property
383 def name(self) -> str:
384 """:return: Name of our remote ref"""
385 return self.ref.name
387 @property
388 def commit(self) -> Commit_ish:
389 """:return: Commit of our remote ref"""
390 return self.ref.commit
392 @classmethod
393 def _from_line(cls, repo: "Repo", line: str, fetch_line: str) -> "FetchInfo":
394 """Parse information from the given line as returned by git-fetch -v
395 and return a new FetchInfo object representing this information.
397 We can handle a line as follows:
398 "%c %-\\*s %-\\*s -> %s%s"
400 Where c is either ' ', !, +, -, \\*, or =
401 ! means error
402 + means success forcing update
403 - means a tag was updated
404 * means birth of new branch or tag
405 = means the head was up to date ( and not moved )
406 ' ' means a fast-forward
408 fetch line is the corresponding line from FETCH_HEAD, like
409 acb0fa8b94ef421ad60c8507b634759a472cd56c not-for-merge branch '0.1.7RC' of /tmp/tmpya0vairemote_repo"""
410 match = cls._re_fetch_result.match(line)
411 if match is None:
412 raise ValueError("Failed to parse line: %r" % line)
414 # parse lines
415 remote_local_ref_str: str
416 (
417 control_character,
418 operation,
419 local_remote_ref,
420 remote_local_ref_str,
421 note,
422 ) = match.groups()
423 # assert is_flagKeyLiteral(control_character), f"{control_character}"
424 control_character = cast(flagKeyLiteral, control_character)
425 try:
426 _new_hex_sha, _fetch_operation, fetch_note = fetch_line.split("\t")
427 ref_type_name, fetch_note = fetch_note.split(" ", 1)
428 except ValueError as e: # unpack error
429 raise ValueError("Failed to parse FETCH_HEAD line: %r" % fetch_line) from e
431 # parse flags from control_character
432 flags = 0
433 try:
434 flags |= cls._flag_map[control_character]
435 except KeyError as e:
436 raise ValueError("Control character %r unknown as parsed from line %r" % (control_character, line)) from e
437 # END control char exception handling
439 # parse operation string for more info - makes no sense for symbolic refs, but we parse it anyway
440 old_commit: Union[Commit_ish, None] = None
441 is_tag_operation = False
442 if "rejected" in operation:
443 flags |= cls.REJECTED
444 if "new tag" in operation:
445 flags |= cls.NEW_TAG
446 is_tag_operation = True
447 if "tag update" in operation:
448 flags |= cls.TAG_UPDATE
449 is_tag_operation = True
450 if "new branch" in operation:
451 flags |= cls.NEW_HEAD
452 if "..." in operation or ".." in operation:
453 split_token = "..."
454 if control_character == " ":
455 split_token = split_token[:-1]
456 old_commit = repo.rev_parse(operation.split(split_token)[0])
457 # END handle refspec
459 # handle FETCH_HEAD and figure out ref type
460 # If we do not specify a target branch like master:refs/remotes/origin/master,
461 # the fetch result is stored in FETCH_HEAD which destroys the rule we usually
462 # have. In that case we use a symbolic reference which is detached
463 ref_type: Optional[Type[SymbolicReference]] = None
464 if remote_local_ref_str == "FETCH_HEAD":
465 ref_type = SymbolicReference
466 elif ref_type_name == "tag" or is_tag_operation:
467 # the ref_type_name can be branch, whereas we are still seeing a tag operation. It happens during
468 # testing, which is based on actual git operations
469 ref_type = TagReference
470 elif ref_type_name in ("remote-tracking", "branch"):
471 # note: remote-tracking is just the first part of the 'remote-tracking branch' token.
472 # We don't parse it correctly, but its enough to know what to do, and its new in git 1.7something
473 ref_type = RemoteReference
474 elif "/" in ref_type_name:
475 # If the fetch spec look something like this '+refs/pull/*:refs/heads/pull/*', and is thus pretty
476 # much anything the user wants, we will have trouble to determine what's going on
477 # For now, we assume the local ref is a Head
478 ref_type = Head
479 else:
480 raise TypeError("Cannot handle reference type: %r" % ref_type_name)
481 # END handle ref type
483 # create ref instance
484 if ref_type is SymbolicReference:
485 remote_local_ref = ref_type(repo, "FETCH_HEAD")
486 else:
487 # determine prefix. Tags are usually pulled into refs/tags, they may have subdirectories.
488 # It is not clear sometimes where exactly the item is, unless we have an absolute path as indicated
489 # by the 'ref/' prefix. Otherwise even a tag could be in refs/remotes, which is when it will have the
490 # 'tags/' subdirectory in its path.
491 # We don't want to test for actual existence, but try to figure everything out analytically.
492 ref_path: Optional[PathLike] = None
493 remote_local_ref_str = remote_local_ref_str.strip()
495 if remote_local_ref_str.startswith(Reference._common_path_default + "/"):
496 # always use actual type if we get absolute paths
497 # Will always be the case if something is fetched outside of refs/remotes (if its not a tag)
498 ref_path = remote_local_ref_str
499 if ref_type is not TagReference and not remote_local_ref_str.startswith(
500 RemoteReference._common_path_default + "/"
501 ):
502 ref_type = Reference
503 # END downgrade remote reference
504 elif ref_type is TagReference and "tags/" in remote_local_ref_str:
505 # even though its a tag, it is located in refs/remotes
506 ref_path = join_path(RemoteReference._common_path_default, remote_local_ref_str)
507 else:
508 ref_path = join_path(ref_type._common_path_default, remote_local_ref_str)
509 # END obtain refpath
511 # even though the path could be within the git conventions, we make
512 # sure we respect whatever the user wanted, and disabled path checking
513 remote_local_ref = ref_type(repo, ref_path, check_path=False)
514 # END create ref instance
516 note = (note and note.strip()) or ""
518 return cls(remote_local_ref, flags, note, old_commit, local_remote_ref)
520 @classmethod
521 def iter_items(cls, repo: "Repo", *args: Any, **kwargs: Any) -> NoReturn: # -> Iterator['FetchInfo']:
522 raise NotImplementedError
525class Remote(LazyMixin, IterableObj):
527 """Provides easy read and write access to a git remote.
529 Everything not part of this interface is considered an option for the current
530 remote, allowing constructs like remote.pushurl to query the pushurl.
532 NOTE: When querying configuration, the configuration accessor will be cached
533 to speed up subsequent accesses."""
535 __slots__ = ("repo", "name", "_config_reader")
536 _id_attribute_ = "name"
538 def __init__(self, repo: "Repo", name: str) -> None:
539 """Initialize a remote instance
541 :param repo: The repository we are a remote of
542 :param name: the name of the remote, i.e. 'origin'"""
543 self.repo = repo
544 self.name = name
545 self.url: str
547 def __getattr__(self, attr: str) -> Any:
548 """Allows to call this instance like
549 remote.special( \\*args, \\*\\*kwargs) to call git-remote special self.name"""
550 if attr == "_config_reader":
551 return super(Remote, self).__getattr__(attr)
553 # sometimes, probably due to a bug in python itself, we are being called
554 # even though a slot of the same name exists
555 try:
556 return self._config_reader.get(attr)
557 except cp.NoOptionError:
558 return super(Remote, self).__getattr__(attr)
559 # END handle exception
561 def _config_section_name(self) -> str:
562 return 'remote "%s"' % self.name
564 def _set_cache_(self, attr: str) -> None:
565 if attr == "_config_reader":
566 # NOTE: This is cached as __getattr__ is overridden to return remote config values implicitly, such as
567 # in print(r.pushurl)
568 self._config_reader = SectionConstraint(self.repo.config_reader("repository"), self._config_section_name())
569 else:
570 super(Remote, self)._set_cache_(attr)
572 def __str__(self) -> str:
573 return self.name
575 def __repr__(self) -> str:
576 return '<git.%s "%s">' % (self.__class__.__name__, self.name)
578 def __eq__(self, other: object) -> bool:
579 return isinstance(other, type(self)) and self.name == other.name
581 def __ne__(self, other: object) -> bool:
582 return not (self == other)
584 def __hash__(self) -> int:
585 return hash(self.name)
587 def exists(self) -> bool:
588 """
589 :return: True if this is a valid, existing remote.
590 Valid remotes have an entry in the repository's configuration"""
591 try:
592 self.config_reader.get("url")
593 return True
594 except cp.NoOptionError:
595 # we have the section at least ...
596 return True
597 except cp.NoSectionError:
598 return False
599 # end
601 @classmethod
602 def iter_items(cls, repo: "Repo", *args: Any, **kwargs: Any) -> Iterator["Remote"]:
603 """:return: Iterator yielding Remote objects of the given repository"""
604 for section in repo.config_reader("repository").sections():
605 if not section.startswith("remote "):
606 continue
607 lbound = section.find('"')
608 rbound = section.rfind('"')
609 if lbound == -1 or rbound == -1:
610 raise ValueError("Remote-Section has invalid format: %r" % section)
611 yield Remote(repo, section[lbound + 1 : rbound])
612 # END for each configuration section
614 def set_url(self, new_url: str, old_url: Optional[str] = None, **kwargs: Any) -> "Remote":
615 """Configure URLs on current remote (cf command git remote set_url)
617 This command manages URLs on the remote.
619 :param new_url: string being the URL to add as an extra remote URL
620 :param old_url: when set, replaces this URL with new_url for the remote
621 :return: self
622 """
623 scmd = "set-url"
624 kwargs["insert_kwargs_after"] = scmd
625 if old_url:
626 self.repo.git.remote(scmd, self.name, new_url, old_url, **kwargs)
627 else:
628 self.repo.git.remote(scmd, self.name, new_url, **kwargs)
629 return self
631 def add_url(self, url: str, **kwargs: Any) -> "Remote":
632 """Adds a new url on current remote (special case of git remote set_url)
634 This command adds new URLs to a given remote, making it possible to have
635 multiple URLs for a single remote.
637 :param url: string being the URL to add as an extra remote URL
638 :return: self
639 """
640 return self.set_url(url, add=True)
642 def delete_url(self, url: str, **kwargs: Any) -> "Remote":
643 """Deletes a new url on current remote (special case of git remote set_url)
645 This command deletes new URLs to a given remote, making it possible to have
646 multiple URLs for a single remote.
648 :param url: string being the URL to delete from the remote
649 :return: self
650 """
651 return self.set_url(url, delete=True)
653 @property
654 def urls(self) -> Iterator[str]:
655 """:return: Iterator yielding all configured URL targets on a remote as strings"""
656 try:
657 remote_details = self.repo.git.remote("get-url", "--all", self.name)
658 assert isinstance(remote_details, str)
659 for line in remote_details.split("\n"):
660 yield line
661 except GitCommandError as ex:
662 ## We are on git < 2.7 (i.e TravisCI as of Oct-2016),
663 # so `get-utl` command does not exist yet!
664 # see: https://github.com/gitpython-developers/GitPython/pull/528#issuecomment-252976319
665 # and: http://stackoverflow.com/a/32991784/548792
666 #
667 if "Unknown subcommand: get-url" in str(ex):
668 try:
669 remote_details = self.repo.git.remote("show", self.name)
670 assert isinstance(remote_details, str)
671 for line in remote_details.split("\n"):
672 if " Push URL:" in line:
673 yield line.split(": ")[-1]
674 except GitCommandError as _ex:
675 if any(msg in str(_ex) for msg in ["correct access rights", "cannot run ssh"]):
676 # If ssh is not setup to access this repository, see issue 694
677 remote_details = self.repo.git.config("--get-all", "remote.%s.url" % self.name)
678 assert isinstance(remote_details, str)
679 for line in remote_details.split("\n"):
680 yield line
681 else:
682 raise _ex
683 else:
684 raise ex
686 @property
687 def refs(self) -> IterableList[RemoteReference]:
688 """
689 :return:
690 IterableList of RemoteReference objects. It is prefixed, allowing
691 you to omit the remote path portion, i.e.::
692 remote.refs.master # yields RemoteReference('/refs/remotes/origin/master')"""
693 out_refs: IterableList[RemoteReference] = IterableList(RemoteReference._id_attribute_, "%s/" % self.name)
694 out_refs.extend(RemoteReference.list_items(self.repo, remote=self.name))
695 return out_refs
697 @property
698 def stale_refs(self) -> IterableList[Reference]:
699 """
700 :return:
701 IterableList RemoteReference objects that do not have a corresponding
702 head in the remote reference anymore as they have been deleted on the
703 remote side, but are still available locally.
705 The IterableList is prefixed, hence the 'origin' must be omitted. See
706 'refs' property for an example.
708 To make things more complicated, it can be possible for the list to include
709 other kinds of references, for example, tag references, if these are stale
710 as well. This is a fix for the issue described here:
711 https://github.com/gitpython-developers/GitPython/issues/260
712 """
713 out_refs: IterableList[Reference] = IterableList(RemoteReference._id_attribute_, "%s/" % self.name)
714 for line in self.repo.git.remote("prune", "--dry-run", self).splitlines()[2:]:
715 # expecting
716 # * [would prune] origin/new_branch
717 token = " * [would prune] "
718 if not line.startswith(token):
719 continue
720 ref_name = line.replace(token, "")
721 # sometimes, paths start with a full ref name, like refs/tags/foo, see #260
722 if ref_name.startswith(Reference._common_path_default + "/"):
723 out_refs.append(Reference.from_path(self.repo, ref_name))
724 else:
725 fqhn = "%s/%s" % (RemoteReference._common_path_default, ref_name)
726 out_refs.append(RemoteReference(self.repo, fqhn))
727 # end special case handling
728 # END for each line
729 return out_refs
731 @classmethod
732 def create(cls, repo: "Repo", name: str, url: str, **kwargs: Any) -> "Remote":
733 """Create a new remote to the given repository
734 :param repo: Repository instance that is to receive the new remote
735 :param name: Desired name of the remote
736 :param url: URL which corresponds to the remote's name
737 :param kwargs: Additional arguments to be passed to the git-remote add command
738 :return: New Remote instance
739 :raise GitCommandError: in case an origin with that name already exists"""
740 scmd = "add"
741 kwargs["insert_kwargs_after"] = scmd
742 repo.git.remote(scmd, name, Git.polish_url(url), **kwargs)
743 return cls(repo, name)
745 # add is an alias
746 @classmethod
747 def add(cls, repo: "Repo", name: str, url: str, **kwargs: Any) -> "Remote":
748 return cls.create(repo, name, url, **kwargs)
750 @classmethod
751 def remove(cls, repo: "Repo", name: str) -> str:
752 """Remove the remote with the given name
753 :return: the passed remote name to remove
754 """
755 repo.git.remote("rm", name)
756 if isinstance(name, cls):
757 name._clear_cache()
758 return name
760 # alias
761 rm = remove
763 def rename(self, new_name: str) -> "Remote":
764 """Rename self to the given new_name
765 :return: self"""
766 if self.name == new_name:
767 return self
769 self.repo.git.remote("rename", self.name, new_name)
770 self.name = new_name
771 self._clear_cache()
773 return self
775 def update(self, **kwargs: Any) -> "Remote":
776 """Fetch all changes for this remote, including new branches which will
777 be forced in ( in case your local remote branch is not part the new remote branches
778 ancestry anymore ).
780 :param kwargs:
781 Additional arguments passed to git-remote update
783 :return: self"""
784 scmd = "update"
785 kwargs["insert_kwargs_after"] = scmd
786 self.repo.git.remote(scmd, self.name, **kwargs)
787 return self
789 def _get_fetch_info_from_stderr(
790 self,
791 proc: "Git.AutoInterrupt",
792 progress: Union[Callable[..., Any], RemoteProgress, None],
793 kill_after_timeout: Union[None, float] = None,
794 ) -> IterableList["FetchInfo"]:
796 progress = to_progress_instance(progress)
798 # skip first line as it is some remote info we are not interested in
799 output: IterableList["FetchInfo"] = IterableList("name")
801 # lines which are no progress are fetch info lines
802 # this also waits for the command to finish
803 # Skip some progress lines that don't provide relevant information
804 fetch_info_lines = []
805 # Basically we want all fetch info lines which appear to be in regular form, and thus have a
806 # command character. Everything else we ignore,
807 cmds = set(FetchInfo._flag_map.keys())
809 progress_handler = progress.new_message_handler()
810 handle_process_output(
811 proc,
812 None,
813 progress_handler,
814 finalizer=None,
815 decode_streams=False,
816 kill_after_timeout=kill_after_timeout,
817 )
819 stderr_text = progress.error_lines and "\n".join(progress.error_lines) or ""
820 proc.wait(stderr=stderr_text)
821 if stderr_text:
822 log.warning("Error lines received while fetching: %s", stderr_text)
824 for line in progress.other_lines:
825 line = force_text(line)
826 for cmd in cmds:
827 if len(line) > 1 and line[0] == " " and line[1] == cmd:
828 fetch_info_lines.append(line)
829 continue
831 # read head information
832 fetch_head = SymbolicReference(self.repo, "FETCH_HEAD")
833 with open(fetch_head.abspath, "rb") as fp:
834 fetch_head_info = [line.decode(defenc) for line in fp.readlines()]
836 l_fil = len(fetch_info_lines)
837 l_fhi = len(fetch_head_info)
838 if l_fil != l_fhi:
839 msg = "Fetch head lines do not match lines provided via progress information\n"
840 msg += "length of progress lines %i should be equal to lines in FETCH_HEAD file %i\n"
841 msg += "Will ignore extra progress lines or fetch head lines."
842 msg %= (l_fil, l_fhi)
843 log.debug(msg)
844 log.debug(b"info lines: " + str(fetch_info_lines).encode("UTF-8"))
845 log.debug(b"head info: " + str(fetch_head_info).encode("UTF-8"))
846 if l_fil < l_fhi:
847 fetch_head_info = fetch_head_info[:l_fil]
848 else:
849 fetch_info_lines = fetch_info_lines[:l_fhi]
850 # end truncate correct list
851 # end sanity check + sanitization
853 for err_line, fetch_line in zip(fetch_info_lines, fetch_head_info):
854 try:
855 output.append(FetchInfo._from_line(self.repo, err_line, fetch_line))
856 except ValueError as exc:
857 log.debug("Caught error while parsing line: %s", exc)
858 log.warning("Git informed while fetching: %s", err_line.strip())
859 return output
861 def _get_push_info(
862 self,
863 proc: "Git.AutoInterrupt",
864 progress: Union[Callable[..., Any], RemoteProgress, None],
865 kill_after_timeout: Union[None, float] = None,
866 ) -> PushInfoList:
867 progress = to_progress_instance(progress)
869 # read progress information from stderr
870 # we hope stdout can hold all the data, it should ...
871 # read the lines manually as it will use carriage returns between the messages
872 # to override the previous one. This is why we read the bytes manually
873 progress_handler = progress.new_message_handler()
874 output: PushInfoList = PushInfoList()
876 def stdout_handler(line: str) -> None:
877 try:
878 output.append(PushInfo._from_line(self, line))
879 except ValueError:
880 # If an error happens, additional info is given which we parse below.
881 pass
883 handle_process_output(
884 proc,
885 stdout_handler,
886 progress_handler,
887 finalizer=None,
888 decode_streams=False,
889 kill_after_timeout=kill_after_timeout,
890 )
891 stderr_text = progress.error_lines and "\n".join(progress.error_lines) or ""
892 try:
893 proc.wait(stderr=stderr_text)
894 except Exception as e:
895 # This is different than fetch (which fails if there is any std_err
896 # even if there is an output)
897 if not output:
898 raise
899 elif stderr_text:
900 log.warning("Error lines received while fetching: %s", stderr_text)
901 output.error = e
903 return output
905 def _assert_refspec(self) -> None:
906 """Turns out we can't deal with remotes if the refspec is missing"""
907 config = self.config_reader
908 unset = "placeholder"
909 try:
910 if config.get_value("fetch", default=unset) is unset:
911 msg = "Remote '%s' has no refspec set.\n"
912 msg += "You can set it as follows:"
913 msg += " 'git config --add \"remote.%s.fetch +refs/heads/*:refs/heads/*\"'."
914 raise AssertionError(msg % (self.name, self.name))
915 finally:
916 config.release()
918 def fetch(
919 self,
920 refspec: Union[str, List[str], None] = None,
921 progress: Union[RemoteProgress, None, "UpdateProgress"] = None,
922 verbose: bool = True,
923 kill_after_timeout: Union[None, float] = None,
924 **kwargs: Any,
925 ) -> IterableList[FetchInfo]:
926 """Fetch the latest changes for this remote
928 :param refspec:
929 A "refspec" is used by fetch and push to describe the mapping
930 between remote ref and local ref. They are combined with a colon in
931 the format <src>:<dst>, preceded by an optional plus sign, +.
932 For example: git fetch $URL refs/heads/master:refs/heads/origin means
933 "grab the master branch head from the $URL and store it as my origin
934 branch head". And git push $URL refs/heads/master:refs/heads/to-upstream
935 means "publish my master branch head as to-upstream branch at $URL".
936 See also git-push(1).
938 Taken from the git manual
940 Fetch supports multiple refspecs (as the
941 underlying git-fetch does) - supplying a list rather than a string
942 for 'refspec' will make use of this facility.
943 :param progress: See 'push' method
944 :param verbose: Boolean for verbose output
945 :param kill_after_timeout:
946 To specify a timeout in seconds for the git command, after which the process
947 should be killed. It is set to None by default.
948 :param kwargs: Additional arguments to be passed to git-fetch
949 :return:
950 IterableList(FetchInfo, ...) list of FetchInfo instances providing detailed
951 information about the fetch results
953 :note:
954 As fetch does not provide progress information to non-ttys, we cannot make
955 it available here unfortunately as in the 'push' method."""
956 if refspec is None:
957 # No argument refspec, then ensure the repo's config has a fetch refspec.
958 self._assert_refspec()
960 kwargs = add_progress(kwargs, self.repo.git, progress)
961 if isinstance(refspec, list):
962 args: Sequence[Optional[str]] = refspec
963 else:
964 args = [refspec]
966 proc = self.repo.git.fetch(
967 self, *args, as_process=True, with_stdout=False, universal_newlines=True, v=verbose, **kwargs
968 )
969 res = self._get_fetch_info_from_stderr(proc, progress, kill_after_timeout=kill_after_timeout)
970 if hasattr(self.repo.odb, "update_cache"):
971 self.repo.odb.update_cache()
972 return res
974 def pull(
975 self,
976 refspec: Union[str, List[str], None] = None,
977 progress: Union[RemoteProgress, "UpdateProgress", None] = None,
978 kill_after_timeout: Union[None, float] = None,
979 **kwargs: Any,
980 ) -> IterableList[FetchInfo]:
981 """Pull changes from the given branch, being the same as a fetch followed
982 by a merge of branch with your local branch.
984 :param refspec: see 'fetch' method
985 :param progress: see 'push' method
986 :param kill_after_timeout: see 'fetch' method
987 :param kwargs: Additional arguments to be passed to git-pull
988 :return: Please see 'fetch' method"""
989 if refspec is None:
990 # No argument refspec, then ensure the repo's config has a fetch refspec.
991 self._assert_refspec()
992 kwargs = add_progress(kwargs, self.repo.git, progress)
993 proc = self.repo.git.pull(
994 self, refspec, with_stdout=False, as_process=True, universal_newlines=True, v=True, **kwargs
995 )
996 res = self._get_fetch_info_from_stderr(proc, progress, kill_after_timeout=kill_after_timeout)
997 if hasattr(self.repo.odb, "update_cache"):
998 self.repo.odb.update_cache()
999 return res
1001 def push(
1002 self,
1003 refspec: Union[str, List[str], None] = None,
1004 progress: Union[RemoteProgress, "UpdateProgress", Callable[..., RemoteProgress], None] = None,
1005 kill_after_timeout: Union[None, float] = None,
1006 **kwargs: Any,
1007 ) -> IterableList[PushInfo]:
1008 """Push changes from source branch in refspec to target branch in refspec.
1010 :param refspec: see 'fetch' method
1011 :param progress:
1012 Can take one of many value types:
1014 * None to discard progress information
1015 * A function (callable) that is called with the progress information.
1016 Signature: ``progress(op_code, cur_count, max_count=None, message='')``.
1017 `Click here <http://goo.gl/NPa7st>`__ for a description of all arguments
1018 given to the function.
1019 * An instance of a class derived from ``git.RemoteProgress`` that
1020 overrides the ``update()`` function.
1022 :note: No further progress information is returned after push returns.
1023 :param kill_after_timeout:
1024 To specify a timeout in seconds for the git command, after which the process
1025 should be killed. It is set to None by default.
1026 :param kwargs: Additional arguments to be passed to git-push
1027 :return:
1028 list(PushInfo, ...) list of PushInfo instances, each
1029 one informing about an individual head which had been updated on the remote
1030 side.
1031 If the push contains rejected heads, these will have the PushInfo.ERROR bit set
1032 in their flags.
1033 If the operation fails completely, the length of the returned IterableList will
1034 be 0."""
1035 kwargs = add_progress(kwargs, self.repo.git, progress)
1036 proc = self.repo.git.push(
1037 self,
1038 refspec,
1039 porcelain=True,
1040 as_process=True,
1041 universal_newlines=True,
1042 kill_after_timeout=kill_after_timeout,
1043 **kwargs,
1044 )
1045 return self._get_push_info(proc, progress, kill_after_timeout=kill_after_timeout)
1047 @property
1048 def config_reader(self) -> SectionConstraint[GitConfigParser]:
1049 """
1050 :return:
1051 GitConfigParser compatible object able to read options for only our remote.
1052 Hence you may simple type config.get("pushurl") to obtain the information"""
1053 return self._config_reader
1055 def _clear_cache(self) -> None:
1056 try:
1057 del self._config_reader
1058 except AttributeError:
1059 pass
1060 # END handle exception
1062 @property
1063 def config_writer(self) -> SectionConstraint:
1064 """
1065 :return: GitConfigParser compatible object able to write options for this remote.
1066 :note:
1067 You can only own one writer at a time - delete it to release the
1068 configuration file and make it usable by others.
1070 To assure consistent results, you should only query options through the
1071 writer. Once you are done writing, you are free to use the config reader
1072 once again."""
1073 writer = self.repo.config_writer()
1075 # clear our cache to assure we re-read the possibly changed configuration
1076 self._clear_cache()
1077 return SectionConstraint(writer, self._config_section_name())