remote: separate cloud remote and cloud cache classes (#4019)

* remote: move get_file_checksum() into tree

* makes RemoteTree and RepoTree consistent with regard to checksum
  calculation

* remote: save() now takes explicit tree parameter

* when tree is remote.tree, save will be a move + link operation
  (same as default existing behavior)
* when saving path from a different tree, save will be a copy operation

* tests: update for moved remote/tree functions

* remote: move get_checksum into tree

* remote: separate cloud remote and cache classes

* tests: update unit tests for remote/cache separation

* remote: cloud cache should extend cloud remote

* remote: move LocalRemote.get

* tests: update func tests

* dependency: update for moved get_checksum

* remote: fix state lookup bug

dvc/cache.py

@@ -28,13 +28,13 @@ def _make_remote_property(name):
     """
 
     def getter(self):
-        from dvc.remote import Remote
+        from dvc.remote import Cache as CloudCache
 
         remote = self.config.get(name)
         if not remote:
             return None
 
-        return Remote(self.repo, name=remote)
+        return CloudCache(self.repo, name=remote)
 
     getter.__name__ = name
     return cached_property(getter)
@@ -50,7 +50,7 @@ class Cache:
     CACHE_DIR = "cache"
 
     def __init__(self, repo):
-        from dvc.remote import Remote
+        from dvc.remote import Cache as CloudCache
 
         self.repo = repo
         self.config = config = repo.config["cache"]
@@ -62,7 +62,7 @@ class Cache:
         else:
             settings = {**config, "url": config["dir"]}
 
-        self.local = Remote(repo, **settings)
+        self.local = CloudCache(repo, **settings)
 
     s3 = _make_remote_property("s3")
     gs = _make_remote_property("gs")

dvc/dependency/repo.py

@@ -64,9 +64,7 @@ class RepoDependency(LocalDependency):
 
                 # We are polluting our repo cache with some dir listing here
                 if tree.isdir(path):
-                    return self.repo.cache.local.get_dir_checksum(
-                        path, tree=tree
-                    )
+                    return self.repo.cache.local.get_checksum(path, tree)
                 return tree.get_file_checksum(path)
 
     def status(self):

dvc/external_repo.py

@@ -126,8 +126,9 @@ class BaseExternalRepo:
                 raise PathMissingError(path, self.url)
             save_info = self.local_cache.save(
                 path,
+                self.repo_tree,
                 None,
-                tree=self.repo_tree,
+                save_link=False,
                 download_callback=download_update,
             )
             save_infos.append(save_info)

dvc/output/base.py

@@ -267,7 +267,7 @@ class BaseOutput:
 
     def commit(self):
         if self.use_cache:
-            self.cache.save(self.path_info, self.info)
+            self.cache.save(self.path_info, self.cache.tree, self.info)
 
     def dumpd(self):
         ret = copy(self.info)

dvc/remote/__init__.py

@@ -1,16 +1,25 @@
 import posixpath
 from urllib.parse import urlparse
 
-from dvc.remote.azure import AzureRemote
+from dvc.remote.azure import AzureCache, AzureRemote
 from dvc.remote.gdrive import GDriveRemote
-from dvc.remote.gs import GSRemote
-from dvc.remote.hdfs import HDFSRemote
+from dvc.remote.gs import GSCache, GSRemote
+from dvc.remote.hdfs import HDFSCache, HDFSRemote
 from dvc.remote.http import HTTPRemote
 from dvc.remote.https import HTTPSRemote
-from dvc.remote.local import LocalRemote
+from dvc.remote.local import LocalCache, LocalRemote
 from dvc.remote.oss import OSSRemote
-from dvc.remote.s3 import S3Remote
-from dvc.remote.ssh import SSHRemote
+from dvc.remote.s3 import S3Cache, S3Remote
+from dvc.remote.ssh import SSHCache, SSHRemote
+
+CACHES = [
+    AzureCache,
+    GSCache,
+    HDFSCache,
+    S3Cache,
+    SSHCache,
+    # LocalCache is the default
+]
 
 REMOTES = [
     AzureRemote,
@@ -26,21 +35,30 @@ REMOTES = [
 ]
 
 
-def _get(remote_conf):
-    for remote in REMOTES:
+def _get(remote_conf, remotes, default):
+    for remote in remotes:
         if remote.supported(remote_conf):
             return remote
-    return LocalRemote
+    return default
 
 
-def Remote(repo, **kwargs):
+def _get_conf(repo, **kwargs):
     name = kwargs.get("name")
     if name:
         remote_conf = repo.config["remote"][name.lower()]
     else:
         remote_conf = kwargs
-    remote_conf = _resolve_remote_refs(repo.config, remote_conf)
-    return _get(remote_conf)(repo, remote_conf)
+    return _resolve_remote_refs(repo.config, remote_conf)
+
+
+def Remote(repo, **kwargs):
+    remote_conf = _get_conf(repo, **kwargs)
+    return _get(remote_conf, REMOTES, LocalRemote)(repo, remote_conf)
+
+
+def Cache(repo, **kwargs):
+    remote_conf = _get_conf(repo, **kwargs)
+    return _get(remote_conf, CACHES, LocalCache)(repo, remote_conf)
 
 
 def _resolve_remote_refs(config, remote_conf):

dvc/remote/azure.py

@@ -7,7 +7,7 @@ from funcy import cached_property, wrap_prop
 
 from dvc.path_info import CloudURLInfo
 from dvc.progress import Tqdm
-from dvc.remote.base import BaseRemote, BaseRemoteTree
+from dvc.remote.base import BaseRemote, BaseRemoteTree, CacheMixin
 from dvc.scheme import Schemes
 
 logger = logging.getLogger(__name__)
@@ -108,6 +108,9 @@ class AzureRemoteTree(BaseRemoteTree):
         logger.debug(f"Removing {path_info}")
         self.blob_service.delete_blob(path_info.bucket, path_info.path)
 
+    def get_file_checksum(self, path_info):
+        return self.get_etag(path_info)
+
     def _upload(
         self, from_file, to_info, name=None, no_progress_bar=False, **_kwargs
     ):
@@ -134,10 +137,11 @@ class AzureRemoteTree(BaseRemoteTree):
 class AzureRemote(BaseRemote):
     scheme = Schemes.AZURE
     REQUIRES = {"azure-storage-blob": "azure.storage.blob"}
+    TREE_CLS = AzureRemoteTree
     PARAM_CHECKSUM = "etag"
     COPY_POLL_SECONDS = 5
     LIST_OBJECT_PAGE_SIZE = 5000
-    TREE_CLS = AzureRemoteTree
 
-    def get_file_checksum(self, path_info):
-        return self.tree.get_etag(path_info)
+
+class AzureCache(AzureRemote, CacheMixin):
+    pass

dvc/remote/gdrive.py

@@ -545,6 +545,9 @@ class GDriveRemoteTree(BaseRemoteTree):
         item_id = self._get_item_id(path_info)
         self.gdrive_delete_file(item_id)
 
+    def get_file_checksum(self, path_info):
+        raise NotImplementedError
+
     def _upload(self, from_file, to_info, name=None, no_progress_bar=False):
         dirname = to_info.parent
         assert dirname
@@ -562,11 +565,8 @@ class GDriveRemoteTree(BaseRemoteTree):
 class GDriveRemote(BaseRemote):
     scheme = Schemes.GDRIVE
     REQUIRES = {"pydrive2": "pydrive2"}
+    TREE_CLS = GDriveRemoteTree
     DEFAULT_VERIFY = True
     # Always prefer traverse for GDrive since API usage quotas are a concern.
     TRAVERSE_WEIGHT_MULTIPLIER = 1
     TRAVERSE_PREFIX_LEN = 2
-    TREE_CLS = GDriveRemoteTree
-
-    def get_file_checksum(self, path_info):
-        raise NotImplementedError

dvc/remote/gs.py

@@ -9,7 +9,7 @@ from funcy import cached_property, wrap_prop
 from dvc.exceptions import DvcException
 from dvc.path_info import CloudURLInfo
 from dvc.progress import Tqdm
-from dvc.remote.base import BaseRemote, BaseRemoteTree
+from dvc.remote.base import BaseRemote, BaseRemoteTree, CacheMixin
 from dvc.scheme import Schemes
 
 logger = logging.getLogger(__name__)
@@ -157,6 +157,20 @@ class GSRemoteTree(BaseRemoteTree):
         to_bucket = self.gs.bucket(to_info.bucket)
         from_bucket.copy_blob(blob, to_bucket, new_name=to_info.path)
 
+    def get_file_checksum(self, path_info):
+        import base64
+        import codecs
+
+        bucket = path_info.bucket
+        path = path_info.path
+        blob = self.gs.bucket(bucket).get_blob(path)
+        if not blob:
+            return None
+
+        b64_md5 = blob.md5_hash
+        md5 = base64.b64decode(b64_md5)
+        return codecs.getencoder("hex")(md5)[0].decode("utf-8")
+
     def _upload(self, from_file, to_info, name=None, no_progress_bar=False):
         bucket = self.gs.bucket(to_info.bucket)
         _upload_to_bucket(
@@ -184,19 +198,9 @@ class GSRemoteTree(BaseRemoteTree):
 class GSRemote(BaseRemote):
     scheme = Schemes.GS
     REQUIRES = {"google-cloud-storage": "google.cloud.storage"}
-    PARAM_CHECKSUM = "md5"
     TREE_CLS = GSRemoteTree
+    PARAM_CHECKSUM = "md5"
 
-    def get_file_checksum(self, path_info):
-        import base64
-        import codecs
 
-        bucket = path_info.bucket
-        path = path_info.path
-        blob = self.tree.gs.bucket(bucket).get_blob(path)
-        if not blob:
-            return None
-
-        b64_md5 = blob.md5_hash
-        md5 = base64.b64decode(b64_md5)
-        return codecs.getencoder("hex")(md5)[0].decode("utf-8")
+class GSCache(GSRemote, CacheMixin):
+    pass

dvc/remote/hdfs.py

@@ -11,7 +11,7 @@ from urllib.parse import urlparse
 from dvc.scheme import Schemes
 from dvc.utils import fix_env, tmp_fname
 
-from .base import BaseRemote, BaseRemoteTree, RemoteCmdError
+from .base import BaseRemote, BaseRemoteTree, CacheMixin, RemoteCmdError
 from .pool import get_connection
 
 logger = logging.getLogger(__name__)
@@ -122,28 +122,6 @@ class HDFSRemoteTree(BaseRemoteTree):
                     self.remove(tmp_info)
                     raise
 
-    def _upload(self, from_file, to_info, **_kwargs):
-        with self.hdfs(to_info) as hdfs:
-            hdfs.mkdir(posixpath.dirname(to_info.path))
-            tmp_file = tmp_fname(to_info.path)
-            with open(from_file, "rb") as fobj:
-                hdfs.upload(tmp_file, fobj)
-            hdfs.rename(tmp_file, to_info.path)
-
-    def _download(self, from_info, to_file, **_kwargs):
-        with self.hdfs(from_info) as hdfs:
-            with open(to_file, "wb+") as fobj:
-                hdfs.download(from_info.path, fobj)
-
-
-class HDFSRemote(BaseRemote):
-    scheme = Schemes.HDFS
-    REGEX = r"^hdfs://((?P<user>.*)@)?.*$"
-    PARAM_CHECKSUM = "checksum"
-    REQUIRES = {"pyarrow": "pyarrow"}
-    TRAVERSE_PREFIX_LEN = 2
-    TREE_CLS = HDFSRemoteTree
-
     def hadoop_fs(self, cmd, user=None):
         cmd = "hadoop fs -" + cmd
         if user:
@@ -182,3 +160,29 @@ class HDFSRemote(BaseRemote):
             f"checksum {path_info.path}", user=path_info.user
         )
         return self._group(regex, stdout, "checksum")
+
+    def _upload(self, from_file, to_info, **_kwargs):
+        with self.hdfs(to_info) as hdfs:
+            hdfs.mkdir(posixpath.dirname(to_info.path))
+            tmp_file = tmp_fname(to_info.path)
+            with open(from_file, "rb") as fobj:
+                hdfs.upload(tmp_file, fobj)
+            hdfs.rename(tmp_file, to_info.path)
+
+    def _download(self, from_info, to_file, **_kwargs):
+        with self.hdfs(from_info) as hdfs:
+            with open(to_file, "wb+") as fobj:
+                hdfs.download(from_info.path, fobj)
+
+
+class HDFSRemote(BaseRemote):
+    scheme = Schemes.HDFS
+    REGEX = r"^hdfs://((?P<user>.*)@)?.*$"
+    PARAM_CHECKSUM = "checksum"
+    REQUIRES = {"pyarrow": "pyarrow"}
+    TREE_CLS = HDFSRemoteTree
+    TRAVERSE_PREFIX_LEN = 2
+
+
+class HDFSCache(HDFSRemote, CacheMixin):
+    pass

dvc/remote/http.py

@@ -121,6 +121,25 @@ class HTTPRemoteTree(BaseRemoteTree):
     def exists(self, path_info):
         return bool(self.request("HEAD", path_info.url))
 
+    def get_file_checksum(self, path_info):
+        url = path_info.url
+        headers = self.request("HEAD", url).headers
+        etag = headers.get("ETag") or headers.get("Content-MD5")
+
+        if not etag:
+            raise DvcException(
+                "could not find an ETag or "
+                "Content-MD5 header for '{url}'".format(url=url)
+            )
+
+        if etag.startswith("W/"):
+            raise DvcException(
+                "Weak ETags are not supported."
+                " (Etag: '{etag}', URL: '{url}')".format(etag=etag, url=url)
+            )
+
+        return etag
+
     def _download(self, from_info, to_file, name=None, no_progress_bar=False):
         response = self.request("GET", from_info.url, stream=True)
         if response.status_code != 200:
@@ -174,26 +193,7 @@ class HTTPRemote(BaseRemote):
     CAN_TRAVERSE = False
     TREE_CLS = HTTPRemoteTree
 
-    def get_file_checksum(self, path_info):
-        url = path_info.url
-        headers = self.tree.request("HEAD", url).headers
-        etag = headers.get("ETag") or headers.get("Content-MD5")
-
-        if not etag:
-            raise DvcException(
-                "could not find an ETag or "
-                "Content-MD5 header for '{url}'".format(url=url)
-            )
-
-        if etag.startswith("W/"):
-            raise DvcException(
-                "Weak ETags are not supported."
-                " (Etag: '{etag}', URL: '{url}')".format(etag=etag, url=url)
-            )
-
-        return etag
-
-    def list_cache_paths(self, prefix=None, progress_callback=None):
+    def list_paths(self, prefix=None, progress_callback=None):
         raise NotImplementedError
 
     def gc(self):

dvc/remote/local.py

@@ -18,6 +18,7 @@ from dvc.remote.base import (
     STATUS_NEW,
     BaseRemote,
     BaseRemoteTree,
+    CacheMixin,
     index_locked,
 )
 from dvc.remote.index import RemoteIndexNoop
@@ -43,7 +44,8 @@ class LocalRemoteTree(BaseRemoteTree):
 
     def __init__(self, remote, config):
         super().__init__(remote, config)
-        self.path_info = config.get("url")
+        url = config.get("url")
+        self.path_info = self.PATH_CLS(url) if url else None
 
     @property
     def repo(self):
@@ -193,6 +195,9 @@ class LocalRemoteTree(BaseRemoteTree):
         os.chmod(tmp_info, self.file_mode)
         os.rename(tmp_info, to_info)
 
+    def get_file_checksum(self, path_info):
+        return file_md5(path_info)[0]
+
     @staticmethod
     def getsize(path_info):
         return os.path.getsize(path_info)
@@ -241,26 +246,119 @@ def _log_exceptions(func, operation):
 
 class LocalRemote(BaseRemote):
     scheme = Schemes.LOCAL
-    PARAM_CHECKSUM = "md5"
-    PARAM_PATH = "path"
-    TRAVERSE_PREFIX_LEN = 2
     INDEX_CLS = RemoteIndexNoop
     TREE_CLS = LocalRemoteTree
 
+    PARAM_CHECKSUM = "md5"
+    PARAM_PATH = "path"
+    DEFAULT_CACHE_TYPES = ["reflink", "copy"]
+    TRAVERSE_PREFIX_LEN = 2
     UNPACKED_DIR_SUFFIX = ".unpacked"
 
-    DEFAULT_CACHE_TYPES = ["reflink", "copy"]
-
     CACHE_MODE = 0o444
 
-    def __init__(self, repo, config):
-        super().__init__(repo, config)
-        self.cache_dir = config.get("url")
-
     @property
     def state(self):
         return self.repo.state
 
+    def get(self, md5):
+        if not md5:
+            return None
+
+        return self.checksum_to_path_info(md5).url
+
+    def _unprotect_file(self, path):
+        if System.is_symlink(path) or System.is_hardlink(path):
+            logger.debug(f"Unprotecting '{path}'")
+            tmp = os.path.join(os.path.dirname(path), "." + uuid())
+
+            # The operations order is important here - if some application
+            # would access the file during the process of copyfile then it
+            # would get only the part of file. So, at first, the file should be
+            # copied with the temporary name, and then original file should be
+            # replaced by new.
+            copyfile(path, tmp, name="Unprotecting '{}'".format(relpath(path)))
+            remove(path)
+            os.rename(tmp, path)
+
+        else:
+            logger.debug(
+                "Skipping copying for '{}', since it is not "
+                "a symlink or a hardlink.".format(path)
+            )
+
+        os.chmod(path, self.tree.file_mode)
+
+    def _unprotect_dir(self, path):
+        assert is_working_tree(self.repo.tree)
+
+        for fname in self.repo.tree.walk_files(path):
+            self._unprotect_file(fname)
+
+    def unprotect(self, path_info):
+        path = path_info.fspath
+        if not os.path.exists(path):
+            raise DvcException(f"can't unprotect non-existing data '{path}'")
+
+        if os.path.isdir(path):
+            self._unprotect_dir(path)
+        else:
+            self._unprotect_file(path)
+
+    def protect(self, path_info):
+        path = os.fspath(path_info)
+        mode = self.CACHE_MODE
+
+        try:
+            os.chmod(path, mode)
+        except OSError as exc:
+            # There is nothing we need to do in case of a read-only file system
+            if exc.errno == errno.EROFS:
+                return
+
+            # In shared cache scenario, we might not own the cache file, so we
+            # need to check if cache file is already protected.
+            if exc.errno not in [errno.EPERM, errno.EACCES]:
+                raise
+
+            actual = stat.S_IMODE(os.stat(path).st_mode)
+            if actual != mode:
+                raise
+
+    def is_protected(self, path_info):
+        try:
+            mode = os.stat(path_info).st_mode
+        except FileNotFoundError:
+            return False
+
+        return stat.S_IMODE(mode) == self.CACHE_MODE
+
+    def list_paths(self, prefix=None, progress_callback=None):
+        assert self.path_info is not None
+        if prefix:
+            path_info = self.path_info / prefix[:2]
+            if not self.tree.exists(path_info):
+                return
+        else:
+            path_info = self.path_info
+        if progress_callback:
+            for path in walk_files(path_info):
+                progress_callback()
+                yield path
+        else:
+            yield from walk_files(path_info)
+
+    def _remove_unpacked_dir(self, checksum):
+        info = self.checksum_to_path_info(checksum)
+        path_info = info.with_name(info.name + self.UNPACKED_DIR_SUFFIX)
+        self.tree.remove(path_info)
+
+
+class LocalCache(LocalRemote, CacheMixin):
+    def __init__(self, repo, config):
+        super().__init__(repo, config)
+        self.cache_dir = config.get("url")
+
     @property
     def cache_dir(self):
         return self.tree.path_info.fspath if self.tree.path_info else None
@@ -285,26 +383,17 @@ class LocalRemote(BaseRemote):
             f"{self.cache_path}{os.sep}{checksum[0:2]}{os.sep}{checksum[2:]}"
         )
 
-    def list_cache_paths(self, prefix=None, progress_callback=None):
-        assert self.path_info is not None
-        if prefix:
-            path_info = self.path_info / prefix[:2]
-            if not self.tree.exists(path_info):
-                return
-        else:
-            path_info = self.path_info
-        if progress_callback:
-            for path in walk_files(path_info):
-                progress_callback()
-                yield path
-        else:
-            yield from walk_files(path_info)
-
-    def get(self, md5):
-        if not md5:
-            return None
-
-        return self.checksum_to_path_info(md5).url
+    def checksums_exist(self, checksums, jobs=None, name=None):
+        return [
+            checksum
+            for checksum in Tqdm(
+                checksums,
+                unit="file",
+                desc="Querying "
+                + ("cache in " + name if name else "local cache"),
+            )
+            if not self.changed_cache_file(checksum)
+        ]
 
     def already_cached(self, path_info):
         assert path_info.scheme in ["", "local"]
@@ -322,21 +411,6 @@ class LocalRemote(BaseRemote):
 
         super()._verify_link(path_info, link_type)
 
-    def get_file_checksum(self, path_info):
-        return file_md5(path_info)[0]
-
-    def cache_exists(self, checksums, jobs=None, name=None):
-        return [
-            checksum
-            for checksum in Tqdm(
-                checksums,
-                unit="file",
-                desc="Querying "
-                + ("cache in " + name if name else "local cache"),
-            )
-            if not self.changed_cache_file(checksum)
-        ]
-
     @index_locked
     def status(
         self,
@@ -376,7 +450,7 @@ class LocalRemote(BaseRemote):
 
         logger.debug("Collecting information from local cache...")
         local_exists = frozenset(
-            self.cache_exists(md5s, jobs=jobs, name=self.cache_dir)
+            self.checksums_exist(md5s, jobs=jobs, name=self.cache_dir)
         )
 
         # This is a performance optimization. We can safely assume that,
@@ -396,7 +470,7 @@ class LocalRemote(BaseRemote):
                 md5s.difference_update(remote_exists)
             if md5s:
                 remote_exists.update(
-                    remote.cache_exists(
+                    remote.checksums_exist(
                         md5s, jobs=jobs, name=str(remote.path_info)
                     )
                 )
@@ -439,7 +513,7 @@ class LocalRemote(BaseRemote):
         indexed_dir_exists = set()
         if indexed_dirs:
             indexed_dir_exists.update(
-                remote._cache_object_exists(indexed_dirs)
+                remote._list_checksums_exists(indexed_dirs)
             )
             missing_dirs = indexed_dirs.difference(indexed_dir_exists)
             if missing_dirs:
@@ -451,7 +525,7 @@ class LocalRemote(BaseRemote):
 
         # Check if non-indexed (new) dir checksums exist on remote
         dir_exists = dir_md5s.intersection(indexed_dir_exists)
-        dir_exists.update(remote._cache_object_exists(dir_md5s - dir_exists))
+        dir_exists.update(remote._list_checksums_exists(dir_md5s - dir_exists))
 
         # If .dir checksum exists on the remote, assume directory contents
         # still exists on the remote
@@ -658,74 +732,3 @@ class LocalRemote(BaseRemote):
                 "nor on remote. Missing cache files: {}".format(missing_desc)
             )
             logger.warning(msg)
-
-    def _unprotect_file(self, path):
-        if System.is_symlink(path) or System.is_hardlink(path):
-            logger.debug(f"Unprotecting '{path}'")
-            tmp = os.path.join(os.path.dirname(path), "." + uuid())
-
-            # The operations order is important here - if some application
-            # would access the file during the process of copyfile then it
-            # would get only the part of file. So, at first, the file should be
-            # copied with the temporary name, and then original file should be
-            # replaced by new.
-            copyfile(path, tmp, name="Unprotecting '{}'".format(relpath(path)))
-            remove(path)
-            os.rename(tmp, path)
-
-        else:
-            logger.debug(
-                "Skipping copying for '{}', since it is not "
-                "a symlink or a hardlink.".format(path)
-            )
-
-        os.chmod(path, self.tree.file_mode)
-
-    def _unprotect_dir(self, path):
-        assert is_working_tree(self.repo.tree)
-
-        for fname in self.repo.tree.walk_files(path):
-            self._unprotect_file(fname)
-
-    def unprotect(self, path_info):
-        path = path_info.fspath
-        if not os.path.exists(path):
-            raise DvcException(f"can't unprotect non-existing data '{path}'")
-
-        if os.path.isdir(path):
-            self._unprotect_dir(path)
-        else:
-            self._unprotect_file(path)
-
-    def protect(self, path_info):
-        path = os.fspath(path_info)
-        mode = self.CACHE_MODE
-
-        try:
-            os.chmod(path, mode)
-        except OSError as exc:
-            # There is nothing we need to do in case of a read-only file system
-            if exc.errno == errno.EROFS:
-                return
-
-            # In shared cache scenario, we might not own the cache file, so we
-            # need to check if cache file is already protected.
-            if exc.errno not in [errno.EPERM, errno.EACCES]:
-                raise
-
-            actual = stat.S_IMODE(os.stat(path).st_mode)
-            if actual != mode:
-                raise
-
-    def _remove_unpacked_dir(self, checksum):
-        info = self.checksum_to_path_info(checksum)
-        path_info = info.with_name(info.name + self.UNPACKED_DIR_SUFFIX)
-        self.tree.remove(path_info)
-
-    def is_protected(self, path_info):
-        try:
-            mode = os.stat(path_info).st_mode
-        except FileNotFoundError:
-            return False
-
-        return stat.S_IMODE(mode) == self.CACHE_MODE

dvc/remote/s3.py

@@ -8,7 +8,7 @@ from dvc.config import ConfigError
 from dvc.exceptions import DvcException, ETagMismatchError
 from dvc.path_info import CloudURLInfo
 from dvc.progress import Tqdm
-from dvc.remote.base import BaseRemote, BaseRemoteTree
+from dvc.remote.base import BaseRemote, BaseRemoteTree, CacheMixin
 from dvc.scheme import Schemes
 
 logger = logging.getLogger(__name__)
@@ -305,6 +305,9 @@ class S3RemoteTree(BaseRemoteTree):
         if etag != cached_etag:
             raise ETagMismatchError(etag, cached_etag)
 
+    def get_file_checksum(self, path_info):
+        return self.get_etag(self.s3, path_info.bucket, path_info.path)
+
     def _upload(self, from_file, to_info, name=None, no_progress_bar=False):
         total = os.path.getsize(from_file)
         with Tqdm(
@@ -339,7 +342,6 @@ class S3Remote(BaseRemote):
     PARAM_CHECKSUM = "etag"
     TREE_CLS = S3RemoteTree
 
-    def get_file_checksum(self, path_info):
-        return self.tree.get_etag(
-            self.tree.s3, path_info.bucket, path_info.path
-        )
+
+class S3Cache(S3Remote, CacheMixin):
+    pass

dvc/remote/ssh/__init__.py

@@ -14,7 +14,7 @@ from funcy import first, memoize, silent, wrap_with
 
 import dvc.prompt as prompt
 from dvc.progress import Tqdm
-from dvc.remote.base import BaseRemote, BaseRemoteTree
+from dvc.remote.base import BaseRemote, BaseRemoteTree, CacheMixin
 from dvc.remote.pool import get_connection
 from dvc.scheme import Schemes
 from dvc.utils import to_chunks
@@ -225,6 +225,13 @@ class SSHRemoteTree(BaseRemoteTree):
         with self.ssh(from_info) as ssh:
             ssh.reflink(from_info.path, to_info.path)
 
+    def get_file_checksum(self, path_info):
+        if path_info.scheme != self.scheme:
+            raise NotImplementedError
+
+        with self.ssh(path_info) as ssh:
+            return ssh.md5(path_info.path)
+
     def getsize(self, path_info):
         with self.ssh(path_info) as ssh:
             return ssh.getsize(path_info.path)
@@ -253,26 +260,18 @@ class SSHRemoteTree(BaseRemoteTree):
 class SSHRemote(BaseRemote):
     scheme = Schemes.SSH
     REQUIRES = {"paramiko": "paramiko"}
-
     JOBS = 4
+    TREE_CLS = SSHRemoteTree
+
     PARAM_CHECKSUM = "md5"
     # At any given time some of the connections will go over network and
     # paramiko stuff, so we would ideally have it double of server processors.
     # We use conservative setting of 4 instead to not exhaust max sessions.
     CHECKSUM_JOBS = 4
-    TRAVERSE_PREFIX_LEN = 2
-    TREE_CLS = SSHRemoteTree
-
     DEFAULT_CACHE_TYPES = ["copy"]
+    TRAVERSE_PREFIX_LEN = 2
 
-    def get_file_checksum(self, path_info):
-        if path_info.scheme != self.scheme:
-            raise NotImplementedError
-
-        with self.tree.ssh(path_info) as ssh:
-            return ssh.md5(path_info.path)
-
-    def list_cache_paths(self, prefix=None, progress_callback=None):
+    def list_paths(self, prefix=None, progress_callback=None):
         if prefix:
             root = posixpath.join(self.path_info.path, prefix[:2])
         else:
@@ -316,7 +315,7 @@ class SSHRemote(BaseRemote):
 
             return results
 
-    def cache_exists(self, checksums, jobs=None, name=None):
+    def checksums_exist(self, checksums, jobs=None, name=None):
         """This is older implementation used in remote/base.py
         We are reusing it in RemoteSSH, because SSH's batch_exists proved to be
         faster than current approach (relying on exists(path_info)) applied in
@@ -345,3 +344,7 @@ class SSHRemote(BaseRemote):
                 in_remote = itertools.chain.from_iterable(results)
                 ret = list(itertools.compress(checksums, in_remote))
                 return ret
+
+
+class SSHCache(SSHRemote, CacheMixin):
+    pass

tests/func/remote/test_index.py

@@ -15,14 +15,14 @@ def remote(tmp_dir, dvc, tmp_path_factory, mocker):
     dvc.config["remote"]["upstream"] = {"url": url}
     dvc.config["core"]["remote"] = "upstream"
 
-    # patch cache_exists since the RemoteLOCAL normally overrides
-    # RemoteBASE.cache_exists.
-    def cache_exists(self, *args, **kwargs):
-        return BaseRemote.cache_exists(self, *args, **kwargs)
+    # patch checksums_exist since the LocalRemote normally overrides
+    # BaseRemote.checksums_exist.
+    def checksums_exist(self, *args, **kwargs):
+        return BaseRemote.checksums_exist(self, *args, **kwargs)
 
-    mocker.patch.object(LocalRemote, "cache_exists", cache_exists)
+    mocker.patch.object(LocalRemote, "checksums_exist", checksums_exist)
 
-    # patch index class since RemoteLOCAL normally overrides index class
+    # patch index class since LocalRemote normally overrides index class
     mocker.patch.object(LocalRemote, "INDEX_CLS", RemoteIndex)
 
     return dvc.cloud.get_remote("upstream")

tests/func/test_add.py

@@ -340,7 +340,7 @@ class TestAddCommit(TestDvc):
 
 def test_should_collect_dir_cache_only_once(mocker, tmp_dir, dvc):
     tmp_dir.gen({"data/data": "foo"})
-    get_dir_checksum_counter = mocker.spy(LocalRemote, "get_dir_checksum")
+    get_dir_checksum_counter = mocker.spy(LocalRemoteTree, "get_dir_checksum")
     ret = main(["add", "data"])
     assert ret == 0
 

tests/func/test_checkout.py

@@ -16,7 +16,7 @@ from dvc.exceptions import (
     NoOutputOrStageError,
 )
 from dvc.main import main
-from dvc.remote import S3Remote
+from dvc.remote import S3Cache, S3Remote
 from dvc.remote.local import LocalRemote
 from dvc.repo import Repo as DvcRepo
 from dvc.stage import Stage
@@ -755,7 +755,7 @@ def test_checkout_recursive(tmp_dir, dvc):
     not S3.should_test(), reason="Only run with S3 credentials"
 )
 def test_checkout_for_external_outputs(tmp_dir, dvc):
-    dvc.cache.s3 = S3Remote(dvc, {"url": S3.get_url()})
+    dvc.cache.s3 = S3Cache(dvc, {"url": S3.get_url()})
 
     remote = S3Remote(dvc, {"url": S3.get_url()})
     file_path = remote.path_info / "foo"

tests/func/test_data_cloud.py

@@ -24,6 +24,7 @@ from dvc.remote import (
     SSHRemote,
 )
 from dvc.remote.base import STATUS_DELETED, STATUS_NEW, STATUS_OK
+from dvc.remote.local import LocalRemoteTree
 from dvc.stage.exceptions import StageNotFound
 from dvc.utils import file_md5
 from dvc.utils.fs import remove
@@ -547,7 +548,9 @@ class TestRecursiveSyncOperations(Local, TestDataCloudBase):
 
     def _prepare_repo(self):
         remote = self.cloud.get_remote()
-        self.main(["remote", "add", "-d", TEST_REMOTE, remote.cache_dir])
+        self.main(
+            ["remote", "add", "-d", TEST_REMOTE, remote.path_info.fspath]
+        )
 
         self.dvc.add(self.DATA)
         self.dvc.add(self.DATA_SUB)
@@ -614,7 +617,7 @@ class TestRecursiveSyncOperations(Local, TestDataCloudBase):
 
 def test_checksum_recalculation(mocker, dvc, tmp_dir):
     tmp_dir.gen({"foo": "foo"})
-    test_get_file_checksum = mocker.spy(LocalRemote, "get_file_checksum")
+    test_get_file_checksum = mocker.spy(LocalRemoteTree, "get_file_checksum")
     url = Local.get_url()
     ret = main(["remote", "add", "-d", TEST_REMOTE, url])
     assert ret == 0
@@ -693,7 +696,7 @@ def test_verify_checksums(
     remove("dir")
     remove(dvc.cache.local.cache_dir)
 
-    checksum_spy = mocker.spy(dvc.cache.local, "get_file_checksum")
+    checksum_spy = mocker.spy(dvc.cache.local.tree, "get_file_checksum")
 
     dvc.pull()
     assert checksum_spy.call_count == 0

tests/func/test_remote.py

@@ -10,7 +10,7 @@ from dvc.config import Config
 from dvc.exceptions import DownloadError, UploadError
 from dvc.main import main
 from dvc.path_info import PathInfo
-from dvc.remote.base import BaseRemote, RemoteCacheRequiredError
+from dvc.remote.base import BaseRemoteTree, RemoteCacheRequiredError
 from dvc.remote.local import LocalRemoteTree
 from dvc.utils.fs import remove
 from tests.basic_env import TestDvc
@@ -151,24 +151,24 @@ def test_dir_checksum_should_be_key_order_agnostic(tmp_dir, dvc):
     path_info = PathInfo("data")
     with dvc.state:
         with patch.object(
-            BaseRemote,
+            BaseRemoteTree,
             "_collect_dir",
             return_value=[
                 {"relpath": "1", "md5": "1"},
                 {"relpath": "2", "md5": "2"},
             ],
         ):
-            checksum1 = dvc.cache.local.get_dir_checksum(path_info)
+            checksum1 = dvc.cache.local.get_checksum(path_info)
 
         with patch.object(
-            BaseRemote,
+            BaseRemoteTree,
             "_collect_dir",
             return_value=[
                 {"md5": "1", "relpath": "1"},
                 {"md5": "2", "relpath": "2"},
             ],
         ):
-            checksum2 = dvc.cache.local.get_dir_checksum(path_info)
+            checksum2 = dvc.cache.local.get_checksum(path_info)
 
     assert checksum1 == checksum2
 

tests/func/test_s3.py

@@ -5,7 +5,7 @@ import moto.s3.models as s3model
 import pytest
 from moto import mock_s3
 
-from dvc.remote.s3 import S3Remote, S3RemoteTree
+from dvc.remote.s3 import S3Cache, S3Remote, S3RemoteTree
 from tests.remotes import S3
 
 # from https://github.com/spulec/moto/blob/v1.3.5/tests/test_s3/test_s3.py#L40
@@ -54,16 +54,16 @@ def test_copy_singlepart_preserve_etag():
     ],
 )
 def test_link_created_on_non_nested_path(base_info, tmp_dir, dvc, scm):
-    remote = S3Remote(dvc, {"url": str(base_info.parent)})
-    s3 = remote.tree.s3
+    cache = S3Cache(dvc, {"url": str(base_info.parent)})
+    s3 = cache.tree.s3
     s3.create_bucket(Bucket=base_info.bucket)
     s3.put_object(
         Bucket=base_info.bucket, Key=(base_info / "from").path, Body="data"
     )
-    remote.link(base_info / "from", base_info / "to")
+    cache.link(base_info / "from", base_info / "to")
 
-    assert remote.tree.exists(base_info / "from")
-    assert remote.tree.exists(base_info / "to")
+    assert cache.tree.exists(base_info / "from")
+    assert cache.tree.exists(base_info / "to")
 
 
 @mock_s3

tests/func/test_tree.py

@@ -218,7 +218,7 @@ def test_repotree_cache_save(tmp_dir, dvc, scm, erepo_dir, setup_remote):
     with erepo_dir.dvc.state:
         cache = dvc.cache.local
         with cache.state:
-            cache.save(PathInfo(erepo_dir / "dir"), None, tree=tree)
+            cache.save(PathInfo(erepo_dir / "dir"), tree, None)
     for checksum in expected:
         assert os.path.exists(cache.checksum_to_path_info(checksum))
 

tests/unit/output/test_local.py

@@ -3,7 +3,7 @@ import os
 from mock import patch
 
 from dvc.output import LocalOutput
-from dvc.remote.local import LocalRemote
+from dvc.remote.local import LocalCache
 from dvc.stage import Stage
 from dvc.utils import relpath
 from tests.basic_env import TestDvc
@@ -79,7 +79,7 @@ class TestGetFilesNumber(TestDvc):
 
     @patch.object(LocalOutput, "checksum", "12345678.dir")
     @patch.object(
-        LocalRemote,
+        LocalCache,
         "get_dir_cache",
         return_value=[{"md5": "asdf"}, {"md5": "qwe"}],
     )

tests/unit/remote/test_azure.py

@@ -42,7 +42,7 @@ def test_get_file_checksum(tmp_dir):
     to_info = remote.tree.PATH_CLS(Azure.get_url())
     remote.tree.upload(PathInfo("foo"), to_info)
     assert remote.tree.exists(to_info)
-    checksum = remote.get_file_checksum(to_info)
+    checksum = remote.tree.get_file_checksum(to_info)
     assert checksum
     assert isinstance(checksum, str)
     assert checksum.strip("'").strip('"') == checksum

tests/unit/remote/test_base.py

@@ -41,18 +41,18 @@ def test_cmd_error(dvc):
             REMOTE_CLS(dvc, config).tree.remove("file")
 
 
-@mock.patch.object(BaseRemote, "_cache_checksums_traverse")
-@mock.patch.object(BaseRemote, "_cache_object_exists")
-def test_cache_exists(object_exists, traverse, dvc):
+@mock.patch.object(BaseRemote, "_list_checksums_traverse")
+@mock.patch.object(BaseRemote, "_list_checksums_exists")
+def test_checksums_exist(object_exists, traverse, dvc):
     remote = BaseRemote(dvc, {})
 
     # remote does not support traverse
     remote.CAN_TRAVERSE = False
     with mock.patch.object(
-        remote, "cache_checksums", return_value=list(range(256))
+        remote, "list_checksums", return_value=list(range(256))
     ):
         checksums = set(range(1000))
-        remote.cache_exists(checksums)
+        remote.checksums_exist(checksums)
         object_exists.assert_called_with(checksums, None, None)
         traverse.assert_not_called()
 
@@ -62,10 +62,10 @@ def test_cache_exists(object_exists, traverse, dvc):
     object_exists.reset_mock()
     traverse.reset_mock()
     with mock.patch.object(
-        remote, "cache_checksums", return_value=list(range(256))
+        remote, "list_checksums", return_value=list(range(256))
     ):
         checksums = list(range(1000))
-        remote.cache_exists(checksums)
+        remote.checksums_exist(checksums)
         # verify that _cache_paths_with_max() short circuits
         # before returning all 256 remote checksums
         max_checksums = math.ceil(
@@ -83,10 +83,10 @@ def test_cache_exists(object_exists, traverse, dvc):
     traverse.reset_mock()
     remote.JOBS = 16
     with mock.patch.object(
-        remote, "cache_checksums", return_value=list(range(256))
+        remote, "list_checksums", return_value=list(range(256))
     ):
         checksums = list(range(1000000))
-        remote.cache_exists(checksums)
+        remote.checksums_exist(checksums)
         object_exists.assert_not_called()
         traverse.assert_called_with(
             256 * pow(16, remote.TRAVERSE_PREFIX_LEN),
@@ -97,44 +97,44 @@ def test_cache_exists(object_exists, traverse, dvc):
 
 
 @mock.patch.object(
-    BaseRemote, "cache_checksums", return_value=[],
+    BaseRemote, "list_checksums", return_value=[],
 )
 @mock.patch.object(
     BaseRemote, "path_to_checksum", side_effect=lambda x: x,
 )
-def test_cache_checksums_traverse(path_to_checksum, cache_checksums, dvc):
+def test_list_checksums_traverse(path_to_checksum, list_checksums, dvc):
     remote = BaseRemote(dvc, {})
     remote.tree.path_info = PathInfo("foo")
 
     # parallel traverse
     size = 256 / remote.JOBS * remote.LIST_OBJECT_PAGE_SIZE
-    list(remote._cache_checksums_traverse(size, {0}))
+    list(remote._list_checksums_traverse(size, {0}))
     for i in range(1, 16):
-        cache_checksums.assert_any_call(
+        list_checksums.assert_any_call(
             prefix=f"{i:03x}", progress_callback=CallableOrNone
         )
     for i in range(1, 256):
-        cache_checksums.assert_any_call(
+        list_checksums.assert_any_call(
             prefix=f"{i:02x}", progress_callback=CallableOrNone
         )
 
     # default traverse (small remote)
     size -= 1
-    cache_checksums.reset_mock()
-    list(remote._cache_checksums_traverse(size - 1, {0}))
-    cache_checksums.assert_called_with(
+    list_checksums.reset_mock()
+    list(remote._list_checksums_traverse(size - 1, {0}))
+    list_checksums.assert_called_with(
         prefix=None, progress_callback=CallableOrNone
     )
 
 
-def test_cache_checksums(dvc):
+def test_list_checksums(dvc):
     remote = BaseRemote(dvc, {})
     remote.tree.path_info = PathInfo("foo")
 
     with mock.patch.object(
-        remote, "list_cache_paths", return_value=["12/3456", "bar"]
+        remote, "list_paths", return_value=["12/3456", "bar"]
     ):
-        checksums = list(remote.cache_checksums())
+        checksums = list(remote.list_checksums())
         assert checksums == ["123456"]
 
 

tests/unit/remote/test_local.py

@@ -5,7 +5,7 @@ import pytest
 
 from dvc.cache import NamedCache
 from dvc.path_info import PathInfo
-from dvc.remote.local import LocalRemote
+from dvc.remote.local import LocalCache
 
 
 def test_status_download_optimization(mocker, dvc):
@@ -13,28 +13,28 @@ def test_status_download_optimization(mocker, dvc):
         And the desired files to fetch are already on the local cache,
         Don't check the existence of the desired files on the remote cache
     """
-    remote = LocalRemote(dvc, {})
+    cache = LocalCache(dvc, {})
 
     infos = NamedCache()
     infos.add("local", "acbd18db4cc2f85cedef654fccc4a4d8", "foo")
     infos.add("local", "37b51d194a7513e45b56f6524f2d51f2", "bar")
 
     local_exists = list(infos["local"])
-    mocker.patch.object(remote, "cache_exists", return_value=local_exists)
+    mocker.patch.object(cache, "checksums_exist", return_value=local_exists)
 
     other_remote = mocker.Mock()
     other_remote.url = "other_remote"
-    other_remote.cache_exists.return_value = []
+    other_remote.checksums_exist.return_value = []
 
-    remote.status(infos, other_remote, download=True)
+    cache.status(infos, other_remote, download=True)
 
-    assert other_remote.cache_exists.call_count == 0
+    assert other_remote.checksums_exist.call_count == 0
 
 
 @pytest.mark.parametrize("link_name", ["hardlink", "symlink"])
 def test_is_protected(tmp_dir, dvc, link_name):
-    remote = LocalRemote(dvc, {})
-    link_method = getattr(remote.tree, link_name)
+    cache = LocalCache(dvc, {})
+    link_method = getattr(cache.tree, link_name)
 
     (tmp_dir / "foo").write_text("foo")
 
@@ -43,47 +43,47 @@ def test_is_protected(tmp_dir, dvc, link_name):
 
     link_method(foo, link)
 
-    assert not remote.is_protected(foo)
-    assert not remote.is_protected(link)
+    assert not cache.is_protected(foo)
+    assert not cache.is_protected(link)
 
-    remote.protect(foo)
+    cache.protect(foo)
 
-    assert remote.is_protected(foo)
-    assert remote.is_protected(link)
+    assert cache.is_protected(foo)
+    assert cache.is_protected(link)
 
-    remote.unprotect(link)
+    cache.unprotect(link)
 
-    assert not remote.is_protected(link)
+    assert not cache.is_protected(link)
     if os.name == "nt" and link_name == "hardlink":
         # NOTE: NTFS doesn't allow deleting read-only files, which forces us to
         # set write perms on the link, which propagates to the source.
-        assert not remote.is_protected(foo)
+        assert not cache.is_protected(foo)
     else:
-        assert remote.is_protected(foo)
+        assert cache.is_protected(foo)
 
 
 @pytest.mark.parametrize("err", [errno.EPERM, errno.EACCES])
 def test_protect_ignore_errors(tmp_dir, mocker, err):
     tmp_dir.gen("foo", "foo")
     foo = PathInfo("foo")
-    remote = LocalRemote(None, {})
+    cache = LocalCache(None, {})
 
-    remote.protect(foo)
+    cache.protect(foo)
 
     mock_chmod = mocker.patch(
         "os.chmod", side_effect=OSError(err, "something")
     )
-    remote.protect(foo)
+    cache.protect(foo)
     assert mock_chmod.called
 
 
 def test_protect_ignore_erofs(tmp_dir, mocker):
     tmp_dir.gen("foo", "foo")
     foo = PathInfo("foo")
-    remote = LocalRemote(None, {})
+    cache = LocalCache(None, {})
 
     mock_chmod = mocker.patch(
         "os.chmod", side_effect=OSError(errno.EROFS, "read-only fs")
     )
-    remote.protect(foo)
+    cache.protect(foo)
     assert mock_chmod.called