mirror of
https://github.com/Stone-Red-Code/dvc.git
synced 2026-11-06 07:18:28 +01:00
git: add support for working with refspecs and stashes (#4952)
* git: add support for dulwich Stash * git: start adding support for custom refs * git: add Stash and helper context managers for preserving workspace state * scm: cleanup refs/stash and add push_refspec * tests: add tests for git refs and stash * fix push_refspec * tests: add tests for git push_refspec * scm: add git fetch-refspecs * add test for fetch refspecs * scm: add test for git refs containing * remove extraneous log msgs * fix typing issues * stash: only try to push if workspace is dirty * update dulwich to 0.20.14 * scm: move dulwich into scm.backend * move stash internals to backends * scm: move git code into dvc/scm/git/... module * git: use ABC for BaseGitBackend * fix rebase issue * git: fix refspec tests on windows - git file:// URLs need posix style paths * git: move get_refs_containing to backend * scm: add callback for handling diverged refs during fetch
This commit is contained in:
Peter Rowlands (변기호)
@@ -3,7 +3,8 @@
|
||||
import logging
|
||||
import os
|
||||
import shlex
|
||||
from functools import partial
|
||||
from contextlib import contextmanager
|
||||
from functools import partial, partialmethod
|
||||
from typing import Optional
|
||||
|
||||
from funcy import cached_property, first
|
||||
@@ -22,6 +23,11 @@ from dvc.utils import fix_env, is_binary, relpath
|
||||
from dvc.utils.fs import path_isin
|
||||
from dvc.utils.serialize import modify_yaml
|
||||
|
||||
from .backend.base import NoGitBackendError
|
||||
from .backend.dulwich import DulwichBackend
|
||||
from .backend.gitpython import GitPythonBackend
|
||||
from .stash import Stash
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@@ -62,6 +68,8 @@ class Git(Base):
|
||||
|
||||
GITIGNORE = ".gitignore"
|
||||
GIT_DIR = ".git"
|
||||
DEFAULT_BACKENDS = (DulwichBackend, GitPythonBackend)
|
||||
LOCAL_BRANCH_PREFIX = "refs/heads/"
|
||||
|
||||
def __init__(self, root_dir=os.curdir, search_parent_directories=True):
|
||||
"""Git class constructor.
|
||||
@@ -89,10 +97,16 @@ class Git(Base):
|
||||
self.ignored_paths = []
|
||||
self.files_to_track = set()
|
||||
|
||||
self.backends = [backend(self) for backend in self.DEFAULT_BACKENDS]
|
||||
|
||||
@property
|
||||
def root_dir(self) -> str:
|
||||
return self.repo.working_tree_dir
|
||||
|
||||
@cached_property
|
||||
def stash(self):
|
||||
return Stash(self)
|
||||
|
||||
@staticmethod
|
||||
def clone(url, to_path, rev=None, shallow_branch=None):
|
||||
import git
|
||||
@@ -183,14 +197,6 @@ class Git(Base):
|
||||
|
||||
return entry, gitignore
|
||||
|
||||
def is_ignored(self, path):
|
||||
from dulwich import ignore
|
||||
from dulwich.repo import Repo
|
||||
|
||||
repo = Repo(self.root_dir)
|
||||
manager = ignore.IgnoreFilterManager.from_repo(repo)
|
||||
return manager.is_ignored(relpath(path, self.root_dir))
|
||||
|
||||
def ignore(self, path):
|
||||
entry, gitignore = self._get_gitignore(path)
|
||||
|
||||
@@ -257,11 +263,11 @@ class Git(Base):
|
||||
def commit(self, msg):
|
||||
self.repo.index.commit(msg)
|
||||
|
||||
def checkout(self, branch, create_new=False):
|
||||
def checkout(self, branch, create_new=False, **kwargs):
|
||||
if create_new:
|
||||
self.repo.git.checkout("HEAD", b=branch)
|
||||
self.repo.git.checkout("HEAD", b=branch, **kwargs)
|
||||
else:
|
||||
self.repo.git.checkout(branch)
|
||||
self.repo.git.checkout(branch, **kwargs)
|
||||
|
||||
def pull(self, **kwargs):
|
||||
infos = self.repo.remote().pull(**kwargs)
|
||||
@@ -448,6 +454,8 @@ class Git(Base):
|
||||
return False
|
||||
|
||||
def close(self):
|
||||
for backend in self.backends:
|
||||
backend.close()
|
||||
self.repo.close()
|
||||
|
||||
@cached_property
|
||||
@@ -491,3 +499,65 @@ class Git(Base):
|
||||
if isinstance(commit, TagObject):
|
||||
commit = commit.object
|
||||
return commit
|
||||
|
||||
def _backend_func(self, name, *args, **kwargs):
|
||||
for backend in self.backends:
|
||||
try:
|
||||
func = getattr(backend, name)
|
||||
return func(*args, **kwargs)
|
||||
except (AttributeError, NotImplementedError):
|
||||
pass
|
||||
raise NoGitBackendError(name)
|
||||
|
||||
is_ignored = partialmethod(_backend_func, "is_ignored")
|
||||
set_ref = partialmethod(_backend_func, "set_ref")
|
||||
get_ref = partialmethod(_backend_func, "get_ref")
|
||||
remove_ref = partialmethod(_backend_func, "remove_ref")
|
||||
get_refs_containing = partialmethod(_backend_func, "get_refs_containing")
|
||||
push_refspec = partialmethod(_backend_func, "push_refspec")
|
||||
fetch_refspecs = partialmethod(_backend_func, "fetch_refspecs")
|
||||
_stash_iter = partialmethod(_backend_func, "_stash_iter")
|
||||
_stash_push = partialmethod(_backend_func, "_stash_push")
|
||||
_stash_apply = partialmethod(_backend_func, "_stash_apply")
|
||||
reflog_delete = partialmethod(_backend_func, "reflog_delete")
|
||||
|
||||
@contextmanager
|
||||
def detach_head(self, rev: Optional[str] = None):
|
||||
"""Context manager for performing detached HEAD SCM operations.
|
||||
|
||||
Detaches and restores HEAD similar to interactive git rebase.
|
||||
Restore is equivalent to 'reset --soft', meaning the caller is
|
||||
is responsible for preserving & restoring working tree state
|
||||
(i.e. via stash) when applicable.
|
||||
|
||||
Yields revision of detached head.
|
||||
"""
|
||||
if not rev:
|
||||
rev = "HEAD"
|
||||
orig_head = self.get_ref("HEAD", follow=False)
|
||||
logger.debug("Detaching HEAD at '%s'", rev)
|
||||
self.checkout(rev, detach=True)
|
||||
try:
|
||||
yield self.get_ref("HEAD")
|
||||
finally:
|
||||
prefix = self.LOCAL_BRANCH_PREFIX
|
||||
if orig_head.startswith(prefix):
|
||||
orig_head = orig_head[len(prefix) :]
|
||||
logger.debug("Restore HEAD to '%s'", orig_head)
|
||||
self.checkout(orig_head)
|
||||
|
||||
@contextmanager
|
||||
def stash_workspace(self, **kwargs):
|
||||
"""Stash and restore any workspace changes.
|
||||
|
||||
Yields revision of the stash commit. Yields None if there were no
|
||||
changes to stash.
|
||||
"""
|
||||
logger.debug("Stashing workspace")
|
||||
rev = self.stash.push(**kwargs)
|
||||
try:
|
||||
yield rev
|
||||
finally:
|
||||
if rev:
|
||||
logger.debug("Restoring stashed workspace")
|
||||
self.stash.pop()
|
||||
@@ -0,0 +1,123 @@
|
||||
import os
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import TYPE_CHECKING, Callable, Iterable, Optional, Tuple
|
||||
|
||||
from dvc.scm.base import SCMError
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from dvc.scm.git import Git
|
||||
|
||||
|
||||
class NoGitBackendError(SCMError):
|
||||
def __init__(self, func):
|
||||
super().__init__(f"No valid Git backend for '{func}'")
|
||||
|
||||
|
||||
class BaseGitBackend(ABC):
|
||||
"""Base Git backend class."""
|
||||
|
||||
def __init__(self, scm: "Git", **kwargs):
|
||||
self.scm = scm
|
||||
self.root_dir = os.fspath(scm.root_dir)
|
||||
|
||||
def close(self):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def is_ignored(self, path):
|
||||
"""Return True if the specified path is gitignored."""
|
||||
|
||||
@abstractmethod
|
||||
def set_ref(
|
||||
self,
|
||||
name: str,
|
||||
new_ref: str,
|
||||
old_ref: Optional[str] = None,
|
||||
message: Optional[str] = None,
|
||||
symbolic: Optional[bool] = False,
|
||||
):
|
||||
"""Set the specified git ref.
|
||||
|
||||
Optional kwargs:
|
||||
old_ref: If specified, ref will only be set if it currently equals
|
||||
old_ref. Has no effect is symbolic is True.
|
||||
message: Optional reflog message.
|
||||
symbolic: If True, ref will be set as a symbolic ref to new_ref
|
||||
rather than the dereferenced object.
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def get_ref(self, name, follow: Optional[bool] = True) -> Optional[str]:
|
||||
"""Return the value of specified ref.
|
||||
|
||||
If follow is false, symbolic refs will not be dereferenced.
|
||||
Returns None if the ref does not exist.
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def remove_ref(self, name: str, old_ref: Optional[str] = None):
|
||||
"""Remove the specified git ref.
|
||||
|
||||
If old_ref is specified, ref will only be removed if it currently
|
||||
equals old_ref.
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def get_refs_containing(self, rev: str, pattern: Optional[str] = None):
|
||||
"""Iterate over all git refs containing the specfied revision."""
|
||||
|
||||
@abstractmethod
|
||||
def push_refspec(self, url: str, src: Optional[str], dest: str):
|
||||
"""Push refspec to a remote Git repo.
|
||||
|
||||
Args:
|
||||
url: Remote repo Git URL (Note this must be a Git URL and not
|
||||
a remote name).
|
||||
src: Local refspec. If src ends with "/" it will be treated as a
|
||||
prefix, and all refs inside src will be pushed using dest
|
||||
as destination refspec prefix. If src is None, dest will be
|
||||
deleted from the remote.
|
||||
dest: Remote refspec.
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def fetch_refspecs(
|
||||
self,
|
||||
url: str,
|
||||
refspecs: Iterable[str],
|
||||
force: Optional[bool] = False,
|
||||
on_diverged: Optional[Callable[[bytes, bytes], bool]] = None,
|
||||
):
|
||||
"""Fetch refspecs from a remote Git repo.
|
||||
|
||||
Args:
|
||||
url: Remote repo Git URL (Note this must be a Git URL and not
|
||||
a remote name).
|
||||
refspecs: Iterable containing refspecs to fetch.
|
||||
Note that this will not match subkeys.
|
||||
force: If True, local refs will be overwritten.
|
||||
on_diverged: Callback function which will be called if local ref
|
||||
and remote have diverged and force is False. If the callback
|
||||
returns True the local ref will be overwritten.
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def _stash_iter(self, ref: str):
|
||||
"""Iterate over stash commits in the specified ref."""
|
||||
|
||||
@abstractmethod
|
||||
def _stash_push(
|
||||
self,
|
||||
ref: str,
|
||||
message: Optional[str] = None,
|
||||
include_untracked: Optional[bool] = False,
|
||||
) -> Tuple[Optional[str], bool]:
|
||||
"""Push a commit onto the specified stash."""
|
||||
|
||||
@abstractmethod
|
||||
def _stash_apply(self, rev: str):
|
||||
"""Apply the specified stash revision."""
|
||||
|
||||
@abstractmethod
|
||||
def reflog_delete(self, ref: str, updateref: bool = False):
|
||||
"""Delete the specified reflog entry."""
|
||||
@@ -0,0 +1,208 @@
|
||||
import logging
|
||||
import os
|
||||
from typing import TYPE_CHECKING, Callable, Iterable, Optional, Tuple
|
||||
|
||||
from dvc.scm.base import SCMError
|
||||
from dvc.utils import relpath
|
||||
|
||||
from .base import BaseGitBackend
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from dvc.scm.git import Git
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class DulwichBackend(BaseGitBackend): # pylint:disable=abstract-method
|
||||
"""Dulwich Git backend."""
|
||||
|
||||
def __init__(self, scm: "Git", **kwargs):
|
||||
from dulwich.repo import Repo
|
||||
|
||||
super().__init__(scm, **kwargs)
|
||||
self.repo = Repo(self.root_dir)
|
||||
self._stashes: dict = {}
|
||||
|
||||
def close(self):
|
||||
self.repo.close()
|
||||
|
||||
def _get_stash(self, ref: str):
|
||||
from dulwich.stash import Stash as DulwichStash
|
||||
|
||||
if ref not in self._stashes:
|
||||
self._stashes[ref] = DulwichStash(self.repo, ref=os.fsencode(ref))
|
||||
return self._stashes[ref]
|
||||
|
||||
def is_ignored(self, path):
|
||||
from dulwich import ignore
|
||||
|
||||
manager = ignore.IgnoreFilterManager.from_repo(self.repo)
|
||||
return manager.is_ignored(relpath(path, self.root_dir))
|
||||
|
||||
def set_ref(
|
||||
self,
|
||||
name: str,
|
||||
new_ref: str,
|
||||
old_ref: Optional[str] = None,
|
||||
message: Optional[str] = None,
|
||||
symbolic: Optional[bool] = False,
|
||||
):
|
||||
name_b = os.fsencode(name)
|
||||
new_ref_b = os.fsencode(new_ref)
|
||||
old_ref_b = os.fsencode(old_ref) if old_ref else None
|
||||
message_b = message.encode("utf-8") if message else None
|
||||
if symbolic:
|
||||
return self.repo.refs.set_symbolic_ref(
|
||||
name_b, new_ref_b, message=message
|
||||
)
|
||||
if not self.repo.refs.set_if_equals(
|
||||
name_b, old_ref_b, new_ref_b, message=message_b
|
||||
):
|
||||
raise SCMError(f"Failed to set '{name}'")
|
||||
|
||||
def get_ref(self, name, follow: Optional[bool] = True) -> Optional[str]:
|
||||
from dulwich.refs import parse_symref_value
|
||||
|
||||
name_b = os.fsencode(name)
|
||||
if follow:
|
||||
try:
|
||||
ref = self.repo.refs[name_b]
|
||||
except KeyError:
|
||||
ref = None
|
||||
else:
|
||||
ref = self.repo.refs.read_ref(name_b)
|
||||
try:
|
||||
if ref:
|
||||
ref = parse_symref_value(ref)
|
||||
except ValueError:
|
||||
pass
|
||||
if ref:
|
||||
return os.fsdecode(ref)
|
||||
return None
|
||||
|
||||
def remove_ref(self, name: str, old_ref: Optional[str] = None):
|
||||
name_b = name.encode("utf-8")
|
||||
old_ref_b = old_ref.encode("utf-8") if old_ref else None
|
||||
if not self.repo.refs.remove_if_equals(name_b, old_ref_b):
|
||||
raise SCMError(f"Failed to remove '{name}'")
|
||||
|
||||
def get_refs_containing(self, rev: str, pattern: Optional[str] = None):
|
||||
raise NotImplementedError
|
||||
|
||||
def push_refspec(self, url: str, src: Optional[str], dest: str):
|
||||
from dulwich.client import get_transport_and_path
|
||||
from dulwich.objects import ZERO_SHA
|
||||
|
||||
if src is not None and src.endswith("/"):
|
||||
src_b = os.fsencode(src)
|
||||
keys = self.repo.refs.subkeys(src_b)
|
||||
values = [self.repo.refs[b"".join([src_b, key])] for key in keys]
|
||||
dest_refs = [b"".join([os.fsencode(dest), key]) for key in keys]
|
||||
else:
|
||||
if src is None:
|
||||
values = [ZERO_SHA]
|
||||
else:
|
||||
values = [self.repo.refs[os.fsencode(src)]]
|
||||
dest_refs = [os.fsencode(dest)]
|
||||
|
||||
def update_refs(refs):
|
||||
for ref, value in zip(dest_refs, values):
|
||||
refs[ref] = value
|
||||
return refs
|
||||
|
||||
try:
|
||||
client, path = get_transport_and_path(url)
|
||||
except Exception as exc:
|
||||
raise SCMError("Could not get remote client") from exc
|
||||
|
||||
def progress(msg):
|
||||
logger.trace("git send_pack: %s", msg)
|
||||
|
||||
client.send_pack(
|
||||
path,
|
||||
update_refs,
|
||||
self.repo.object_store.generate_pack_data,
|
||||
progress=progress,
|
||||
)
|
||||
|
||||
def fetch_refspecs(
|
||||
self,
|
||||
url: str,
|
||||
refspecs: Iterable[str],
|
||||
force: Optional[bool] = False,
|
||||
on_diverged: Optional[Callable[[bytes, bytes], bool]] = None,
|
||||
):
|
||||
from dulwich.client import get_transport_and_path
|
||||
from dulwich.objectspec import parse_reftuples
|
||||
from dulwich.porcelain import DivergedBranches, check_diverged
|
||||
|
||||
fetch_refs = []
|
||||
|
||||
def determine_wants(remote_refs):
|
||||
fetch_refs.extend(
|
||||
parse_reftuples(
|
||||
remote_refs,
|
||||
self.repo.refs,
|
||||
[os.fsencode(refspec) for refspec in refspecs],
|
||||
force=force,
|
||||
)
|
||||
)
|
||||
return [
|
||||
remote_refs[lh]
|
||||
for (lh, _, _) in fetch_refs
|
||||
if remote_refs[lh] not in self.repo.object_store
|
||||
]
|
||||
|
||||
try:
|
||||
client, path = get_transport_and_path(url)
|
||||
except Exception as exc:
|
||||
raise SCMError("Could not get remote client") from exc
|
||||
|
||||
def progress(msg):
|
||||
logger.trace("git fetch: %s", msg)
|
||||
|
||||
fetch_result = client.fetch(
|
||||
path, self.repo, progress=progress, determine_wants=determine_wants
|
||||
)
|
||||
for (lh, rh, _) in fetch_refs:
|
||||
try:
|
||||
if rh in self.repo.refs:
|
||||
check_diverged(
|
||||
self.repo, self.repo.refs[rh], fetch_result.refs[lh]
|
||||
)
|
||||
except DivergedBranches:
|
||||
if not force:
|
||||
overwrite = False
|
||||
if on_diverged:
|
||||
overwrite = on_diverged(rh, fetch_result.refs[lh])
|
||||
if not overwrite:
|
||||
continue
|
||||
self.repo.refs[rh] = fetch_result.refs[lh]
|
||||
|
||||
def _stash_iter(self, ref: str):
|
||||
stash = self._get_stash(ref)
|
||||
yield from stash.stashes()
|
||||
|
||||
def _stash_push(
|
||||
self,
|
||||
ref: str,
|
||||
message: Optional[str] = None,
|
||||
include_untracked: Optional[bool] = False,
|
||||
) -> Tuple[Optional[str], bool]:
|
||||
from dvc.scm.git import Stash
|
||||
|
||||
if include_untracked or ref == Stash.DEFAULT_STASH:
|
||||
# dulwich stash.push does not support include_untracked and does
|
||||
# not touch working tree
|
||||
raise NotImplementedError
|
||||
|
||||
stash = self._get_stash(ref)
|
||||
message_b = message.encode("utf-8") if message else None
|
||||
stash.push(message=message_b)
|
||||
return os.fsdecode(stash[0].new_sha), True
|
||||
|
||||
def _stash_apply(self, rev: str):
|
||||
raise NotImplementedError
|
||||
|
||||
def reflog_delete(self, ref: str, updateref: bool = False):
|
||||
raise NotImplementedError
|
||||
@@ -0,0 +1,101 @@
|
||||
import logging
|
||||
from typing import Callable, Iterable, Optional, Tuple
|
||||
|
||||
from .base import BaseGitBackend
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class GitPythonBackend(BaseGitBackend): # pylint:disable=abstract-method
|
||||
"""git-python Git backend."""
|
||||
|
||||
@property
|
||||
def git(self):
|
||||
return self.scm.repo.git
|
||||
|
||||
def is_ignored(self, path):
|
||||
raise NotImplementedError
|
||||
|
||||
def set_ref(
|
||||
self,
|
||||
name: str,
|
||||
new_ref: str,
|
||||
old_ref: Optional[str] = None,
|
||||
message: Optional[str] = None,
|
||||
symbolic: Optional[bool] = False,
|
||||
):
|
||||
raise NotImplementedError
|
||||
|
||||
def get_ref(self, name, follow: Optional[bool] = True) -> Optional[str]:
|
||||
raise NotImplementedError
|
||||
|
||||
def remove_ref(self, name: str, old_ref: Optional[str] = None):
|
||||
raise NotImplementedError
|
||||
|
||||
def get_refs_containing(self, rev: str, pattern: Optional[str] = None):
|
||||
from git.exc import GitCommandError
|
||||
|
||||
try:
|
||||
if pattern:
|
||||
args = [pattern]
|
||||
else:
|
||||
args = []
|
||||
for line in self.git.for_each_ref(
|
||||
*args, contains=rev, format=r"%(refname)"
|
||||
).splitlines():
|
||||
line = line.strip()
|
||||
if line:
|
||||
yield line
|
||||
except GitCommandError:
|
||||
pass
|
||||
|
||||
def push_refspec(self, url: str, src: Optional[str], dest: str):
|
||||
raise NotImplementedError
|
||||
|
||||
def fetch_refspecs(
|
||||
self,
|
||||
url: str,
|
||||
refspecs: Iterable[str],
|
||||
force: Optional[bool] = False,
|
||||
on_diverged: Optional[Callable[[bytes, bytes], bool]] = None,
|
||||
):
|
||||
raise NotImplementedError
|
||||
|
||||
def _stash_iter(self, ref: str):
|
||||
raise NotImplementedError
|
||||
|
||||
def _stash_push(
|
||||
self,
|
||||
ref: str,
|
||||
message: Optional[str] = None,
|
||||
include_untracked: Optional[bool] = False,
|
||||
) -> Tuple[Optional[str], bool]:
|
||||
from dvc.scm.git import Stash
|
||||
|
||||
args = ["push"]
|
||||
if message:
|
||||
args.extend(["-m", message])
|
||||
if include_untracked:
|
||||
args.append("--include-untracked")
|
||||
self.git.stash(*args)
|
||||
commit = self.scm.resolve_commit("stash@{0}")
|
||||
if ref != Stash.DEFAULT_STASH:
|
||||
# `git stash` CLI doesn't support using custom refspecs,
|
||||
# so we push a commit onto refs/stash, make our refspec
|
||||
# point to the new commit, then pop it from refs/stash
|
||||
# `git stash create` is intended to be used for this kind of
|
||||
# behavior but it doesn't support --include-untracked so we need to
|
||||
# use push
|
||||
self.scm.set_ref(ref, commit.hexsha, message=commit.message)
|
||||
self.git.stash("drop")
|
||||
return commit.hexsha, False
|
||||
|
||||
def _stash_apply(self, rev: str):
|
||||
self.git.stash("apply", rev)
|
||||
|
||||
def reflog_delete(self, ref: str, updateref: bool = False):
|
||||
args = ["delete"]
|
||||
if updateref:
|
||||
args.append("--updateref")
|
||||
args.append(ref)
|
||||
self.git.reflog(*args)
|
||||
@@ -0,0 +1,84 @@
|
||||
"""git stash convenience wrapper."""
|
||||
|
||||
import logging
|
||||
import os
|
||||
from typing import Optional
|
||||
|
||||
from dvc.scm.base import SCMError
|
||||
from dvc.utils.fs import remove
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class Stash:
|
||||
"""Wrapper for representing any arbitrary Git stash ref."""
|
||||
|
||||
DEFAULT_STASH = "refs/stash"
|
||||
|
||||
def __init__(self, scm, ref: Optional[str] = None):
|
||||
self.ref = ref if ref else self.DEFAULT_STASH
|
||||
self.scm = scm
|
||||
|
||||
@property
|
||||
def git(self):
|
||||
return self.scm.repo.git
|
||||
|
||||
def __iter__(self):
|
||||
yield from self.scm._stash_iter(self.ref)
|
||||
|
||||
def __len__(self):
|
||||
return len(self.list())
|
||||
|
||||
def __getitem__(self, index):
|
||||
return self.list()[index]
|
||||
|
||||
def list(self):
|
||||
return list(iter(self))
|
||||
|
||||
def push(
|
||||
self,
|
||||
message: Optional[str] = None,
|
||||
include_untracked: Optional[bool] = False,
|
||||
) -> Optional[str]:
|
||||
if not self.scm.is_dirty(untracked_files=include_untracked):
|
||||
logger.debug("No changes to stash")
|
||||
return None
|
||||
|
||||
logger.debug("Stashing changes in '%s'", self.ref)
|
||||
rev, reset = self.scm._stash_push( # pylint: disable=protected-access
|
||||
self.ref, message=message, include_untracked=include_untracked
|
||||
)
|
||||
if reset:
|
||||
self.git.reset(hard=True)
|
||||
return rev
|
||||
|
||||
def pop(self):
|
||||
logger.debug("Popping from stash '%s'", self.ref)
|
||||
ref = "{0}@{{0}}".format(self.ref)
|
||||
rev = self.scm.resolve_rev(ref)
|
||||
self.apply(rev)
|
||||
self.drop()
|
||||
return rev
|
||||
|
||||
def apply(self, rev):
|
||||
logger.debug("Applying stash commit '%s'", rev)
|
||||
self.scm._stash_apply(rev) # pylint: disable=protected-access
|
||||
|
||||
def drop(self, index: int = 0):
|
||||
ref = "{0}@{{{1}}}".format(self.ref, index)
|
||||
if index < 0 or index >= len(self):
|
||||
raise SCMError(f"Invalid stash ref '{ref}'")
|
||||
logger.debug("Dropping '%s'", ref)
|
||||
self.scm.reflog_delete(ref, updateref=True)
|
||||
|
||||
# if we removed the last reflog entry, delete the ref and reflog
|
||||
if len(self) == 0:
|
||||
self.scm.remove_ref(self.ref)
|
||||
parts = self.ref.split("/")
|
||||
reflog = os.path.join(self.scm.root_dir, ".git", "logs", *parts)
|
||||
remove(reflog)
|
||||
|
||||
def clear(self):
|
||||
logger.debug("Clear stash '%s'", self.ref)
|
||||
for _ in range(len(self)):
|
||||
self.drop()
|
||||
@@ -50,7 +50,7 @@ install_requires = [
|
||||
"colorama>=0.3.9",
|
||||
"configobj>=5.0.6",
|
||||
"gitpython>3",
|
||||
"dulwich>=0.20.11",
|
||||
"dulwich>=0.20.14",
|
||||
"setuptools>=34.0.0",
|
||||
"nanotime>=0.5.2",
|
||||
"pyasn1>=0.4.1",
|
||||
|
||||
@@ -163,3 +163,120 @@ def test_gitignore_should_append_newline_to_gitignore(tmp_dir, scm):
|
||||
assert gitignore.read_text().endswith("\n")
|
||||
|
||||
assert contents.splitlines() == ["/foo", "/bar"]
|
||||
|
||||
|
||||
def test_git_detach_head(tmp_dir, scm):
|
||||
tmp_dir.scm_gen({"file": "0"}, commit="init")
|
||||
init_rev = scm.get_rev()
|
||||
|
||||
with scm.detach_head() as rev:
|
||||
assert init_rev == rev
|
||||
assert init_rev == (tmp_dir / ".git" / "HEAD").read_text().strip()
|
||||
assert (
|
||||
"ref: refs/heads/master"
|
||||
== (tmp_dir / ".git" / "HEAD").read_text().strip()
|
||||
)
|
||||
|
||||
|
||||
def test_git_stash_workspace(tmp_dir, scm):
|
||||
tmp_dir.scm_gen({"file": "0"}, commit="init")
|
||||
tmp_dir.gen("file", "1")
|
||||
|
||||
with scm.stash_workspace():
|
||||
assert not scm.repo.is_dirty()
|
||||
assert "0" == (tmp_dir / "file").read_text()
|
||||
assert scm.repo.is_dirty()
|
||||
assert "1" == (tmp_dir / "file").read_text()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"ref, include_untracked",
|
||||
[
|
||||
(None, True),
|
||||
(None, False),
|
||||
("refs/foo/stash", True),
|
||||
("refs/foo/stash", False),
|
||||
],
|
||||
)
|
||||
def test_git_stash_push(tmp_dir, scm, ref, include_untracked):
|
||||
from dvc.scm.git import Stash
|
||||
|
||||
tmp_dir.scm_gen({"file": "0"}, commit="init")
|
||||
tmp_dir.gen({"file": "1", "untracked": "0"})
|
||||
|
||||
stash = Stash(scm, ref=ref)
|
||||
rev = stash.push(include_untracked=include_untracked)
|
||||
assert rev == scm.get_ref(stash.ref)
|
||||
assert "0" == (tmp_dir / "file").read_text()
|
||||
assert include_untracked != (tmp_dir / "untracked").exists()
|
||||
assert len(stash) == 1
|
||||
|
||||
stash.apply(rev)
|
||||
assert "1" == (tmp_dir / "file").read_text()
|
||||
assert "0" == (tmp_dir / "untracked").read_text()
|
||||
|
||||
parts = list(stash.ref.split("/"))
|
||||
assert os.path.exists(os.path.join(os.fspath(tmp_dir), ".git", *parts))
|
||||
assert os.path.exists(
|
||||
os.path.join(os.fspath(tmp_dir), ".git", "logs", *parts)
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("ref", [None, "refs/foo/stash"])
|
||||
def test_git_stash_drop(tmp_dir, scm, ref):
|
||||
from dvc.scm.git import Stash
|
||||
|
||||
tmp_dir.scm_gen({"file": "0"}, commit="init")
|
||||
tmp_dir.gen("file", "1")
|
||||
|
||||
stash = Stash(scm, ref=ref)
|
||||
stash.push()
|
||||
|
||||
tmp_dir.gen("file", "2")
|
||||
expected = stash.push()
|
||||
|
||||
stash.drop(1)
|
||||
assert expected == scm.get_ref(stash.ref)
|
||||
assert len(stash) == 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize("ref", [None, "refs/foo/stash"])
|
||||
def test_git_stash_pop(tmp_dir, scm, ref):
|
||||
from dvc.scm.git import Stash
|
||||
|
||||
tmp_dir.scm_gen({"file": "0"}, commit="init")
|
||||
tmp_dir.gen("file", "1")
|
||||
|
||||
stash = Stash(scm, ref=ref)
|
||||
first = stash.push()
|
||||
|
||||
tmp_dir.gen("file", "2")
|
||||
second = stash.push()
|
||||
|
||||
assert second == stash.pop()
|
||||
assert len(stash) == 1
|
||||
assert first == scm.get_ref(stash.ref)
|
||||
assert "2" == (tmp_dir / "file").read_text()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("ref", [None, "refs/foo/stash"])
|
||||
def test_git_stash_clear(tmp_dir, scm, ref):
|
||||
from dvc.scm.git import Stash
|
||||
|
||||
tmp_dir.scm_gen({"file": "0"}, commit="init")
|
||||
tmp_dir.gen("file", "1")
|
||||
|
||||
stash = Stash(scm, ref=ref)
|
||||
stash.push()
|
||||
|
||||
tmp_dir.gen("file", "2")
|
||||
stash.push()
|
||||
|
||||
stash.clear()
|
||||
assert len(stash) == 0
|
||||
|
||||
parts = list(stash.ref.split("/"))
|
||||
assert not os.path.exists(os.path.join(os.fspath(tmp_dir), ".git", *parts))
|
||||
assert not os.path.exists(
|
||||
os.path.join(os.fspath(tmp_dir), ".git", "logs", *parts)
|
||||
)
|
||||
|
||||
@@ -2,6 +2,7 @@ import os
|
||||
|
||||
import pytest
|
||||
|
||||
from dvc.scm.base import SCMError
|
||||
from tests.basic_env import TestDvcGit
|
||||
|
||||
|
||||
@@ -123,3 +124,125 @@ def test_branch_revs(tmp_dir, scm):
|
||||
for rev in scm.branch_revs("master", init_rev):
|
||||
assert rev == expected.pop()
|
||||
assert len(expected) == 0
|
||||
|
||||
|
||||
def test_set_ref(tmp_dir, scm):
|
||||
tmp_dir.scm_gen({"file": "0"}, commit="init")
|
||||
init_rev = scm.get_rev()
|
||||
tmp_dir.scm_gen({"file": "1"}, commit="commit")
|
||||
commit_rev = scm.get_rev()
|
||||
|
||||
scm.set_ref("refs/foo/bar", init_rev)
|
||||
assert (
|
||||
init_rev
|
||||
== (tmp_dir / ".git" / "refs" / "foo" / "bar").read_text().strip()
|
||||
)
|
||||
|
||||
with pytest.raises(SCMError):
|
||||
scm.set_ref("refs/foo/bar", commit_rev, old_ref=commit_rev)
|
||||
scm.set_ref("refs/foo/bar", commit_rev, old_ref=init_rev)
|
||||
assert (
|
||||
commit_rev
|
||||
== (tmp_dir / ".git" / "refs" / "foo" / "bar").read_text().strip()
|
||||
)
|
||||
|
||||
scm.set_ref("refs/foo/baz", "refs/heads/master", symbolic=True)
|
||||
assert (
|
||||
"ref: refs/heads/master"
|
||||
== (tmp_dir / ".git" / "refs" / "foo" / "baz").read_text().strip()
|
||||
)
|
||||
|
||||
|
||||
def test_get_ref(tmp_dir, scm):
|
||||
tmp_dir.scm_gen({"file": "0"}, commit="init")
|
||||
init_rev = scm.get_rev()
|
||||
tmp_dir.gen(
|
||||
{
|
||||
os.path.join(".git", "refs", "foo", "bar"): init_rev,
|
||||
os.path.join(
|
||||
".git", "refs", "foo", "baz"
|
||||
): "ref: refs/heads/master",
|
||||
}
|
||||
)
|
||||
|
||||
assert init_rev == scm.get_ref("refs/foo/bar")
|
||||
assert init_rev == scm.get_ref("refs/foo/baz")
|
||||
assert "refs/heads/master" == scm.get_ref("refs/foo/baz", follow=False)
|
||||
assert scm.get_ref("refs/foo/qux") is None
|
||||
|
||||
|
||||
def test_remove_ref(tmp_dir, scm):
|
||||
tmp_dir.scm_gen({"file": "0"}, commit="init")
|
||||
init_rev = scm.get_rev()
|
||||
tmp_dir.gen(os.path.join(".git", "refs", "foo", "bar"), init_rev)
|
||||
tmp_dir.scm_gen({"file": "1"}, commit="commit")
|
||||
commit_rev = scm.get_rev()
|
||||
|
||||
with pytest.raises(SCMError):
|
||||
scm.remove_ref("refs/foo/bar", old_ref=commit_rev)
|
||||
scm.remove_ref("refs/foo/bar", old_ref=init_rev)
|
||||
assert not (tmp_dir / ".git" / "refs" / "foo" / "bar").exists()
|
||||
|
||||
|
||||
def test_refs_containing(tmp_dir, scm):
|
||||
tmp_dir.scm_gen({"file": "0"}, commit="init")
|
||||
init_rev = scm.get_rev()
|
||||
tmp_dir.gen(
|
||||
{
|
||||
os.path.join(".git", "refs", "foo", "bar"): init_rev,
|
||||
os.path.join(".git", "refs", "foo", "baz"): init_rev,
|
||||
}
|
||||
)
|
||||
|
||||
expected = {"refs/foo/bar", "refs/foo/baz", "refs/heads/master"}
|
||||
assert expected == set(scm.get_refs_containing(init_rev))
|
||||
|
||||
|
||||
def test_push_refspec(tmp_dir, scm, make_tmp_dir):
|
||||
tmp_dir.scm_gen({"file": "0"}, commit="init")
|
||||
init_rev = scm.get_rev()
|
||||
tmp_dir.gen(
|
||||
{
|
||||
os.path.join(".git", "refs", "foo", "bar"): init_rev,
|
||||
os.path.join(".git", "refs", "foo", "baz"): init_rev,
|
||||
}
|
||||
)
|
||||
remote_dir = make_tmp_dir("git-remote", scm=True)
|
||||
url = "file://{}".format(remote_dir.resolve().as_posix())
|
||||
|
||||
scm.push_refspec(url, "refs/foo/bar", "refs/foo/bar")
|
||||
assert init_rev == remote_dir.scm.get_ref("refs/foo/bar")
|
||||
|
||||
remote_dir.scm.checkout("refs/foo/bar")
|
||||
assert init_rev == remote_dir.scm.get_rev()
|
||||
assert "0" == (remote_dir / "file").read_text()
|
||||
|
||||
scm.push_refspec(url, "refs/foo/", "refs/foo/")
|
||||
assert init_rev == remote_dir.scm.get_ref("refs/foo/baz")
|
||||
|
||||
scm.push_refspec(url, None, "refs/foo/baz")
|
||||
assert remote_dir.scm.get_ref("refs/foo/baz") is None
|
||||
|
||||
|
||||
def test_fetch_refspecs(tmp_dir, scm, make_tmp_dir):
|
||||
remote_dir = make_tmp_dir("git-remote", scm=True)
|
||||
url = "file://{}".format(remote_dir.resolve().as_posix())
|
||||
|
||||
remote_dir.scm_gen({"file": "0"}, commit="init")
|
||||
init_rev = remote_dir.scm.get_rev()
|
||||
remote_dir.gen(
|
||||
{
|
||||
os.path.join(".git", "refs", "foo", "bar"): init_rev,
|
||||
os.path.join(".git", "refs", "foo", "baz"): init_rev,
|
||||
}
|
||||
)
|
||||
|
||||
scm.fetch_refspecs(
|
||||
url, ["refs/foo/bar:refs/foo/bar", "refs/foo/baz:refs/foo/baz"]
|
||||
)
|
||||
assert init_rev == scm.get_ref("refs/foo/bar")
|
||||
assert init_rev == scm.get_ref("refs/foo/baz")
|
||||
|
||||
remote_dir.scm.checkout("refs/foo/bar")
|
||||
assert init_rev == remote_dir.scm.get_rev()
|
||||
assert "0" == (remote_dir / "file").read_text()
|
||||
|
||||
Reference in New Issue
Block a user