Repository: iterative/dvc
Pull request: tests: cover run cache with functional tests
Also, I tried to fix #4016, and also fix discrepancies in push/fetch counts.

* [x] ❗ I have followed the [Contributing to DVC](https://dvc.org/doc/user-guide/contributing/core) checklist.

* [x] 📖 If this PR requires [documentation](https://dvc.org/doc) updates, I have created a separate PR (or issue, at least) in [dvc.org](https://github.com/iterative/dvc.org) and linked it here. If the CLI API is changed, I have updated [tab completion scripts](https://github.com/iterative/dvc/tree/master/scripts/completion).

* [x] ❌ I will check DeepSource, CodeClimate, and other sanity checks below. (We consider them recommendatory and don't expect everything to be addressed. Please fix things that actually improve code or fix bugs.)

Thank you for the contribution - we'll try to review it as soon as possible. 🙏


Problem statement:
run cache: push/pull is broken
## Bug Report

### Please provide information about your setup

#### Setup:
```console
cd $(mktemp -d)
mkdir {remote,repo}
cd repo
dvc init --no-scm
echo "foo" > foo
dvc run -n copy-foo-bar -d foo -o bar "cp foo bar"
dvc remote add -d remote ../remote
```

Output:
```console
$ dvc push --run-cache -v
2020-06-11 12:50:21,102 DEBUG: fetched: [(3,)]
2020-06-11 12:50:21,103 DEBUG: fetched: [(3,)]
2020-06-11 12:50:21,103 ERROR: unexpected error - 'LocalRemote' object has no attribute 'upload'
------------------------------------------------------------
Traceback (most recent call last):
  File "/home/saugat/repos/iterative/dvc/dvc/main.py", line 53, in main
    ret = cmd.run()
  File "/home/saugat/repos/iterative/dvc/dvc/command/data_sync.py", line 58, in run
    run_cache=self.args.run_cache,
  File "/home/saugat/repos/iterative/dvc/dvc/repo/__init__.py", line 35, in wrapper
    ret = f(repo, *args, **kwargs)
  File "/home/saugat/repos/iterative/dvc/dvc/repo/push.py", line 17, in push
    used_run_cache = self.stage_cache.push(remote) if run_cache else []
  File "/home/saugat/repos/iterative/dvc/dvc/stage/cache.py", line 180, in push
    return self._transfer(remote.upload, self.repo.cache.local, remote)
AttributeError: 'LocalRemote' object has no attribute 'upload'
------------------------------------------------------------
```


**Output of `dvc version`:**

```console
$ dvc version
DVC version: 1.0.0a10+4042d5
Python version: 3.6.6
Platform: Linux-5.5.2-arch1-1-x86_64-with-arch
Binary: False
Package: None
Supported remotes: azure, gdrive, gs, hdfs, http, https, s3, ssh, oss
Cache: reflink - not supported, hardlink - supported, symlink - supported
Filesystem type (cache directory): ('tmpfs', 'tmpfs')
Repo: dvc (no_scm)
Filesystem type (workspace): ('tmpfs', 'tmpfs')
```

cc @pmrowla

Hints from issue discussion:
Here's a test that I have been writing:
```python
def test_push_pull(tmp_dir, dvc, erepo_dir, run_copy, setup_remote):
    tmp_dir.gen("foo", "foo")
    run_copy("foo", "bar", name="copy-foo-bar")
    url = setup_remote(dvc)
    dvc.push(run_cache=True)  # fails here already
    setup_remote(erepo_dir.dvc, url)
    with erepo_dir.chdir():
        erepo_dir.dvc.pull(run_cache=True)
        assert os.listdir(erepo_dir.dvc.stage_cache.cache_dir)
```
In `stage/cache.py`: `_transfer`, `push()`, `pull()` should be updated to reflect that

* `remote.exists`
* `remote.walk_files`
* `remote.upload`
* `remote.download`

have all been moved to `remote.tree.func()`
My bad for not covering that one with proper tests. Was in a rush back then. Should've at least created an issue and should've added it to the next sprint. Lesson learned. :slightly_frowning_face:

Commit message: tests: cover run cache with functional tests

Patch to review:
diff --git a/dvc/repo/fetch.py b/dvc/repo/fetch.py
index dcab08ec47..fe3cd9c6f6 100644
--- a/dvc/repo/fetch.py
+++ b/dvc/repo/fetch.py
@@ -72,7 +72,7 @@ def _fetch(
     if failed:
         raise DownloadError(failed)
 
-    return downloaded
+    return downloaded + len(used_run_cache)
 
 
 def _fetch_external(self, repo_url, repo_rev, files, jobs):
diff --git a/dvc/repo/push.py b/dvc/repo/push.py
index f0c88e298e..a950dd2844 100644
--- a/dvc/repo/push.py
+++ b/dvc/repo/push.py
@@ -32,4 +32,4 @@ def push(
         used_run_cache=used_run_cache,
     )
 
-    return self.cloud.push(used, jobs, remote=remote)
+    return len(used_run_cache) + self.cloud.push(used, jobs, remote=remote)
diff --git a/dvc/repo/reproduce.py b/dvc/repo/reproduce.py
index 4226357974..1c33e23b40 100644
--- a/dvc/repo/reproduce.py
+++ b/dvc/repo/reproduce.py
@@ -63,6 +63,7 @@ def reproduce(
 ):
     from dvc.utils import parse_target
 
+    assert target is None or isinstance(target, str)
     if not target and not all_pipelines:
         raise InvalidArgumentError(
             "Neither `target` nor `--all-pipelines` are specified."
diff --git a/dvc/stage/cache.py b/dvc/stage/cache.py
index f20be749fa..177f1e5300 100644
--- a/dvc/stage/cache.py
+++ b/dvc/stage/cache.py
@@ -177,11 +177,15 @@ def _transfer(func, from_remote, to_remote):
 
     def push(self, remote):
         remote = self.repo.cloud.get_remote(remote)
-        return self._transfer(remote.upload, self.repo.cache.local, remote)
+        return self._transfer(
+            remote.tree.upload, self.repo.cache.local.tree, remote.tree
+        )
 
     def pull(self, remote):
         remote = self.repo.cloud.get_remote(remote)
-        return self._transfer(remote.download, remote, self.repo.cache.local)
+        return self._transfer(
+            remote.tree.download, remote.tree, self.repo.cache.local.tree
+        )
 
     def get_used_cache(self, used_run_cache, *args, **kwargs):
         from dvc.cache import NamedCache
diff --git a/tests/func/test_run_cache.py b/tests/func/test_run_cache.py
new file mode 100644
index 0000000000..7b510b5a3a
--- /dev/null
+++ b/tests/func/test_run_cache.py
@@ -0,0 +1,153 @@
+import os
+
+from dvc.dvcfile import PIPELINE_LOCK
+from dvc.utils import relpath
+
+
+def _recurse_count_files(path):
+    return len([os.path.join(r, f) for r, _, fs in os.walk(path) for f in fs])
+
+
+def test_push_pull(tmp_dir, dvc, erepo_dir, run_copy, setup_remote):
+    tmp_dir.gen("foo", "foo")
+    run_copy("foo", "bar", name="copy-foo-bar")
+    url = setup_remote(dvc)
+    assert dvc.push(run_cache=True) == 2
+    setup_remote(erepo_dir.dvc, url)
+    with erepo_dir.chdir():
+        assert not os.path.exists(erepo_dir.dvc.stage_cache.cache_dir)
+        assert erepo_dir.dvc.pull(run_cache=True)["fetched"] == 2
+        assert os.listdir(erepo_dir.dvc.stage_cache.cache_dir)
+
+
+def test_restore(tmp_dir, dvc, run_copy, mocker):
+    tmp_dir.gen("foo", "foo")
+    run_copy("foo", "bar", name="copy-foo-bar")
+
+    mock_restore = mocker.spy(dvc.stage_cache, "restore")
+    mock_run = mocker.patch("dvc.stage.run.cmd_run")
+
+    # removing any information that `dvc` could use to re-generate from
+    (tmp_dir / "bar").unlink()
+    (tmp_dir / PIPELINE_LOCK).unlink()
+
+    (stage,) = dvc.reproduce("copy-foo-bar")
+
+    mock_restore.assert_called_once_with(stage)
+    mock_run.assert_not_called()
+    assert (tmp_dir / "bar").exists() and not (tmp_dir / "foo").unlink()
+    assert (tmp_dir / PIPELINE_LOCK).exists()
+
+
+def test_save(tmp_dir, dvc, run_copy):
+    run_cache_dir = dvc.stage_cache.cache_dir
+    assert not os.path.exists(run_cache_dir)
+
+    tmp_dir.gen("foo", "foo")
+    stage = run_copy("foo", "bar", name="copy-foo-bar")
+    assert _recurse_count_files(run_cache_dir) == 1
+    with dvc.state:
+        assert dvc.stage_cache.is_cached(stage)
+
+
+def test_do_not_save_on_no_exec_and_dry(tmp_dir, dvc, run_copy):
+    run_cache_dir = dvc.stage_cache.cache_dir
+    assert not os.path.exists(run_cache_dir)
+
+    tmp_dir.gen("foo", "foo")
+    stage = run_copy("foo", "bar", name="copy-foo-bar", no_exec=True)
+
+    assert _recurse_count_files(run_cache_dir) == 0
+    with dvc.state:
+        assert not dvc.stage_cache.is_cached(stage)
+
+    (stage,) = dvc.reproduce("copy-foo-bar", dry=True)
+
+    assert _recurse_count_files(run_cache_dir) == 0
+    with dvc.state:
+        assert not dvc.stage_cache.is_cached(stage)
+
+
+def test_uncached_outs_are_cached(tmp_dir, dvc, run_copy):
+    tmp_dir.gen("foo", "foo")
+    stage = dvc.run(
+        deps=["foo"],
+        cmd="cp foo bar",
+        outs_no_cache=["bar"],
+        name="copy-foo-bar",
+    )
+    with dvc.state:
+        stage.outs[0].checksum = stage.outs[0].get_checksum()
+    assert os.path.exists(relpath(stage.outs[0].cache_path))
+
+
+def test_memory_for_multiple_runs_of_same_stage(
+    tmp_dir, dvc, run_copy, mocker
+):
+    tmp_dir.gen("foo", "foo")
+    assert not os.path.exists(dvc.stage_cache.cache_dir)
+    run_copy("foo", "bar", name="copy-foo-bar")
+    assert _recurse_count_files(dvc.stage_cache.cache_dir) == 1
+    tmp_dir.gen("foo", "foobar")
+    run_copy("foo", "bar", name="copy-foo-bar")
+    assert _recurse_count_files(dvc.stage_cache.cache_dir) == 2
+
+    from dvc.stage import run as _run
+
+    mock_restore = mocker.spy(dvc.stage_cache, "restore")
+    mock_run = mocker.spy(_run, "cmd_run")
+
+    (tmp_dir / "bar").unlink()
+    (tmp_dir / PIPELINE_LOCK).unlink()
+    (stage,) = dvc.reproduce("copy-foo-bar")
+
+    assert (tmp_dir / PIPELINE_LOCK).exists()
+    assert (tmp_dir / "bar").read_text() == "foobar"
+    mock_run.assert_not_called()
+    mock_restore.assert_called_once_with(stage)
+    mock_restore.reset_mock()
+
+    (tmp_dir / PIPELINE_LOCK).unlink()
+    tmp_dir.gen("foo", "foo")
+    dvc.reproduce("copy-foo-bar")
+
+    assert (tmp_dir / "bar").read_text() == "foo"
+    mock_run.assert_not_called()
+    mock_restore.assert_called_once_with(stage)
+    assert (tmp_dir / "bar").exists() and not (tmp_dir / "foo").unlink()
+    assert (tmp_dir / PIPELINE_LOCK).exists()
+
+
+def test_memory_runs_of_multiple_stages(tmp_dir, dvc, run_copy, mocker):
+    tmp_dir.gen("foo", "foo")
+    assert not os.path.exists(dvc.stage_cache.cache_dir)
+
+    run_copy("foo", "foo.bak", name="backup-foo")
+    assert _recurse_count_files(dvc.stage_cache.cache_dir) == 1
+
+    tmp_dir.gen("bar", "bar")
+    run_copy("bar", "bar.bak", name="backup-bar")
+    assert _recurse_count_files(dvc.stage_cache.cache_dir) == 2
+
+    from dvc.stage import run as _run
+
+    mock_restore = mocker.spy(dvc.stage_cache, "restore")
+    mock_run = mocker.spy(_run, "cmd_run")
+
+    (tmp_dir / "foo.bak").unlink()
+    (tmp_dir / "bar.bak").unlink()
+    (tmp_dir / PIPELINE_LOCK).unlink()
+    (stage,) = dvc.reproduce("backup-foo")
+
+    assert (tmp_dir / "foo.bak").read_text() == "foo"
+    assert (tmp_dir / PIPELINE_LOCK).exists()
+    mock_run.assert_not_called()
+    mock_restore.assert_called_once_with(stage)
+    mock_restore.reset_mock()
+
+    (stage,) = dvc.reproduce("backup-bar")
+
+    assert (tmp_dir / "bar.bak").read_text() == "bar"
+    assert (tmp_dir / PIPELINE_LOCK).exists()
+    mock_run.assert_not_called()
+    mock_restore.assert_called_once_with(stage)


Code review comments:
diff_hunk: @@ -177,11 +177,15 @@ def _transfer(func, from_remote, to_remote):
 
     def push(self, remote):
         remote = self.repo.cloud.get_remote(remote)
-        return self._transfer(remote.upload, self.repo.cache.local, remote)
+        return self._transfer(
+            remote.tree.upload, self.repo.cache.local.tree, remote.tree
+        )
path: dvc/stage/cache.py
text: @author:
Just changed it to `.tree` instead of changing everywhere in `_transfer()`.

@user1:
@author `tree.upload/download` are not meant to swallow errors, unlike the remote, so it won't work :slightly_frowning_face: 

@author:
@user1, so, what do you propose? I'm still wrapping my heads around tree. Is there any other interface, or are you proposing to accumulate the errors and continue and show error later?

@user1:
@author We just need to wrap those with `_log_exceptions` helper from `local.py`. That will do the trick for now.