Repository: iterative/dvc

Issue:
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 


Discussion:

Patch:
diff --git a/dvc/repo/fetch.py b/dvc/repo/fetch.py
index dcab08ec4..fe3cd9c6f 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/plots/__init__.py b/dvc/repo/plots/__init__.py
index afff0c68f..11e09be6f 100644
--- a/dvc/repo/plots/__init__.py
+++ b/dvc/repo/plots/__init__.py
@@ -5,6 +5,7 @@ from funcy import first, project
 from dvc.exceptions import DvcException, NoPlotsError, OutputNotFoundError
 from dvc.repo.tree import RepoTree
 from dvc.schema import PLOT_PROPS
+from dvc.utils import relpath
 
 logger = logging.getLogger(__name__)
 
@@ -40,23 +41,25 @@ class Plots:
 
             tree = RepoTree(self.repo)
             plots = _collect_plots(self.repo, targets, rev)
-            for datafile, props in plots.items():
+            for path_info, props in plots.items():
+                datafile = relpath(path_info, self.repo.root_dir)
                 data[rev] = {datafile: {"props": props}}
 
                 # Load data from git or dvc cache
                 try:
-                    with tree.open(datafile) as fd:
+                    with tree.open(path_info) as fd:
                         data[rev][datafile]["data"] = fd.read()
-                except FileNotFoundError:
+                except FileNotFoundError as e:
                     # This might happen simply because cache is absent
+                    print(e)
                     pass
 
         return data
 
-    def render(self, data, revs=None, props=None, templates=None):
+    @staticmethod
+    def render(data, revs=None, props=None, templates=None):
         """Renders plots"""
         props = props or {}
-        templates = templates or self.repo.plot_templates
 
         # Merge data by plot file and apply overriding props
         plots = _prepare_plots(data, revs, props)
@@ -81,7 +84,7 @@ class Plots:
         if not data:
             raise NoPlotsError()
 
-        return self.render(data, revs, props)
+        return self.render(data, revs, props, self.repo.plot_templates)
 
     def diff(self, *args, **kwargs):
         from .diff import diff
@@ -135,7 +138,7 @@ def _collect_plots(repo, targets=None, rev=None):
     else:
         outs = (out for stage in repo.stages for out in stage.outs if out.plot)
 
-    return {str(out): _plot_props(out) for out in outs}
+    return {out.path_info: _plot_props(out) for out in outs}
 
 
 def _plot_props(out):
diff --git a/dvc/repo/push.py b/dvc/repo/push.py
index f0c88e298..a950dd284 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 422635797..1c33e23b4 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 f20be749f..2ba6c1cb3 100644
--- a/dvc/stage/cache.py
+++ b/dvc/stage/cache.py
@@ -6,6 +6,7 @@ import yaml
 from funcy import first
 from voluptuous import Invalid
 
+from dvc.remote.local import _log_exceptions
 from dvc.schema import COMPILED_LOCK_FILE_STAGE_SCHEMA
 from dvc.utils import dict_sha256, relpath
 from dvc.utils.fs import makedirs
@@ -177,11 +178,19 @@ class StageCache:
 
     def push(self, remote):
         remote = self.repo.cloud.get_remote(remote)
-        return self._transfer(remote.upload, self.repo.cache.local, remote)
+        return self._transfer(
+            _log_exceptions(remote.tree.upload, "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(
+            _log_exceptions(remote.tree.download, "download"),
+            remote.tree,
+            self.repo.cache.local.tree,
+        )
 
     def get_used_cache(self, used_run_cache, *args, **kwargs):
         from dvc.cache import NamedCache


Test patch:
diff --git a/tests/func/test_run_cache.py b/tests/func/test_run_cache.py
new file mode 100644
index 000000000..7b510b5a3
--- /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)