pygit: account for locked dirs on windows during checkout

dvc/scm/git/backend/pygit2.py

@@ -30,6 +30,12 @@
     from dvc.types import StrPath
 
 
+# NOTE: constant from libgit2 git2/checkout.h
+# This can be removed after next pygit2 release:
+# see https://github.com/libgit2/pygit2/pull/1087
+GIT_CHECKOUT_SKIP_LOCKED_DIRECTORIES = 1 << 18
+
+
 class Pygit2Object(GitObject):
     def __init__(self, obj):
         self.obj = obj
@@ -112,6 +118,16 @@ def default_signature(self):
                 "Git username and email must be configured"
             ) from exc
 
+    @staticmethod
+    def _get_checkout_strategy(strategy: Optional[int] = None):
+        from pygit2 import GIT_CHECKOUT_RECREATE_MISSING, GIT_CHECKOUT_SAFE
+
+        if strategy is None:
+            strategy = GIT_CHECKOUT_SAFE | GIT_CHECKOUT_RECREATE_MISSING
+        if os.name == "nt":
+            strategy |= GIT_CHECKOUT_SKIP_LOCKED_DIRECTORIES
+        return strategy
+
     # Workaround to force git_backend_odb_pack to release open file handles
     # in DVC's mixed git-backend environment.
     # See https://github.com/iterative/dvc/issues/5641
@@ -151,21 +167,23 @@ def checkout(
     ):
         from pygit2 import GIT_CHECKOUT_FORCE, GitError
 
-        checkout_strategy = GIT_CHECKOUT_FORCE if force else None
+        strategy = self._get_checkout_strategy(
+            GIT_CHECKOUT_FORCE if force else None
+        )
 
         with self.release_odb_handles():
             if create_new:
                 commit = self.repo.revparse_single("HEAD")
                 new_branch = self.repo.branches.local.create(branch, commit)
-                self.repo.checkout(new_branch, strategy=checkout_strategy)
+                self.repo.checkout(new_branch, strategy=strategy)
             else:
                 if branch == "-":
                     branch = "@{-1}"
                 try:
                     commit, ref = self._resolve_refish(branch)
                 except (KeyError, GitError):
                     raise RevError(f"unknown Git revision '{branch}'")
-                self.repo.checkout_tree(commit, strategy=checkout_strategy)
+                self.repo.checkout_tree(commit, strategy=strategy)
                 detach = kwargs.get("detach", False)
                 if ref and not detach:
                     self.repo.set_head(ref.name)
@@ -384,20 +402,15 @@ def _stash_push(
         return str(oid), False
 
     def _stash_apply(self, rev: str):
-        from pygit2 import (
-            GIT_CHECKOUT_RECREATE_MISSING,
-            GIT_CHECKOUT_SAFE,
-            GitError,
-        )
+        from pygit2 import GitError
 
         from dvc.scm.git import Stash
 
         def _apply(index):
             try:
                 self.repo.index.read(False)
                 self.repo.stash_apply(
-                    index,
-                    strategy=GIT_CHECKOUT_SAFE | GIT_CHECKOUT_RECREATE_MISSING,
+                    index, strategy=self._get_checkout_strategy()
                 )
             except GitError as exc:
                 raise MergeConflictError(
@@ -481,6 +494,7 @@ def checkout_index(
 
         if ours or theirs:
             strategy |= GIT_CHECKOUT_ALLOW_CONFLICTS
+        strategy = self._get_checkout_strategy(strategy)
 
         index = self.repo.index
         if paths:

tests/unit/scm/test_git.py

@@ -580,3 +580,17 @@ def test_pygit_resolve_refish(tmp_dir, scm, git, use_sha):
     assert str(commit.id) == head
     if not use_sha:
         assert ref.name == f"refs/tags/{tag}"
+
+
+@pytest.mark.skipif(os.name != "nt", reason="Windows only")
+def test_pygit_checkout_subdir(tmp_dir, scm, git):
+    if git.test_backend != "pygit2":
+        pytest.skip()
+
+    tmp_dir.scm_gen("foo", "foo", commit="init")
+    rev = scm.get_rev()
+    tmp_dir.scm_gen({"dir": {"bar": "bar"}}, commit="dir")
+
+    with (tmp_dir / "dir").chdir():
+        git.checkout(rev)
+        assert not (tmp_dir / "dir" / "bar").exists()