stash apply: support reinstate_index and skip_conflicts

Author: pmrowla

scmrepo/git/backend/base.py

@@ -281,7 +281,13 @@ def _stash_push(
         """
 
     @abstractmethod
-    def _stash_apply(self, rev: str):
+    def _stash_apply(
+        self,
+        rev: str,
+        reinstate_index: bool = False,
+        skip_conflicts: bool = False,
+        **kwargs,
+    ):
         """Apply the specified stash revision."""
 
     @abstractmethod

scmrepo/git/backend/dulwich/__init__.py

@@ -695,7 +695,13 @@ def _stash_push(
             ) from exc
         return os.fsdecode(rev), True
 
-    def _stash_apply(self, rev: str):
+    def _stash_apply(
+        self,
+        rev: str,
+        reinstate_index: bool = False,
+        skip_conflicts: bool = False,
+        **kwargs,
+    ):
         raise NotImplementedError
 
     def _stash_drop(self, ref: str, index: int):

scmrepo/git/backend/gitpython.py

@@ -528,11 +528,23 @@ def _stash_push(
             self.git.stash("drop")
         return commit.hexsha, False
 
-    def _stash_apply(self, rev: str):
+    def _stash_apply(
+        self,
+        rev: str,
+        reinstate_index: bool = False,
+        skip_conflicts: bool = False,
+        **kwargs,
+    ):
         from git.exc import GitCommandError
 
+        if skip_conflicts:
+            raise NotImplementedError
         try:
-            self.git.stash("apply", rev)
+            args = ["apply"]
+            if reinstate_index:
+                args.append("--index")
+            args.append(rev)
+            self.git.stash(args)
         except GitCommandError as exc:
             out = str(exc)
             if "CONFLICT" in out or "already exists" in out:

scmrepo/git/backend/pygit2.py

@@ -465,16 +465,25 @@ def _stash_push(
             self.repo.stash_drop()
         return str(oid), False
 
-    def _stash_apply(self, rev: str):
-        from pygit2 import GitError
+    def _stash_apply(
+        self,
+        rev: str,
+        reinstate_index: bool = False,
+        skip_conflicts: bool = False,
+        **kwargs,
+    ):
+        from pygit2 import GIT_CHECKOUT_ALLOW_CONFLICTS, GitError
 
         from scmrepo.git import Stash
 
         def _apply(index):
             try:
                 self.repo.index.read(False)
+                strategy = self._get_checkout_strategy()
+                if skip_conflicts:
+                    strategy |= GIT_CHECKOUT_ALLOW_CONFLICTS
                 self.repo.stash_apply(
-                    index, strategy=self._get_checkout_strategy()
+                    index, strategy=strategy, reinstate_index=reinstate_index
                 )
             except GitError as exc:
                 raise MergeConflictError(

scmrepo/git/stash.py

@@ -44,20 +44,40 @@ def push(
             self.scm.reset(hard=True)
         return rev
 
-    def pop(self):
+    def pop(self, **kwargs):
+        """Pop the last stash commit.
+
+        Supports the same keyword arguments as apply().
+        """
         logger.debug("Popping from stash '%s'", self.ref)
         ref = f"{self.ref}@{{0}}"
         rev = self.scm.resolve_rev(ref)
         try:
-            self.apply(rev)
+            self.apply(rev, **kwargs)
         except Exception as exc:
             raise SCMError("Could not apply stash commit") from exc
         self.drop()
         return rev
 
-    def apply(self, rev):
+    def apply(
+        self,
+        rev: str,
+        reinstate_index: bool = False,
+        skip_conflicts: bool = False,
+    ):
+        """Apply a stash commit.
+
+        Arguments:
+            rev: Stash commit to apply.
+            reinstate_index: If True, stashed index changes will be reapplied.
+            skip_conflicts: If True, conflicting changes will be skipped and
+                will not be applied from the stash. By default, apply will
+                fail if any conflicts are found.
+        """
         logger.debug("Applying stash commit '%s'", rev)
-        self.scm._stash_apply(rev)  # pylint: disable=protected-access
+        self.scm._stash_apply(  # pylint: disable=protected-access
+            rev, reinstate_index=reinstate_index, skip_conflicts=skip_conflicts
+        )
 
     def drop(self, index: int = 0):
         if index < 0 or index >= len(self):

tests/test_pygit2.py

@@ -1,5 +1,6 @@
 import pygit2
 import pytest
+from pytest_mock import MockerFixture
 from pytest_test_utils import TmpDir
 
 from scmrepo.git import Git
@@ -30,3 +31,32 @@ def test_pygit_resolve_refish(tmp_dir: TmpDir, scm: Git, use_sha: str):
     assert str(commit.id) == head
     if not use_sha:
         assert ref.name == f"refs/tags/{tag}"
+
+
+@pytest.mark.parametrize("skip_conflicts", [True, False])
+def test_pygit_stash_apply_conflicts(
+    tmp_dir: TmpDir, scm: Git, skip_conflicts: bool, mocker: MockerFixture
+):
+    from pygit2 import GIT_CHECKOUT_ALLOW_CONFLICTS
+
+    tmp_dir.gen("foo", "foo")
+    scm.add_commit("foo", message="foo")
+    tmp_dir.gen("foo", "bar")
+    scm.stash.push()
+    rev = scm.resolve_rev(r"stash@{0}")
+
+    backend = Pygit2Backend(tmp_dir)
+    mock = mocker.patch.object(backend.repo, "stash_apply")
+    backend._stash_apply(  # pylint: disable=protected-access
+        rev, skip_conflicts=skip_conflicts
+    )
+    expected_strategy = (
+        backend._get_checkout_strategy()  # pylint: disable=protected-access
+    )
+    if skip_conflicts:
+        expected_strategy |= GIT_CHECKOUT_ALLOW_CONFLICTS
+    mock.assert_called_once_with(
+        0,
+        strategy=expected_strategy,
+        reinstate_index=False,
+    )

tests/test_stash.py

@@ -125,3 +125,26 @@ def test_git_stash_clear(tmp_dir: TmpDir, scm: Git, ref: Optional[str]):
     # NOTE: some backends will completely remove reflog file on clear, some
     # will only truncate it, either case means an empty stash
     assert not reflog_file.exists() or not reflog_file.cat()
+
+
+@pytest.mark.skip_git_backend("dulwich")
+def test_git_stash_apply_index(
+    tmp_dir: TmpDir,
+    scm: Git,
+    git: Git,
+):
+    tmp_dir.gen("file", "0")
+    scm.add_commit("file", message="init")
+    tmp_dir.gen("file", "1")
+    scm.add("file")
+    scm.stash.push()
+    rev = scm.resolve_rev(r"stash@{0}")
+
+    stash = Stash(git)
+    stash.apply(rev, reinstate_index=True)
+
+    assert (tmp_dir / "file").read_text() == "1"
+    staged, unstaged, untracked = scm.status()
+    assert dict(staged) == {"modify": ["file"]}
+    assert not dict(unstaged)
+    assert not dict(untracked)