diff --git a/dvc/scm/git.py b/dvc/scm/git/__init__.py similarity index 82% rename from dvc/scm/git.py rename to dvc/scm/git/__init__.py index 7de2736cb..285b1d3c9 100644 --- a/dvc/scm/git.py +++ b/dvc/scm/git/__init__.py @@ -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() diff --git a/dvc/scm/git/backend/__init__.py b/dvc/scm/git/backend/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/dvc/scm/git/backend/base.py b/dvc/scm/git/backend/base.py new file mode 100644 index 000000000..ea60a1ea1 --- /dev/null +++ b/dvc/scm/git/backend/base.py @@ -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.""" diff --git a/dvc/scm/git/backend/dulwich.py b/dvc/scm/git/backend/dulwich.py new file mode 100644 index 000000000..a65fcc7b3 --- /dev/null +++ b/dvc/scm/git/backend/dulwich.py @@ -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 diff --git a/dvc/scm/git/backend/gitpython.py b/dvc/scm/git/backend/gitpython.py new file mode 100644 index 000000000..fdc6c0023 --- /dev/null +++ b/dvc/scm/git/backend/gitpython.py @@ -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) diff --git a/dvc/scm/git/stash.py b/dvc/scm/git/stash.py new file mode 100644 index 000000000..6646f12c0 --- /dev/null +++ b/dvc/scm/git/stash.py @@ -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() diff --git a/setup.py b/setup.py index 6acbe4fac..899527700 100644 --- a/setup.py +++ b/setup.py @@ -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", diff --git a/tests/func/test_scm.py b/tests/func/test_scm.py index bdfb5a971..644fda3b2 100644 --- a/tests/func/test_scm.py +++ b/tests/func/test_scm.py @@ -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) + ) diff --git a/tests/unit/scm/test_git.py b/tests/unit/scm/test_git.py index 0c30db04f..43bf94473 100644 --- a/tests/unit/scm/test_git.py +++ b/tests/unit/scm/test_git.py @@ -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()