rwlock: check that .dvc/lock is locked

Mostly a sanity check, but we did have a potential bug in `destroy`,
whcih this PR fixes along the way.

Kudos @shcheklein for catching this one.

dvc/lock.py

@@ -68,6 +68,10 @@ def unlock(self):
         self._lock.close()
         self._lock = None
 
+    @property
+    def is_locked(self):
+        return bool(self._lock)
+
     def __enter__(self):
         self.lock()
 

dvc/repo/destroy.py

@@ -1,8 +1,15 @@
 from dvc.utils.fs import remove
+from . import locked
 
 
-def destroy(self):
-    for stage in self.stages:
+@locked
+def _destroy_stages(repo):
+    for stage in repo.stages:
         stage.remove(remove_outs=False)
 
-    remove(self.dvc_dir)
+
+# NOTE: not locking `destroy`, as `remove` will need to delete `.dvc` dir,
+# which will cause issues on Windows, as `.dvc/lock` will be busy.
+def destroy(repo):
+    _destroy_stages(repo)
+    remove(repo.dvc_dir)

dvc/stage.py

@@ -143,6 +143,8 @@ def rwlocked(call, read=None, write=None):
 
     stage = call._args[0]
 
+    assert stage.repo.lock.is_locked
+
     def _chain(names):
         return [
             item.path_info

tests/func/test_repro.py

@@ -841,7 +841,7 @@ def check_already_cached(self, stage):
 
         patch_run = patch.object(stage, "_run", wraps=stage._run)
 
-        with self.dvc.state:
+        with self.dvc.lock, self.dvc.state:
             with patch_download as mock_download:
                 with patch_checkout as mock_checkout:
                     with patch_run as mock_run:

tests/func/test_run.py

@@ -655,7 +655,8 @@ def test_run_deterministic_overwrite(deterministic_run):
 
 
 def test_run_deterministic_callback(deterministic_run):
-    deterministic_run.stage.remove()
+    with deterministic_run.stage.repo.lock:
+        deterministic_run.stage.remove()
     deterministic_run.deps = []
     deterministic_run.run()
     with mock.patch("dvc.prompt.confirm", return_value=True):

tests/func/test_stage.py

@@ -108,7 +108,7 @@ def test_ignored_in_checksum(self):
         d = load_stage_file(stage.relpath)
         self.assertNotIn(Stage.PARAM_WDIR, d.keys())
 
-        with self.dvc.state:
+        with self.dvc.lock, self.dvc.state:
             stage = Stage.load(self.dvc, stage.relpath)
             self.assertFalse(stage.changed())
 

tests/unit/test_stage.py

@@ -111,5 +111,6 @@ def test_stage_run_ignore_sigint(dvc_repo, mocker):
 def test_always_changed(dvc_repo):
     stage = Stage(dvc_repo, "path", always_changed=True)
     stage.save()
-    assert stage.changed()
-    assert stage.status()["path"] == ["always changed"]
+    with dvc_repo.lock:
+        assert stage.changed()
+        assert stage.status()["path"] == ["always changed"]