Repository: iterative/dvc

Issue:
dvc add ignore unprotected files that have not been modified
DVC version 0.29.0 installed via pip on Linux

Following the sequence of commands
dvc unprotect data.dat
(modify data.dat)
dvc add data.dat

Correctly replaces dvc.dat with the correct kind of file link
but
dvc unprotect data.dat
(not modifying)
dvc add data.dat

Causes add to skip data.dat because it is already cached and leaves the file on disk without linking. Presumably the ideal behavior is to have dvc add check what kind of link the cache should use and update the file to that link even if the file was not modified. This way modified and unmodified data have the same behavior.

Currently the workaround is to rm the data file and checkout which is somewhat awkward.



Discussion:
Looks like it is a side effect of https://github.com/iterative/dvc/blob/0.29.0/dvc/stage.py#L423 . We should probably call `save()` on that stage if we see that it is already cached, just to ensure that all the cache is optimized. 
@efiop I am having problems with the work around of rm'ing the file and checking it out. If I have more than one such file, "dvc checkout" will only checkout file1.dat: and then quit with an error message "Error: unexpected error - [errno 1] Operation not permitted: file2.dat". If I call dvc checkout n'times where n is the number of files I can get all of them but this is very very tedious.
@rmcavoy So sorry for the delay! Notification got lost in the flow :slightly_frowning_face: . Does that issue still persist?

Patch:
diff --git a/dvc/analytics.py b/dvc/analytics.py
--- a/dvc/analytics.py
+++ b/dvc/analytics.py
@@ -96,11 +96,12 @@ def _read_user_id(self):
         with open(self.user_id_file, "r") as fobj:
             try:
                 info = json.load(fobj)
-                return info[self.PARAM_USER_ID]
-            except json.JSONDecodeError as exc:
+            except ValueError as exc:
                 logger.debug("Failed to load user_id: {}".format(exc))
                 return None
 
+            return info[self.PARAM_USER_ID]
+
     def _get_user_id(self):
         from dvc.lock import LockError
 
diff --git a/dvc/dependency/http.py b/dvc/dependency/http.py
--- a/dvc/dependency/http.py
+++ b/dvc/dependency/http.py
@@ -16,4 +16,5 @@ def __init__(self, stage, path, info=None, remote=None):
         if path.startswith("remote"):
             path = urljoin(self.remote.cache_dir, urlparse(path).path)
 
-        self.path_info = {"scheme": urlparse(path).scheme, "path": path}
+        self.path_info["scheme"] = urlparse(path).scheme
+        self.path_info["path"] = path
diff --git a/dvc/output/base.py b/dvc/output/base.py
--- a/dvc/output/base.py
+++ b/dvc/output/base.py
@@ -1,6 +1,7 @@
 from __future__ import unicode_literals
 
 import re
+import logging
 from schema import Or, Optional
 
 from dvc.exceptions import DvcException
@@ -8,6 +9,9 @@
 from dvc.remote.base import RemoteBase
 
 
+logger = logging.getLogger(__name__)
+
+
 class OutputDoesNotExistError(DvcException):
     def __init__(self, path):
         msg = "output '{}' does not exist".format(path)
@@ -73,16 +77,15 @@ def __init__(
         self.persist = persist
         self.tags = None if self.IS_DEPENDENCY else (tags or {})
 
-        if (
-            self.use_cache
-            and getattr(self.repo.cache, self.REMOTE.scheme) is None
-        ):
+        if self.use_cache and self.cache is None:
             raise DvcException(
                 "no cache location setup for '{}' outputs.".format(
                     self.REMOTE.scheme
                 )
             )
 
+        self.path_info = {"scheme": self.scheme, "url": self.url}
+
     def __repr__(self):
         return "{class_name}: '{url}'".format(
             class_name=type(self).__name__, url=(self.url or "No url")
@@ -95,6 +98,14 @@ def __str__(self):
     def is_in_repo(self):
         return False
 
+    @property
+    def cache(self):
+        return getattr(self.repo.cache, self.scheme)
+
+    @property
+    def dir_cache(self):
+        return self.cache.get_dir_cache(self.checksum)
+
     def assign_to_stage_file(self, target_repo):
         raise DvcException(
             "change repo is not supported for {}".format(self.scheme)
@@ -122,6 +133,10 @@ def scheme(self):
     def path(self):
         return self.path_info["path"]
 
+    @property
+    def cache_path(self):
+        return self.cache.checksum_to_path(self.checksum)
+
     @property
     def sep(self):
         return "/"
@@ -130,6 +145,10 @@ def sep(self):
     def checksum(self):
         return self.info.get(self.remote.PARAM_CHECKSUM)
 
+    @property
+    def is_dir_checksum(self):
+        return self.remote.is_dir_checksum(self.checksum)
+
     @property
     def exists(self):
         return self.remote.exists(self.path_info)
@@ -146,9 +165,7 @@ def changed_cache(self):
         if not self.use_cache or not self.checksum:
             return True
 
-        cache = self.repo.cache.__getattribute__(self.scheme)
-
-        return cache.changed_cache(self.checksum)
+        return self.cache.changed_cache(self.checksum)
 
     def status(self):
         if self.checksum and self.use_cache and self.changed_cache():
@@ -166,19 +183,62 @@ def status(self):
         return {}
 
     def changed(self):
-        return bool(self.status())
+        status = self.status()
+        logger.debug(str(status))
+        return bool(status)
+
+    @property
+    def is_empty(self):
+        return self.remote.is_empty(self.path_info)
+
+    def isdir(self):
+        return self.remote.isdir(self.path_info)
+
+    def isfile(self):
+        return self.remote.isfile(self.path_info)
 
     def save(self):
         if not self.exists:
             raise self.DoesNotExistError(self)
 
+        if not self.isfile and not self.isdir:
+            raise self.IsNotFileOrDirError(self)
+
+        if self.is_empty:
+            logger.warning("'{}' is empty.".format(self))
+
+        if not self.use_cache:
+            self.info = self.remote.save_info(self.path_info)
+            if self.metric:
+                self.verify_metric()
+            if not self.IS_DEPENDENCY:
+                logger.info(
+                    "Output '{}' doesn't use cache. Skipping saving.".format(
+                        self
+                    )
+                )
+            return
+
+        assert not self.IS_DEPENDENCY
+
+        if not self.changed():
+            logger.info(
+                "Output '{}' didn't change. Skipping saving.".format(self)
+            )
+            return
+
+        if self.is_in_repo:
+            if self.repo.scm.is_tracked(self.path):
+                raise OutputAlreadyTrackedError(self)
+
+            if self.use_cache:
+                self.repo.scm.ignore(self.path)
+
         self.info = self.remote.save_info(self.path_info)
 
     def commit(self):
         if self.use_cache:
-            getattr(self.repo.cache, self.scheme).save(
-                self.path_info, self.info
-            )
+            self.cache.save(self.path_info, self.info)
 
     def dumpd(self):
         ret = self.info.copy()
@@ -221,7 +281,7 @@ def checkout(self, force=False, progress_callback=None, tag=None):
         else:
             info = self.info
 
-        getattr(self.repo.cache, self.scheme).checkout(
+        self.cache.checkout(
             self.path_info,
             info,
             force=force,
@@ -242,7 +302,7 @@ def move(self, out):
 
         self.remote.move(self.path_info, out.path_info)
         self.url = out.url
-        self.path_info = out.path_info
+        self.path_info = out.path_info.copy()
         self.save()
         self.commit()
 
@@ -250,10 +310,13 @@ def move(self, out):
             self.repo.scm.ignore(self.path)
 
     def get_files_number(self):
-        if self.use_cache and self.checksum:
-            return 1
+        if not self.use_cache or not self.checksum:
+            return 0
+
+        if self.is_dir_checksum:
+            return len(self.dir_cache)
 
-        return 0
+        return 1
 
     def unprotect(self):
         if self.exists:
diff --git a/dvc/output/hdfs.py b/dvc/output/hdfs.py
--- a/dvc/output/hdfs.py
+++ b/dvc/output/hdfs.py
@@ -34,4 +34,5 @@ def __init__(
         if remote:
             path = posixpath.join(remote.url, urlparse(path).path.lstrip("/"))
         user = remote.user if remote else self.group("user")
-        self.path_info = {"scheme": "hdfs", "user": user, "path": path}
+        self.path_info["user"] = user
+        self.path_info["path"] = path
diff --git a/dvc/output/local.py b/dvc/output/local.py
--- a/dvc/output/local.py
+++ b/dvc/output/local.py
@@ -7,7 +7,7 @@
 from dvc.istextfile import istextfile
 from dvc.exceptions import DvcException
 from dvc.remote.local import RemoteLOCAL
-from dvc.output.base import OutputBase, OutputAlreadyTrackedError
+from dvc.output.base import OutputBase
 
 
 logger = logging.getLogger(__name__)
@@ -49,9 +49,7 @@ def __init__(
             p = os.path.join(stage.wdir, p)
         p = os.path.abspath(os.path.normpath(p))
 
-        self.path_info = {"scheme": "local", "path": p}
-
-        self._dir_cache = {}
+        self.path_info["path"] = p
 
     def __str__(self):
         return self.rel_path
@@ -85,18 +83,10 @@ def rel_path(self):
     def stage_path(self):
         return os.path.relpath(self.path, self.stage.wdir)
 
-    @property
-    def cache(self):
-        return self.repo.cache.local.get(self.checksum)
-
-    @property
-    def is_dir_cache(self):
-        return self.repo.cache.local.is_dir_cache(self.checksum)
-
     def dumpd(self):
         ret = super(OutputLOCAL, self).dumpd()
         if self.is_in_repo:
-            path = self.remote.unixpath(
+            path = self.remote.to_posixpath(
                 os.path.relpath(self.path, self.stage.wdir)
             )
         else:
@@ -120,52 +110,3 @@ def verify_metric(self):
         if not istextfile(self.path):
             msg = "binary file '{}' cannot be used as metrics."
             raise DvcException(msg.format(self.rel_path))
-
-    def save(self):
-        if not os.path.exists(self.path):
-            raise self.DoesNotExistError(self.rel_path)
-
-        if not os.path.isfile(self.path) and not os.path.isdir(self.path):
-            raise self.IsNotFileOrDirError(self.rel_path)
-
-        if (os.path.isfile(self.path) and os.path.getsize(self.path) == 0) or (
-            os.path.isdir(self.path) and len(os.listdir(self.path)) == 0
-        ):
-            msg = "file/directory '{}' is empty.".format(self.rel_path)
-            logger.warning(msg)
-
-        if not self.use_cache:
-            self.info = self.remote.save_info(self.path_info)
-            self.verify_metric()
-            if not self.IS_DEPENDENCY:
-                msg = "Output '{}' doesn't use cache. Skipping saving."
-                logger.info(msg.format(self.rel_path))
-            return
-
-        assert not self.IS_DEPENDENCY
-
-        if not self.changed():
-            msg = "Output '{}' didn't change. Skipping saving."
-            logger.info(msg.format(self.rel_path))
-            return
-
-        if self.is_in_repo:
-            if self.repo.scm.is_tracked(self.path):
-                raise OutputAlreadyTrackedError(self.rel_path)
-
-            if self.use_cache:
-                self.repo.scm.ignore(self.path)
-
-        self.info = self.remote.save_info(self.path_info)
-
-    @property
-    def dir_cache(self):
-        return self.repo.cache.local.load_dir_cache(self.checksum)
-
-    def get_files_number(self):
-        if self.cache is None:
-            return 0
-
-        if self.is_dir_cache:
-            return len(self.dir_cache)
-        return 1
diff --git a/dvc/output/s3.py b/dvc/output/s3.py
--- a/dvc/output/s3.py
+++ b/dvc/output/s3.py
@@ -35,8 +35,6 @@ def __init__(
         path = urlparse(path).path.lstrip("/")
         if remote:
             path = posixpath.join(remote.prefix, path)
-        self.path_info = {
-            "scheme": self.scheme,
-            "bucket": bucket,
-            "path": path,
-        }
+
+        self.path_info["bucket"] = bucket
+        self.path_info["path"] = path
diff --git a/dvc/output/ssh.py b/dvc/output/ssh.py
--- a/dvc/output/ssh.py
+++ b/dvc/output/ssh.py
@@ -48,10 +48,7 @@ def __init__(
         else:
             path = parsed.path
 
-        self.path_info = {
-            "scheme": "ssh",
-            "host": host,
-            "port": port,
-            "user": user,
-            "path": path,
-        }
+        self.path_info["host"] = host
+        self.path_info["port"] = port
+        self.path_info["user"] = user
+        self.path_info["path"] = path
diff --git a/dvc/remote/azure.py b/dvc/remote/azure.py
--- a/dvc/remote/azure.py
+++ b/dvc/remote/azure.py
@@ -120,7 +120,7 @@ def _list_paths(self, bucket, prefix):
     def list_cache_paths(self):
         return self._list_paths(self.bucket, self.prefix)
 
-    def upload(self, from_infos, to_infos, names=None):
+    def upload(self, from_infos, to_infos, names=None, no_progress_bar=False):
         names = self._verify_path_args(to_infos, from_infos, names)
 
         for from_info, to_info, name in zip(from_infos, to_infos, names):
@@ -142,7 +142,7 @@ def upload(self, from_infos, to_infos, names=None):
             if not name:
                 name = os.path.basename(from_info["path"])
 
-            cb = Callback(name)
+            cb = None if no_progress_bar else Callback(name)
 
             try:
                 self.blob_service.create_blob_from_path(
diff --git a/dvc/remote/base.py b/dvc/remote/base.py
--- a/dvc/remote/base.py
+++ b/dvc/remote/base.py
@@ -2,14 +2,21 @@
 
 from dvc.utils.compat import str
 
+import os
 import re
-import posixpath
+import json
+import ntpath
 import logging
+import tempfile
+import posixpath
+from operator import itemgetter
 from multiprocessing import cpu_count
 
 import dvc.prompt as prompt
 from dvc.config import Config
 from dvc.exceptions import DvcException, ConfirmRemoveError
+from dvc.utils import tmp_fname
+from dvc.state import StateBase
 
 
 logger = logging.getLogger(__name__)
@@ -61,6 +68,9 @@ class RemoteBase(object):
     REQUIRES = {}
     JOBS = 4 * cpu_count()
 
+    PARAM_RELPATH = "relpath"
+    CHECKSUM_DIR_SUFFIX = ".dir"
+
     def __init__(self, repo, config):
         self.repo = repo
         deps_ok = all(self.REQUIRES.values())
@@ -89,6 +99,11 @@ def __init__(self, repo, config):
             ).format(url, missing, " ".join(missing), self.scheme)
             raise RemoteMissingDepsError(msg)
 
+        self.protected = False
+        self.state = StateBase()
+
+        self._dir_info = {}
+
     def __repr__(self):
         return "{class_name}: '{url}'".format(
             class_name=type(self).__name__, url=(self.url or "No url")
@@ -102,6 +117,27 @@ def supported(cls, config):
         url = config[Config.SECTION_REMOTE_URL]
         return cls.match(url) is not None
 
+    @staticmethod
+    def _replace_sep(path, from_sep, to_sep):
+        assert not ntpath.isabs(path)
+        assert not posixpath.isabs(path)
+        return path.replace(from_sep, to_sep)
+
+    @classmethod
+    def to_posixpath(cls, path):
+        return cls._replace_sep(path, "\\", "/")
+
+    @classmethod
+    def to_ntpath(cls, path):
+        return cls._replace_sep(path, "/", "\\")
+
+    def to_ospath(self, path):
+        if self.ospath == ntpath:
+            return self.to_ntpath(path)
+        elif self.ospath == posixpath:
+            return self.to_posixpath(path)
+        assert False
+
     @classmethod
     def match(cls, url):
         return re.match(cls.REGEX, url)
@@ -112,9 +148,125 @@ def group(self, name):
             return None
         return m.group(name)
 
-    def save_info(self, path_info):
+    @property
+    def cache(self):
+        return getattr(self.repo.cache, self.scheme)
+
+    def get_file_checksum(self, path_info):
         raise NotImplementedError
 
+    def _collect_dir(self, path_info):
+        dir_info = []
+
+        p_info = path_info.copy()
+        dpath = p_info["path"]
+        for root, dirs, files in self.walk(path_info):
+            for fname in files:
+                path = self.ospath.join(root, fname)
+                p_info["path"] = path
+                relpath = self.to_posixpath(self.ospath.relpath(path, dpath))
+
+                checksum = self.get_file_checksum(p_info)
+                dir_info.append(
+                    {
+                        self.PARAM_RELPATH: relpath,
+                        self.PARAM_CHECKSUM: checksum,
+                    }
+                )
+
+        # NOTE: sorting the list by path to ensure reproducibility
+        return sorted(dir_info, key=itemgetter(self.PARAM_RELPATH))
+
+    def get_dir_checksum(self, path_info):
+        dir_info = self._collect_dir(path_info)
+        to_info = path_info.copy()
+        to_info["path"] = self.cache.ospath.join(
+            self.cache.prefix, tmp_fname("")
+        )
+
+        tmp = tempfile.NamedTemporaryFile(delete=False).name
+        with open(tmp, "w+") as fobj:
+            json.dump(dir_info, fobj)
+
+        from_info = {"scheme": "local", "path": tmp}
+        self.cache.upload([from_info], [to_info], no_progress_bar=True)
+
+        checksum = self.get_file_checksum(to_info) + self.CHECKSUM_DIR_SUFFIX
+
+        from_info = to_info.copy()
+        to_info["path"] = self.cache.checksum_to_path(checksum)
+        if self.cache.changed_cache_file(checksum):
+            self.cache.move(from_info, to_info)
+
+        self.state.save(path_info, checksum)
+        self.state.save(to_info, checksum)
+
+        return checksum
+
+    def get_dir_cache(self, checksum):
+        assert checksum
+
+        dir_info = self._dir_info.get(checksum)
+        if dir_info:
+            return dir_info
+
+        dir_info = self.load_dir_cache(checksum)
+        self._dir_info[checksum] = dir_info
+        return dir_info
+
+    def load_dir_cache(self, checksum):
+        path_info = self.checksum_to_path_info(checksum)
+
+        fobj = tempfile.NamedTemporaryFile(delete=False)
+        path = fobj.name
+        to_info = {"scheme": "local", "path": path}
+        self.cache.download([path_info], [to_info], no_progress_bar=True)
+
+        try:
+            with open(path, "r") as fobj:
+                d = json.load(fobj)
+        except ValueError:
+            logger.exception("Failed to load dir cache '{}'".format(path_info))
+            return []
+        finally:
+            os.unlink(path)
+
+        if not isinstance(d, list):
+            msg = "dir cache file format error '{}' [skipping the file]"
+            logger.error(msg.format(os.path.relpath(path)))
+            return []
+
+        for info in d:
+            info[self.PARAM_RELPATH] = self.to_ospath(info[self.PARAM_RELPATH])
+
+        return d
+
+    @classmethod
+    def is_dir_checksum(cls, checksum):
+        return checksum.endswith(cls.CHECKSUM_DIR_SUFFIX)
+
+    def get_checksum(self, path_info):
+        if not self.exists(path_info):
+            return None
+
+        checksum = self.state.get(path_info)
+        if checksum:
+            return checksum
+
+        if self.isdir(path_info):
+            checksum = self.get_dir_checksum(path_info)
+        else:
+            checksum = self.get_file_checksum(path_info)
+
+        if checksum:
+            self.state.save(path_info, checksum)
+
+        return checksum
+
+    def save_info(self, path_info):
+        assert path_info["scheme"] == self.scheme
+        return {self.PARAM_CHECKSUM: self.get_checksum(path_info)}
+
     def changed(self, path_info, checksum_info):
         """Checks if data has changed.
 
@@ -165,6 +317,61 @@ def changed(self, path_info, checksum_info):
         logger.debug("'{}' hasn't changed.".format(path_info))
         return False
 
+    def link(self, from_info, to_info):
+        self.copy(from_info, to_info)
+
+    def _save_file(self, path_info, checksum, save_link=True):
+        assert checksum
+
+        cache_info = self.checksum_to_path_info(checksum)
+        if self.changed_cache(checksum):
+            self.move(path_info, cache_info)
+        else:
+            self.remove(path_info)
+
+        self.link(cache_info, path_info)
+
+        if save_link:
+            self.state.save_link(path_info)
+
+        # we need to update path and cache, since in case of reflink,
+        # or copy cache type moving original file results in updates on
+        # next executed command, which causes md5 recalculation
+        self.state.save(path_info, checksum)
+        self.state.save(cache_info, checksum)
+
+    def _save_dir(self, path_info, checksum):
+        cache_info = self.checksum_to_path_info(checksum)
+        dir_info = self.get_dir_cache(checksum)
+
+        entry_info = path_info.copy()
+        for entry in dir_info:
+            entry_checksum = entry[self.PARAM_CHECKSUM]
+            entry_info["path"] = self.ospath.join(
+                path_info["path"], entry[self.PARAM_RELPATH]
+            )
+
+            self._save_file(entry_info, entry_checksum, save_link=False)
+
+        self.state.save_link(path_info)
+        self.state.save(cache_info, checksum)
+        self.state.save(path_info, checksum)
+
+    def is_empty(self, path_info):
+        return False
+
+    def isfile(self, path_info):
+        raise NotImplementedError
+
+    def isdir(self, path_info):
+        return False
+
+    def walk(self, path_info):
+        raise NotImplementedError
+
+    def protect(self, path_info):
+        pass
+
     def save(self, path_info, checksum_info):
         if path_info["scheme"] != self.scheme:
             raise RemoteActionNotImplemented(
@@ -174,13 +381,18 @@ def save(self, path_info, checksum_info):
 
         checksum = checksum_info[self.PARAM_CHECKSUM]
         if not self.changed_cache(checksum):
+            self._checkout(path_info, checksum)
             return
 
-        to_info = self.checksum_to_path_info(checksum)
+        self._save(path_info, checksum)
 
+    def _save(self, path_info, checksum):
+        to_info = self.checksum_to_path_info(checksum)
         logger.info("Saving '{}' to '{}'.".format(path_info, to_info))
-
-        self.copy(path_info, to_info)
+        if self.isdir(path_info):
+            self._save_dir(path_info, checksum)
+            return
+        self._save_file(path_info, checksum)
 
     def download(
         self,
@@ -192,14 +404,15 @@ def download(
     ):
         raise RemoteActionNotImplemented("download", self.scheme)
 
-    def upload(self, from_infos, to_infos, names=None):
+    def upload(self, from_infos, to_infos, names=None, no_progress_bar=False):
         raise RemoteActionNotImplemented("upload", self.scheme)
 
     def remove(self, path_info):
         raise RemoteActionNotImplemented("remove", self.scheme)
 
-    def move(self, path_info):
-        raise RemoteActionNotImplemented("move", self.scheme)
+    def move(self, from_info, to_info):
+        self.copy(from_info, to_info)
+        self.remove(from_info)
 
     def copy(self, from_info, to_info):
         raise RemoteActionNotImplemented("copy", self.scheme)
@@ -226,6 +439,7 @@ def ospath(self):
         return posixpath
 
     def checksum_to_path(self, checksum):
+        assert checksum
         return self.ospath.join(self.prefix, checksum[0:2], checksum[2:])
 
     def path_to_checksum(self, path):
@@ -269,28 +483,51 @@ def gc(self, cinfos):
             removed = True
         return removed
 
-    def changed_cache(self, checksum):
-        cache = self.checksum_to_path_info(checksum)
-        expected = {self.PARAM_CHECKSUM: checksum}
+    def changed_cache_file(self, checksum):
+        cache_info = self.checksum_to_path_info(checksum)
+        actual = self.get_checksum(cache_info)
+
+        logger.debug(
+            "cache '{}' expected '{}' actual '{}'".format(
+                str(cache_info), checksum, actual
+            )
+        )
 
-        if not self.exists(cache):
+        if not checksum or not actual:
             return True
 
-        actual = self.save_info(cache)
+        if actual.split(".")[0] == checksum.split(".")[0]:
+            return False
 
-        logger.debug(
-            "Cache '{}' actual '{}'.".format(str(expected), str(actual))
-        )
+        if self.exists(cache_info):
+            logger.warning("corrupted cache file '{}'.".format(cache_info))
+            self.remove(cache_info)
+
+        return True
 
-        if expected != actual:
-            if self.exists(cache):
-                msg = "corrupted cache file {}"
-                logger.warning(msg.format(str(cache)))
-                self.remove(cache)
+    def _changed_cache_dir(self):
+        # NOTE: only implemented for RemoteLOCAL
+        return True
+
+    def _changed_dir_cache(self, checksum):
+        if not self._changed_cache_dir():
+            return False
+
+        if self.changed_cache_file(checksum):
             return True
 
+        for entry in self.get_dir_cache(checksum):
+            checksum = entry[self.PARAM_CHECKSUM]
+            if self.changed_cache_file(checksum):
+                return True
+
         return False
 
+    def changed_cache(self, checksum):
+        if self.is_dir_checksum(checksum):
+            return self._changed_dir_cache(checksum)
+        return self.changed_cache_file(checksum)
+
     def cache_exists(self, checksums):
         # NOTE: The reason for such an odd logic is that most of the remotes
         # take much shorter time to just retrieve everything they have under
@@ -303,7 +540,7 @@ def cache_exists(self, checksums):
         return list(filter(lambda checksum: checksum in checksums, self.all()))
 
     def already_cached(self, path_info):
-        current = self.save_info(path_info)[self.PARAM_CHECKSUM]
+        current = self.get_checksum(path_info)
 
         if not current:
             return False
@@ -325,16 +562,73 @@ def safe_remove(self, path_info, force=False):
 
         self.remove(path_info)
 
-    def do_checkout(
-        self, path_info, checksum, force=False, progress_callback=None
+    def _checkout_file(
+        self, path_info, checksum, force, progress_callback=None
     ):
+        cache_info = self.checksum_to_path_info(checksum)
         if self.exists(path_info):
             msg = "data '{}' exists. Removing before checkout."
             logger.warning(msg.format(str(path_info)))
             self.safe_remove(path_info, force=force)
 
-        from_info = self.checksum_to_path_info(checksum)
-        self.copy(from_info, path_info)
+        self.link(cache_info, path_info)
+        self.state.save_link(path_info)
+        if progress_callback:
+            progress_callback.update(path_info["url"])
+
+    def makedirs(self, path_info):
+        raise NotImplementedError
+
+    def _checkout_dir(
+        self, path_info, checksum, force, progress_callback=None
+    ):
+        # Create dir separately so that dir is created
+        # even if there are no files in it
+        if not self.exists(path_info):
+            self.makedirs(path_info)
+
+        dir_info = self.get_dir_cache(checksum)
+
+        logger.debug("Linking directory '{}'.".format(path_info))
+
+        entry_info = path_info.copy()
+        for entry in dir_info:
+            relpath = entry[self.PARAM_RELPATH]
+            checksum = entry[self.PARAM_CHECKSUM]
+            entry_cache_info = self.checksum_to_path_info(checksum)
+            entry_info["url"] = self.ospath.join(path_info["url"], relpath)
+            entry_info["path"] = self.ospath.join(path_info["path"], relpath)
+
+            entry_checksum_info = {self.PARAM_CHECKSUM: checksum}
+            if self.changed(entry_info, entry_checksum_info):
+                if self.exists(entry_info):
+                    self.safe_remove(entry_info, force=force)
+                self.link(entry_cache_info, entry_info)
+            if progress_callback:
+                progress_callback.update(entry_info["url"])
+
+        self._remove_redundant_files(path_info, dir_info, force)
+
+        self.state.save_link(path_info)
+
+    def _remove_redundant_files(self, path_info, dir_info, force):
+        existing_files = set(
+            self.ospath.join(root, fname)
+            for root, _, files in self.walk(path_info)
+            for fname in files
+        )
+
+        needed_files = set(
+            self.ospath.join(path_info["path"], entry[self.PARAM_RELPATH])
+            for entry in dir_info
+        )
+
+        delta = existing_files - needed_files
+
+        d_info = path_info.copy()
+        for path in delta:
+            d_info["path"] = path
+            self.safe_remove(d_info, force)
 
     def checkout(
         self, path_info, checksum_info, force=False, progress_callback=None
@@ -363,11 +657,17 @@ def checkout(
         msg = "Checking out '{}' with cache '{}'."
         logger.debug(msg.format(str(path_info), checksum))
 
-        self.do_checkout(
-            path_info,
-            checksum,
-            force=force,
-            progress_callback=progress_callback,
+        return self._checkout(path_info, checksum, force, progress_callback)
+
+    def _checkout(
+        self, path_info, checksum, force=False, progress_callback=None
+    ):
+        if not self.is_dir_checksum(checksum):
+            return self._checkout_file(
+                path_info, checksum, force, progress_callback=progress_callback
+            )
+        return self._checkout_dir(
+            path_info, checksum, force, progress_callback=progress_callback
         )
 
     @staticmethod
diff --git a/dvc/remote/gs.py b/dvc/remote/gs.py
--- a/dvc/remote/gs.py
+++ b/dvc/remote/gs.py
@@ -55,10 +55,12 @@ def gs(self):
             else storage.Client(self.projectname)
         )
 
-    def get_md5(self, bucket, 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
@@ -67,16 +69,6 @@ def get_md5(self, bucket, path):
         md5 = base64.b64decode(b64_md5)
         return codecs.getencoder("hex")(md5)[0].decode("utf-8")
 
-    def save_info(self, path_info):
-        if path_info["scheme"] != "gs":
-            raise NotImplementedError
-
-        return {
-            self.PARAM_CHECKSUM: self.get_md5(
-                path_info["bucket"], path_info["path"]
-            )
-        }
-
     def copy(self, from_info, to_info, gs=None):
         gs = gs if gs else self.gs
 
@@ -120,7 +112,7 @@ def exists(self, path_info):
         paths = self._list_paths(path_info["bucket"], path_info["path"])
         return any(path_info["path"] == path for path in paths)
 
-    def upload(self, from_infos, to_infos, names=None):
+    def upload(self, from_infos, to_infos, names=None, no_progress_bar=False):
         names = self._verify_path_args(to_infos, from_infos, names)
 
         gs = self.gs
@@ -141,7 +133,8 @@ def upload(self, from_infos, to_infos, names=None):
             if not name:
                 name = os.path.basename(from_info["path"])
 
-            progress.update_target(name, 0, None)
+            if not no_progress_bar:
+                progress.update_target(name, 0, None)
 
             try:
                 bucket = gs.bucket(to_info["bucket"])
diff --git a/dvc/remote/hdfs.py b/dvc/remote/hdfs.py
--- a/dvc/remote/hdfs.py
+++ b/dvc/remote/hdfs.py
@@ -63,7 +63,7 @@ def _group(regex, s, gname):
         assert match is not None
         return match.group(gname)
 
-    def checksum(self, path_info):
+    def get_file_checksum(self, path_info):
         regex = r".*\t.*\t(?P<checksum>.*)"
         stdout = self.hadoop_fs(
             "checksum {}".format(path_info["path"]), user=path_info["user"]
@@ -83,14 +83,6 @@ def rm(self, path_info):
             "rm -f {}".format(path_info["path"]), user=path_info["user"]
         )
 
-    def save_info(self, path_info):
-        if path_info["scheme"] != "hdfs":
-            raise NotImplementedError
-
-        assert path_info.get("path")
-
-        return {self.PARAM_CHECKSUM: self.checksum(path_info)}
-
     def remove(self, path_info):
         if path_info["scheme"] != "hdfs":
             raise NotImplementedError
@@ -111,7 +103,7 @@ def exists(self, path_info):
         except RemoteCmdError:
             return False
 
-    def upload(self, from_infos, to_infos, names=None):
+    def upload(self, from_infos, to_infos, names=None, no_progress_bar=False):
         names = self._verify_path_args(to_infos, from_infos, names)
 
         for from_info, to_info, name in zip(from_infos, to_infos, names):
diff --git a/dvc/remote/http.py b/dvc/remote/http.py
--- a/dvc/remote/http.py
+++ b/dvc/remote/http.py
@@ -109,16 +109,11 @@ def func(md5):
 
         return list(filter(func, md5s))
 
-    def save_info(self, path_info):
-        if path_info["scheme"] not in ["http", "https"]:
-            raise NotImplementedError
-
-        return {self.PARAM_CHECKSUM: self._etag(path_info["path"])}
-
     def _content_length(self, url):
         return self._request("HEAD", url).headers.get("Content-Length")
 
-    def _etag(self, url):
+    def get_file_checksum(self, path_info):
+        url = path_info["path"]
         etag = self._request("HEAD", url).headers.get("ETag") or self._request(
             "HEAD", url
         ).headers.get("Content-MD5")
diff --git a/dvc/remote/local.py b/dvc/remote/local.py
--- a/dvc/remote/local.py
+++ b/dvc/remote/local.py
@@ -7,12 +7,10 @@
 import os
 import stat
 import uuid
-import json
 import ntpath
 import shutil
 import posixpath
 import logging
-from operator import itemgetter
 
 from dvc.system import System
 from dvc.remote.base import (
@@ -26,12 +24,11 @@
     remove,
     move,
     copyfile,
-    dict_md5,
     to_chunks,
     tmp_fname,
+    file_md5,
     walk_files,
 )
-from dvc.utils import LARGE_DIR_SIZE
 from dvc.config import Config
 from dvc.exceptions import DvcException
 from dvc.progress import progress
@@ -48,8 +45,6 @@ class RemoteLOCAL(RemoteBase):
     REGEX = r"^(?P<path>.*)$"
     PARAM_CHECKSUM = "md5"
     PARAM_PATH = "path"
-    PARAM_RELPATH = "relpath"
-    MD5_DIR_SUFFIX = ".dir"
 
     DEFAULT_CACHE_TYPES = ["reflink", "copy"]
     CACHE_TYPE_MAP = {
@@ -116,44 +111,21 @@ def get(self, md5):
         if not md5:
             return None
 
-        return os.path.join(self.cache_dir, md5[0:2], md5[2:])
-
-    def changed_cache_file(self, md5):
-        cache = self.get(md5)
-        if self.state.changed(cache, md5=md5):
-            if os.path.exists(cache):
-                msg = "Corrupted cache file {}."
-                logger.warning(msg.format(os.path.relpath(cache)))
-                remove(cache)
-            return True
-        return False
+        return self.checksum_to_path(md5)
 
     def exists(self, path_info):
         assert not isinstance(path_info, list)
         assert path_info["scheme"] == "local"
         return os.path.lexists(path_info["path"])
 
-    def changed_cache(self, md5):
-        cache = self.get(md5)
-        clist = [(cache, md5)]
-
-        while True:
-            if len(clist) == 0:
-                break
+    def makedirs(self, path_info):
+        if not self.exists(path_info):
+            os.makedirs(path_info["path"])
 
-            cache, md5 = clist.pop()
-            if self.changed_cache_file(md5):
-                return True
-
-            if self.is_dir_cache(cache) and self._cache_metadata_changed():
-                for entry in self.load_dir_cache(md5):
-                    md5 = entry[self.PARAM_CHECKSUM]
-                    cache = self.get(md5)
-                    clist.append((cache, md5))
-
-        return False
+    def link(self, cache_info, path_info):
+        cache = cache_info["path"]
+        path = path_info["path"]
 
-    def link(self, cache, path):
         assert os.path.isfile(cache)
 
         dname = os.path.dirname(path)
@@ -174,7 +146,7 @@ def link(self, cache, path):
                 self.CACHE_TYPE_MAP[self.cache_types[0]](cache, path)
 
                 if self.protected:
-                    os.chmod(path, stat.S_IREAD | stat.S_IRGRP | stat.S_IROTH)
+                    self.protect(path_info)
 
                 msg = "Created {}'{}': {} -> {}".format(
                     "protected " if self.protected else "",
@@ -200,276 +172,38 @@ def ospath(self):
             return ntpath
         return posixpath
 
-    @classmethod
-    def to_ospath(cls, path):
-        if os.name == "nt":
-            return cls.ntpath(path)
-        return cls.unixpath(path)
-
-    @staticmethod
-    def unixpath(path):
-        assert not ntpath.isabs(path)
-        assert not posixpath.isabs(path)
-        return path.replace("\\", "/")
-
-    @staticmethod
-    def ntpath(path):
-        assert not ntpath.isabs(path)
-        assert not posixpath.isabs(path)
-        return path.replace("/", "\\")
-
-    def collect_dir_cache(self, dname):
-        dir_info = []
-
-        for root, dirs, files in os.walk(str(dname)):
-            bar = False
-
-            if len(files) > LARGE_DIR_SIZE:
-                msg = (
-                    "Computing md5 for a large directory {}. "
-                    "This is only done once."
-                )
-                logger.info(msg.format(os.path.relpath(root)))
-                bar = True
-                title = os.path.relpath(root)
-                processed = 0
-                total = len(files)
-                progress.update_target(title, 0, total)
-
-            for fname in files:
-                path = os.path.join(root, fname)
-                relpath = self.unixpath(os.path.relpath(path, dname))
-
-                if bar:
-                    progress.update_target(title, processed, total)
-                    processed += 1
-
-                md5 = self.state.update(path)
-                dir_info.append(
-                    {self.PARAM_RELPATH: relpath, self.PARAM_CHECKSUM: md5}
-                )
-
-            if bar:
-                progress.finish_target(title)
-
-        # NOTE: sorting the list by path to ensure reproducibility
-        dir_info = sorted(dir_info, key=itemgetter(self.PARAM_RELPATH))
-
-        md5 = dict_md5(dir_info) + self.MD5_DIR_SUFFIX
-        if self.changed_cache_file(md5):
-            self.dump_dir_cache(md5, dir_info)
-
-        return (md5, dir_info)
-
-    def load_dir_cache(self, md5):
-        path = self.get(md5)
-        assert self.is_dir_cache(path)
-
-        dir_info = self._dir_info.get(md5)
-        if dir_info:
-            return dir_info
-
-        try:
-            with open(path, "r") as fd:
-                d = json.load(fd)
-        except Exception:
-            msg = "Failed to load dir cache '{}'"
-            logger.exception(msg.format(os.path.relpath(path)))
-            return []
-
-        if not isinstance(d, list):
-            msg = "dir cache file format error '{}' [skipping the file]"
-            logger.error(msg.format(os.path.relpath(path)))
-            return []
-
-        for info in d:
-            info["relpath"] = self.to_ospath(info["relpath"])
-
-        self._dir_info[md5] = d
-
-        return d
-
-    def dump_dir_cache(self, md5, dir_info):
-        path = self.get(md5)
-        dname = os.path.dirname(path)
-
-        assert self.is_dir_cache(path)
-        assert isinstance(dir_info, list)
-
-        if not os.path.isdir(dname):
-            os.makedirs(dname)
-
-        # NOTE: Writing first and renaming after that
-        # to make sure that the operation is atomic.
-        tmp = "{}.{}".format(path, str(uuid.uuid4()))
-
-        with open(tmp, "w+") as fd:
-            json.dump(dir_info, fd, sort_keys=True)
-        move(tmp, path)
-
-    @classmethod
-    def is_dir_cache(cls, cache):
-        return cache.endswith(cls.MD5_DIR_SUFFIX)
-
-    def do_checkout(
-        self, path_info, checksum, force=False, progress_callback=None
-    ):
-        path = path_info["path"]
-        cache = self.get(checksum)
-
-        if not self.is_dir_cache(cache):
-            if self.exists(path_info):
-                self.safe_remove(path_info, force=force)
-
-            self.link(cache, path)
-            self.state.update_link(path)
-            if progress_callback:
-                progress_callback.update(os.path.relpath(path))
-            return
-
-        # Create dir separately so that dir is created
-        # even if there are no files in it
-        if not os.path.exists(path):
-            os.makedirs(path)
-
-        dir_relpath = os.path.relpath(path)
-
-        logger.debug("Linking directory '{}'.".format(dir_relpath))
-
-        dir_info = self.load_dir_cache(checksum)
-
-        for processed, entry in enumerate(dir_info):
-            relpath = entry[self.PARAM_RELPATH]
-            m = entry[self.PARAM_CHECKSUM]
-            p = os.path.join(path, relpath)
-            c = self.get(m)
-
-            entry_info = {"scheme": path_info["scheme"], self.PARAM_PATH: p}
-
-            entry_checksum_info = {self.PARAM_CHECKSUM: m}
-
-            if self.changed(entry_info, entry_checksum_info):
-                if self.exists(entry_info):
-                    self.safe_remove(entry_info, force=force)
-
-                self.link(c, p)
-
-            if progress_callback:
-                progress_callback.update(os.path.relpath(p))
-
-        self._discard_working_directory_changes(path, dir_info, force=force)
-
-        self.state.update_link(path)
-
     def already_cached(self, path_info):
         assert path_info["scheme"] in ["", "local"]
 
-        current_md5 = self.state.update(path_info["path"])
+        current_md5 = self.get_checksum(path_info)
 
         if not current_md5:
             return False
 
         return not self.changed_cache(current_md5)
 
-    def _discard_working_directory_changes(self, path, dir_info, force=False):
-        working_dir_files = set(path for path in walk_files(path))
-
-        cached_files = set(
-            os.path.join(path, file["relpath"]) for file in dir_info
-        )
-
-        delta = working_dir_files - cached_files
-
-        for file in delta:
-            self.safe_remove({"scheme": "local", "path": file}, force=force)
-
-    def _move(self, inp, outp):
-        # moving in two stages to make the whole operation atomic in
-        # case inp and outp are in different filesystems and actual
-        # physical copying of data is happening
-        tmp = "{}.{}".format(outp, str(uuid.uuid4()))
-        move(inp, tmp)
-        move(tmp, outp)
-
-    def _save_file(self, path, md5):
-        assert md5 is not None
-
-        cache = self.get(md5)
-
-        if self.changed_cache(md5):
-            self._move(path, cache)
-        else:
-            remove(path)
-
-        self.link(cache, path)
-        self.state.update_link(path)
-
-        # we need to update path and cache, since in case of reflink,
-        # or copy cache type moving original file results in updates on
-        # next executed command, which causes md5 recalculation
-        self.state.update(path, md5)
-        self.state.update(cache, md5)
-
-        return {self.PARAM_CHECKSUM: md5}
-
-    def _save_dir(self, path, md5):
-        dir_info = self.load_dir_cache(md5)
-        dir_relpath = os.path.relpath(path)
-        dir_size = len(dir_info)
-        bar = dir_size > LARGE_DIR_SIZE
-
-        logger.info("Linking directory '{}'.".format(dir_relpath))
-
-        for processed, entry in enumerate(dir_info):
-            relpath = entry[self.PARAM_RELPATH]
-            m = entry[self.PARAM_CHECKSUM]
-            p = os.path.join(path, relpath)
-            c = self.get(m)
-
-            if self.changed_cache(m):
-                self._move(p, c)
-            else:
-                remove(p)
-
-            self.link(c, p)
-
-            self.state.update(p, m)
-            self.state.update(c, m)
-
-            if bar:
-                progress.update_target(dir_relpath, processed, dir_size)
-
-        self.state.update_link(path)
-
-        cache = self.get(md5)
-        self.state.update(cache)
-        self.state.update(path, md5)
+    def is_empty(self, path_info):
+        path = path_info["path"]
 
-        if bar:
-            progress.finish_target(dir_relpath)
+        if self.isfile(path_info) and os.path.getsize(path) == 0:
+            return True
 
-    def save(self, path_info, checksum_info):
-        if path_info["scheme"] != "local":
-            raise NotImplementedError
+        if self.isdir(path_info) and len(os.listdir(path)) == 0:
+            return True
 
-        path = path_info["path"]
+        return False
 
-        msg = "Saving '{}' to cache '{}'."
-        logger.info(
-            msg.format(os.path.relpath(path), os.path.relpath(self.cache_dir))
-        )
+    def isfile(self, path_info):
+        return os.path.isfile(path_info["path"])
 
-        md5 = checksum_info[self.PARAM_CHECKSUM]
-        if os.path.isdir(path):
-            self._save_dir(path, md5)
-        else:
-            self._save_file(path, md5)
+    def isdir(self, path_info):
+        return os.path.isdir(path_info["path"])
 
-    def save_info(self, path_info):
-        if path_info["scheme"] != "local":
-            raise NotImplementedError
+    def walk(self, path_info):
+        return os.walk(path_info["path"])
 
-        return {self.PARAM_CHECKSUM: self.state.update(path_info["path"])}
+    def get_file_checksum(self, path_info):
+        return file_md5(path_info["path"])[0]
 
     def remove(self, path_info):
         if path_info["scheme"] != "local":
@@ -481,13 +215,21 @@ def move(self, from_info, to_info):
         if from_info["scheme"] != "local" or to_info["scheme"] != "local":
             raise NotImplementedError
 
-        move(from_info["path"], to_info["path"])
+        inp = from_info["path"]
+        outp = to_info["path"]
+
+        # moving in two stages to make the whole operation atomic in
+        # case inp and outp are in different filesystems and actual
+        # physical copying of data is happening
+        tmp = "{}.{}".format(outp, str(uuid.uuid4()))
+        move(inp, tmp)
+        move(tmp, outp)
 
     def cache_exists(self, md5s):
         assert isinstance(md5s, list)
         return list(filter(lambda md5: not self.changed_cache_file(md5), md5s))
 
-    def upload(self, from_infos, to_infos, names=None):
+    def upload(self, from_infos, to_infos, names=None, no_progress_bar=False):
         names = self._verify_path_args(to_infos, from_infos, names)
 
         for from_info, to_info, name in zip(from_infos, to_infos, names):
@@ -510,7 +252,12 @@ def upload(self, from_infos, to_infos, names=None):
             tmp_file = tmp_fname(to_info["path"])
 
             try:
-                copyfile(from_info["path"], tmp_file, name=name)
+                copyfile(
+                    from_info["path"],
+                    tmp_file,
+                    name=name,
+                    no_progress_bar=no_progress_bar,
+                )
                 os.rename(tmp_file, to_info["path"])
             except Exception:
                 logger.exception(
@@ -724,17 +471,18 @@ def pull(self, checksum_infos, remote, jobs=None, show_checksums=False):
             download=True,
         )
 
-    def _cache_metadata_changed(self):
+    def _changed_cache_dir(self):
         mtime, size = get_mtime_and_size(self.cache_dir)
         inode = get_inode(self.cache_dir)
 
         existing_record = self.state.get_state_record_for_inode(inode)
-
         if existing_record:
             cached_mtime, cached_size, _, _ = existing_record
-            return not (mtime == cached_mtime and size == cached_size)
+            changed = not (mtime == cached_mtime and size == cached_size)
+        else:
+            changed = True
 
-        return True
+        return changed
 
     def _log_missing_caches(self, checksum_info_dict):
         missing_caches = [
@@ -799,3 +547,7 @@ def unprotect(path_info):
             RemoteLOCAL._unprotect_dir(path)
         else:
             RemoteLOCAL._unprotect_file(path)
+
+    @staticmethod
+    def protect(path_info):
+        os.chmod(path_info["path"], stat.S_IREAD | stat.S_IRGRP | stat.S_IROTH)
diff --git a/dvc/remote/s3.py b/dvc/remote/s3.py
--- a/dvc/remote/s3.py
+++ b/dvc/remote/s3.py
@@ -94,8 +94,12 @@ def get_etag(cls, s3, bucket, path):
 
         return obj["ETag"].strip('"')
 
+    def get_file_checksum(self, path_info):
+        return self.get_etag(self.s3, path_info["bucket"], path_info["path"])
+
     @staticmethod
     def get_head_object(s3, bucket, path, *args, **kwargs):
+
         try:
             obj = s3.head_object(Bucket=bucket, Key=path, *args, **kwargs)
         except Exception as exc:
@@ -104,16 +108,6 @@ def get_head_object(s3, bucket, path, *args, **kwargs):
             )
         return obj
 
-    def save_info(self, path_info):
-        if path_info["scheme"] != "s3":
-            raise NotImplementedError
-
-        return {
-            self.PARAM_CHECKSUM: self.get_etag(
-                self.s3, path_info["bucket"], path_info["path"]
-            )
-        }
-
     @classmethod
     def _copy_multipart(cls, s3, from_info, to_info, size, n_parts):
         mpu = s3.create_multipart_upload(
@@ -241,7 +235,7 @@ def exists(self, path_info):
         paths = self._list_paths(path_info["bucket"], path_info["path"])
         return any(path_info["path"] == path for path in paths)
 
-    def upload(self, from_infos, to_infos, names=None):
+    def upload(self, from_infos, to_infos, names=None, no_progress_bar=False):
         names = self._verify_path_args(to_infos, from_infos, names)
 
         s3 = self.s3
@@ -263,7 +257,7 @@ def upload(self, from_infos, to_infos, names=None):
                 name = os.path.basename(from_info["path"])
 
             total = os.path.getsize(from_info["path"])
-            cb = Callback(name, total)
+            cb = None if no_progress_bar else Callback(name, total)
 
             try:
                 s3.upload_file(
diff --git a/dvc/remote/ssh/__init__.py b/dvc/remote/ssh/__init__.py
--- a/dvc/remote/ssh/__init__.py
+++ b/dvc/remote/ssh/__init__.py
@@ -93,15 +93,19 @@ def exists(self, path_info):
         assert path_info["scheme"] == self.scheme
 
         with self.ssh(**path_info) as ssh:
-            return ssh.file_exists(path_info["path"])
+            return ssh.exists(path_info["path"])
 
-    def md5(self, path_info):
+    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 isdir(self, path_info):
+        with self.ssh(**path_info) as ssh:
+            return ssh.isdir(path_info["path"])
+
     def copy(self, from_info, to_info):
         if (
             from_info["scheme"] != self.scheme
@@ -112,12 +116,6 @@ def copy(self, from_info, to_info):
         with self.ssh(**from_info) as ssh:
             ssh.cp(from_info["path"], to_info["path"])
 
-    def save_info(self, path_info):
-        if path_info["scheme"] != self.scheme:
-            raise NotImplementedError
-
-        return {self.PARAM_CHECKSUM: self.md5(path_info)}
-
     def remove(self, path_info):
         if path_info["scheme"] != self.scheme:
             raise NotImplementedError
@@ -125,6 +123,16 @@ def remove(self, path_info):
         with self.ssh(**path_info) as ssh:
             ssh.remove(path_info["path"])
 
+    def move(self, from_info, to_info):
+        if (
+            from_info["scheme"] != self.scheme
+            or to_info["scheme"] != self.scheme
+        ):
+            raise NotImplementedError
+
+        with self.ssh(**from_info) as ssh:
+            ssh.move(from_info["path"], to_info["path"])
+
     def download(
         self,
         from_infos,
@@ -171,7 +179,7 @@ def download(
 
         ssh.close()
 
-    def upload(self, from_infos, to_infos, names=None):
+    def upload(self, from_infos, to_infos, names=None, no_progress_bar=False):
         names = self._verify_path_args(to_infos, from_infos, names)
 
         with self.ssh(**to_infos[0]) as ssh:
@@ -199,3 +207,12 @@ def upload(self, from_infos, to_infos, names=None):
     def list_cache_paths(self):
         with self.ssh(**self.path_info) as ssh:
             return list(ssh.walk_files(self.prefix))
+
+    def walk(self, path_info):
+        with self.ssh(**path_info) as ssh:
+            for entry in ssh.walk(path_info["path"]):
+                yield entry
+
+    def makedirs(self, path_info):
+        with self.ssh(**path_info) as ssh:
+            ssh.makedirs(path_info["path"])
diff --git a/dvc/remote/ssh/connection.py b/dvc/remote/ssh/connection.py
--- a/dvc/remote/ssh/connection.py
+++ b/dvc/remote/ssh/connection.py
@@ -2,6 +2,7 @@
 import os
 import posixpath
 import logging
+from stat import S_ISDIR
 
 try:
     import paramiko
@@ -72,7 +73,7 @@ def close(self):
 
         self._ssh.close()
 
-    def file_exists(self, path):
+    def exists(self, path):
         self._sftp_connect()
         try:
             return self._sftp.stat(path)
@@ -118,7 +119,7 @@ def makedirs(self, path):
                 "a file with the same name '{}' already exists".format(path)
             )
 
-        head, tail = os.path.split(path)
+        head, tail = posixpath.split(path)
 
         if head:
             self.makedirs(head)
@@ -126,8 +127,11 @@ def makedirs(self, path):
         if tail:
             self._sftp.mkdir(path)
 
-    def walk_files(self, directory):
-        from stat import S_ISLNK, S_ISDIR, S_ISREG
+    def walk(self, directory, topdown=True):
+        # NOTE: original os.walk() implementation [1] with default options was
+        # used as a template.
+        #
+        # [1] https://github.com/python/cpython/blob/master/Lib/os.py
 
         self._sftp_connect()
 
@@ -141,28 +145,61 @@ def walk_files(self, directory):
                 cause=exc,
             )
 
+        dirs = []
+        nondirs = []
         for entry in dir_entries:
-            path = posixpath.join(directory, entry.filename)
+            name = entry.filename
+            if S_ISDIR(entry.st_mode):
+                dirs.append(name)
+            else:
+                nondirs.append(name)
 
-            if S_ISLNK(entry.st_mode):
-                path = self._sftp.readlink(directory)
-                entry = self._sftp.stat(path)
+        if topdown:
+            yield directory, dirs, nondirs
 
-            if S_ISDIR(entry.st_mode):
-                for inner_path in self.walk_files(path):
-                    yield inner_path
+        for dname in dirs:
+            newpath = posixpath.join(directory, dname)
+            for entry in self.walk(newpath, topdown=topdown):
+                yield entry
 
-            elif S_ISREG(entry.st_mode):
-                yield path
+        if not topdown:
+            yield directory, dirs, nondirs
 
-    def remove(self, path):
-        self._sftp_connect()
+    def walk_files(self, directory):
+        for root, dirs, files in self.walk(directory):
+            for fname in files:
+                yield posixpath.join(root, fname)
+
+    def _remove_file(self, path):
         try:
             self._sftp.remove(path)
         except IOError as exc:
             if exc.errno != errno.ENOENT:
                 raise
 
+    def _remove_dir(self, path):
+        for root, dirs, files in self.walk(path, topdown=False):
+            for fname in files:
+                path = posixpath.join(root, fname)
+                self._remove_file(path)
+
+            for dname in dirs:
+                path = posixpath.join(root, dname)
+                self._sftp.rmdir(dname)
+        try:
+            self._sftp.rmdir(path)
+        except IOError as exc:
+            if exc.errno != errno.ENOENT:
+                raise
+
+    def remove(self, path):
+        self._sftp_connect()
+
+        if self.isdir(path):
+            self._remove_dir(path)
+        else:
+            self._remove_file(path)
+
     def download(self, src, dest, no_progress_bar=False, progress_title=None):
         self._sftp_connect()
 
@@ -183,6 +220,11 @@ def download(self, src, dest, no_progress_bar=False, progress_title=None):
 
         os.rename(tmp_file, dest)
 
+    def move(self, src, dst):
+        self.makedirs(posixpath.dirname(dst))
+        self._sftp_connect()
+        self._sftp.rename(src, dst)
+
     def upload(self, src, dest, no_progress_bar=False, progress_title=None):
         self._sftp_connect()
 
diff --git a/dvc/repo/__init__.py b/dvc/repo/__init__.py
--- a/dvc/repo/__init__.py
+++ b/dvc/repo/__init__.py
@@ -205,7 +205,7 @@ def _collect_dir_cache(
             else:
                 return ret
 
-        for i in self.cache.local.load_dir_cache(md5):
+        for i in out.dir_cache:
             i["branch"] = branch
             i[r.PARAM_PATH] = os.path.join(
                 info[r.PARAM_PATH], i[r.PARAM_RELPATH]
@@ -234,9 +234,7 @@ def _collect_used_cache(
         if out.scheme != "local":
             return ret
 
-        md5 = info[out.remote.PARAM_CHECKSUM]
-        cache = self.cache.local.get(md5)
-        if not out.remote.is_dir_cache(cache):
+        if not out.is_dir_checksum:
             return ret
 
         return self._collect_dir_cache(
diff --git a/dvc/repo/diff.py b/dvc/repo/diff.py
--- a/dvc/repo/diff.py
+++ b/dvc/repo/diff.py
@@ -133,9 +133,9 @@ def _check_local_cache(a_out, is_checked):
 
 def _is_dir(path, a_outs, b_outs):
     if a_outs.get(path):
-        return a_outs[path].is_dir_cache
+        return a_outs[path].is_dir_checksum
     else:
-        return b_outs[path].is_dir_cache
+        return b_outs[path].is_dir_checksum
 
 
 def _get_diff_outs(self, diff_dct):
diff --git a/dvc/repo/move.py b/dvc/repo/move.py
--- a/dvc/repo/move.py
+++ b/dvc/repo/move.py
@@ -5,7 +5,7 @@
 
 
 def _expand_target_path(from_path, to_path):
-    if os.path.isdir(to_path) and not os.path.isdir(from_path):
+    if os.path.isdir(to_path):
         return os.path.join(to_path, os.path.basename(from_path))
     return to_path
 
@@ -59,7 +59,7 @@ def move(self, from_path, to_path):
         )
 
     to_out = Output.loads_from(
-        stage, [os.path.basename(to_path)], out.cache, out.metric
+        stage, [os.path.basename(to_path)], out.use_cache, out.metric
     )[0]
 
     with self.state:
diff --git a/dvc/stage.py b/dvc/stage.py
--- a/dvc/stage.py
+++ b/dvc/stage.py
@@ -601,14 +601,14 @@ def load(repo, fname):
         return stage
 
     def dumpd(self):
-        from dvc.remote.local import RemoteLOCAL
+        from dvc.remote.base import RemoteBase
 
         return {
             key: value
             for key, value in {
                 Stage.PARAM_MD5: self.md5,
                 Stage.PARAM_CMD: self.cmd,
-                Stage.PARAM_WDIR: RemoteLOCAL.unixpath(
+                Stage.PARAM_WDIR: RemoteBase.to_posixpath(
                     os.path.relpath(self.wdir, os.path.dirname(self.path))
                 ),
                 Stage.PARAM_LOCKED: self.locked,
diff --git a/dvc/state.py b/dvc/state.py
--- a/dvc/state.py
+++ b/dvc/state.py
@@ -7,7 +7,7 @@
 import logging
 
 from dvc.config import Config
-from dvc.utils import file_md5, remove, current_timestamp
+from dvc.utils import remove, current_timestamp
 from dvc.exceptions import DvcException
 from dvc.utils.fs import get_mtime_and_size, get_inode
 
@@ -29,11 +29,18 @@ def __init__(self, dvc_version, expected, actual):
         )
 
 
-def _file_metadata_changed(actual_mtime, mtime, actual_size, size):
-    return actual_mtime != mtime or actual_size != size
+class StateBase(object):
+    def save(self, path_info, checksum):
+        pass
 
+    def get(self, path_info):
+        return None
+
+    def save_link(self, path_info):
+        pass
 
-class State(object):  # pylint: disable=too-many-instance-attributes
+
+class State(StateBase):  # pylint: disable=too-many-instance-attributes
     """Class for the state database.
 
     Args:
@@ -112,11 +119,6 @@ def __enter__(self):
     def __exit__(self, typ, value, tbck):
         self.dump()
 
-    def _collect(self, path):
-        if os.path.isdir(path):
-            return self.repo.cache.local.collect_dir_cache(path)
-        return (file_md5(path)[0], None)
-
     def changed(self, path, md5):
         """Check if file/directory has the expected md5.
 
@@ -299,53 +301,9 @@ def dump(self):
         self.cursor = None
         self.inserts = 0
 
-    def _do_update(self, path, known_checksum=None):
-        """
-        Make sure the stored info for the given path is up to date.
-        """
-        if not os.path.exists(path):
-            return None, None
-
-        actual_mtime, actual_size = get_mtime_and_size(path)
-        actual_inode = get_inode(path)
-
-        existing_record = self.get_state_record_for_inode(actual_inode)
-
-        if existing_record:
-            md5, info = self._update_existing_state_record(
-                path,
-                actual_inode,
-                actual_mtime,
-                actual_size,
-                existing_record,
-                known_checksum,
-            )
-        else:
-            md5, info = self._insert_new_state_record(
-                path, actual_inode, actual_mtime, actual_size, known_checksum
-            )
-
-        return md5, info
-
-    def _update_existing_state_record(
-        self,
-        path,
-        actual_inode,
-        actual_mtime,
-        actual_size,
-        existing_record,
-        known_checksum=None,
-    ):
-
-        mtime, size, md5, _ = existing_record
-        if _file_metadata_changed(actual_mtime, mtime, actual_size, size):
-            md5, info = self._update_state_for_path_changed(
-                path, actual_inode, actual_mtime, actual_size, known_checksum
-            )
-        else:
-            info = None
-            self._update_state_record_timestamp_for_inode(actual_inode)
-        return md5, info
+    @staticmethod
+    def _file_metadata_changed(actual_mtime, mtime, actual_size, size):
+        return actual_mtime != mtime or actual_size != size
 
     def _update_state_record_timestamp_for_inode(self, actual_inode):
         cmd = 'UPDATE {} SET timestamp = "{}" WHERE inode = {}'
@@ -358,17 +316,8 @@ def _update_state_record_timestamp_for_inode(self, actual_inode):
         )
 
     def _update_state_for_path_changed(
-        self,
-        path,
-        actual_inode,
-        actual_mtime,
-        actual_size,
-        known_checksum=None,
+        self, path, actual_inode, actual_mtime, actual_size, checksum
     ):
-        if known_checksum:
-            md5, info = known_checksum, None
-        else:
-            md5, info = self._collect(path)
         cmd = (
             "UPDATE {} SET "
             'mtime = "{}", size = "{}", '
@@ -380,20 +329,17 @@ def _update_state_for_path_changed(
                 self.STATE_TABLE,
                 actual_mtime,
                 actual_size,
-                md5,
+                checksum,
                 current_timestamp(),
                 self._to_sqlite(actual_inode),
             )
         )
-        return md5, info
 
     def _insert_new_state_record(
-        self, path, actual_inode, actual_mtime, actual_size, known_checksum
+        self, path, actual_inode, actual_mtime, actual_size, checksum
     ):
-        if known_checksum:
-            md5, info = known_checksum, None
-        else:
-            md5, info = self._collect(path)
+        assert checksum is not None
+
         cmd = (
             "INSERT INTO {}(inode, mtime, size, md5, timestamp) "
             'VALUES ({}, "{}", "{}", "{}", "{}")'
@@ -404,12 +350,11 @@ def _insert_new_state_record(
                 self._to_sqlite(actual_inode),
                 actual_mtime,
                 actual_size,
-                md5,
+                checksum,
                 current_timestamp(),
             )
         )
         self.inserts += 1
-        return md5, info
 
     def get_state_record_for_inode(self, inode):
         cmd = "SELECT mtime, size, md5, timestamp from {} " "WHERE inode={}"
@@ -422,43 +367,74 @@ def get_state_record_for_inode(self, inode):
             return results[0]
         return None
 
-    def update(self, path, known_checksum=None):
-        """Gets the checksum for the specified path. Checksum will be
-        retrieved from the state database if available, otherwise it will be
-        computed and cached in the state database for the further use.
+    def save(self, path_info, checksum):
+        """Save checksum for the specified path info.
 
         Args:
-            path (str): path to get the checksum for.
-
-        Returns:
-            str: checksum for the specified path.
+            path_info (dict): path_info to save checksum for.
+            checksum (str): checksum to save.
         """
-        return self._do_update(path, known_checksum)[0]
+        assert path_info["scheme"] == "local"
+        assert checksum is not None
+
+        path = path_info["path"]
+        assert os.path.exists(path)
+
+        actual_mtime, actual_size = get_mtime_and_size(path)
+        actual_inode = get_inode(path)
+
+        existing_record = self.get_state_record_for_inode(actual_inode)
+        if not existing_record:
+            self._insert_new_state_record(
+                path, actual_inode, actual_mtime, actual_size, checksum
+            )
+            return
 
-    def update_info(self, path):
-        """Gets the checksum and the directory info (if applicable) for the
-        specified path.
+        self._update_state_for_path_changed(
+            path, actual_inode, actual_mtime, actual_size, checksum
+        )
+
+    def get(self, path_info):
+        """Gets the checksum for the specified path info. Checksum will be
+        retrieved from the state database if available.
 
         Args:
-            path (str): path to get the checksum and the directory info for.
+            path_info (dict): path info to get the checksum for.
 
         Returns:
-            tuple: checksum for the specified path along with a directory info
-            (list of {relative_path: checksum} entries for each file in the
-            directory) if applicable, otherwise None.
+            str or None: checksum for the specified path info or None if it
+            doesn't exist in the state database.
         """
-        md5, info = self._do_update(path)
-        if not info:
-            info = self.repo.cache.local.load_dir_cache(md5)
-        return (md5, info)
+        assert path_info["scheme"] == "local"
+        path = path_info["path"]
+
+        if not os.path.exists(path):
+            return None
+
+        actual_mtime, actual_size = get_mtime_and_size(path)
+        actual_inode = get_inode(path)
+
+        existing_record = self.get_state_record_for_inode(actual_inode)
+        if not existing_record:
+            return None
 
-    def update_link(self, path):
+        mtime, size, checksum, _ = existing_record
+        if self._file_metadata_changed(actual_mtime, mtime, actual_size, size):
+            return None
+
+        self._update_state_record_timestamp_for_inode(actual_inode)
+        return checksum
+
+    def save_link(self, path_info):
         """Adds the specified path to the list of links created by dvc. This
         list is later used on `dvc checkout` to cleanup old links.
 
         Args:
-            path (str): path to add to the list of links.
+            path_info (dict): path info to add to the list of links.
         """
+        assert path_info["scheme"] == "local"
+        path = path_info["path"]
+
         if not os.path.exists(path):
             return
 
diff --git a/dvc/utils/compat.py b/dvc/utils/compat.py
--- a/dvc/utils/compat.py
+++ b/dvc/utils/compat.py
@@ -5,7 +5,6 @@
 import os
 import errno
 
-
 # Syntax sugar.
 _ver = sys.version_info
 


Test patch:
diff --git a/tests/func/test_add.py b/tests/func/test_add.py
--- a/tests/func/test_add.py
+++ b/tests/func/test_add.py
@@ -224,9 +224,8 @@ def test_dir(self):
 
 class TestShouldUpdateStateEntryForFileAfterAdd(TestDvc):
     def test(self):
-        file_md5_counter = spy(dvc.state.file_md5)
-        with patch.object(dvc.state, "file_md5", file_md5_counter):
-
+        file_md5_counter = spy(dvc.remote.local.file_md5)
+        with patch.object(dvc.remote.local, "file_md5", file_md5_counter):
             ret = main(["config", "cache.type", "copy"])
             self.assertEqual(ret, 0)
 
@@ -245,8 +244,8 @@ def test(self):
 
 class TestShouldUpdateStateEntryForDirectoryAfterAdd(TestDvc):
     def test(self):
-        file_md5_counter = spy(dvc.state.file_md5)
-        with patch.object(dvc.state, "file_md5", file_md5_counter):
+        file_md5_counter = spy(dvc.remote.local.file_md5)
+        with patch.object(dvc.remote.local, "file_md5", file_md5_counter):
 
             ret = main(["config", "cache.type", "copy"])
             self.assertEqual(ret, 0)
@@ -303,27 +302,14 @@ def test(self):
 
 
 class TestShouldCollectDirCacheOnlyOnce(TestDvc):
-    NEW_LARGE_DIR_SIZE = 1
-
-    @patch("dvc.remote.local.LARGE_DIR_SIZE", NEW_LARGE_DIR_SIZE)
     def test(self):
         from dvc.remote.local import RemoteLOCAL
 
-        collect_dir_counter = spy(RemoteLOCAL.collect_dir_cache)
+        get_dir_checksum_counter = spy(RemoteLOCAL.get_dir_checksum)
         with patch.object(
-            RemoteLOCAL, "collect_dir_cache", collect_dir_counter
+            RemoteLOCAL, "get_dir_checksum", get_dir_checksum_counter
         ):
-
-            LARGE_DIR_FILES_NUM = self.NEW_LARGE_DIR_SIZE + 1
-            data_dir = "dir"
-
-            os.makedirs(data_dir)
-
-            for i in range(LARGE_DIR_FILES_NUM):
-                with open(os.path.join(data_dir, str(i)), "w+") as f:
-                    f.write(str(i))
-
-            ret = main(["add", data_dir])
+            ret = main(["add", self.DATA_DIR])
             self.assertEqual(0, ret)
 
             ret = main(["status"])
@@ -331,7 +317,7 @@ def test(self):
 
             ret = main(["status"])
             self.assertEqual(0, ret)
-        self.assertEqual(1, collect_dir_counter.mock.call_count)
+        self.assertEqual(1, get_dir_checksum_counter.mock.call_count)
 
 
 class SymlinkAddTestBase(TestDvc):
diff --git a/tests/func/test_checkout.py b/tests/func/test_checkout.py
--- a/tests/func/test_checkout.py
+++ b/tests/func/test_checkout.py
@@ -69,7 +69,7 @@ def test(self):
 
 class TestCheckoutCorruptedCacheFile(TestRepro):
     def test(self):
-        cache = self.foo_stage.outs[0].cache
+        cache = self.foo_stage.outs[0].cache_path
 
         with open(cache, "a") as fd:
             fd.write("1")
diff --git a/tests/func/test_data_cloud.py b/tests/func/test_data_cloud.py
--- a/tests/func/test_data_cloud.py
+++ b/tests/func/test_data_cloud.py
@@ -9,7 +9,6 @@
 import copy
 import logging
 
-from dvc.state import State
 from mock import patch
 
 from dvc.utils.compat import str
@@ -244,7 +243,7 @@ def _test_cloud(self):
         self.assertEqual(len(stages), 1)
         stage = stages[0]
         self.assertTrue(stage is not None)
-        cache = stage.outs[0].cache
+        cache = stage.outs[0].cache_path
         info = stage.outs[0].dumpd()
         md5 = info["md5"]
 
@@ -253,7 +252,7 @@ def _test_cloud(self):
         stage_dir = stages[0]
         self.assertTrue(stage_dir is not None)
 
-        cache_dir = stage_dir.outs[0].cache
+        cache_dir = stage_dir.outs[0].cache_path
         info_dir = stage_dir.outs[0].dumpd()
         md5_dir = info_dir["md5"]
 
@@ -423,13 +422,13 @@ def _test_cloud(self, remote=None):
         self.assertEqual(len(stages), 1)
         stage = stages[0]
         self.assertTrue(stage is not None)
-        cache = stage.outs[0].cache
+        cache = stage.outs[0].cache_path
 
         stages = self.dvc.add(self.DATA_DIR)
         self.assertEqual(len(stages), 1)
         stage_dir = stages[0]
         self.assertTrue(stage_dir is not None)
-        cache_dir = stage_dir.outs[0].cache
+        cache_dir = stage_dir.outs[0].cache_path
 
         # FIXME check status output
         self.main(["status", "-c", "--show-checksums"] + args)
@@ -730,8 +729,10 @@ def test(self):
 
 class TestCheckSumRecalculation(TestDvc):
     def test(self):
-        test_collect = spy(State._collect)
-        with patch.object(State, "_collect", test_collect):
+        test_get_file_checksum = spy(RemoteLOCAL.get_file_checksum)
+        with patch.object(
+            RemoteLOCAL, "get_file_checksum", test_get_file_checksum
+        ):
             url = get_local_url()
             ret = main(["remote", "add", "-d", TEST_REMOTE, url])
             self.assertEqual(ret, 0)
@@ -743,7 +744,7 @@ def test(self):
             self.assertEqual(ret, 0)
             ret = main(["run", "-d", self.FOO, "echo foo"])
             self.assertEqual(ret, 0)
-        self.assertEqual(test_collect.mock.call_count, 1)
+        self.assertEqual(test_get_file_checksum.mock.call_count, 1)
 
 
 class TestShouldWarnOnNoChecksumInLocalAndRemoteCache(TestDvc):
diff --git a/tests/func/test_dependency.py b/tests/func/test_dependency.py
--- a/tests/func/test_dependency.py
+++ b/tests/func/test_dependency.py
@@ -1,15 +1,21 @@
+import copy
+import pytest
+
 from dvc.stage import Stage
 from dvc.dependency import _get, DEP_MAP
 
-from tests.func.test_output import TestOutScheme
+from tests.func.test_output import TESTS as OUT_TESTS
+
+
+TESTS = copy.copy(OUT_TESTS)
+TESTS.append(("http://example.com/path/to/file", "http"))
+TESTS.append(("https://example.com/path/to/file", "https"))
 
 
-class TestDepScheme(TestOutScheme):
-    MAP = DEP_MAP
+def _get_dep(dvc, path):
+    return _get(Stage(dvc), path, None)
 
-    TESTS = TestOutScheme.TESTS.copy()
-    TESTS["http://example.com/path/to/file"] = "http"
-    TESTS["https://example.com/path/to/file"] = "https"
 
-    def _get(self, path):
-        return _get(Stage(self.dvc), path, None)
+@pytest.mark.parametrize("url,scheme", TESTS)
+def test_scheme(dvc, url, scheme):
+    assert type(_get_dep(dvc, url)) == DEP_MAP[scheme]
diff --git a/tests/func/test_metrics.py b/tests/func/test_metrics.py
--- a/tests/func/test_metrics.py
+++ b/tests/func/test_metrics.py
@@ -582,6 +582,7 @@ class TestCachedMetrics(TestDvc):
     def _do_add(self, branch):
         self.dvc.scm.checkout(branch)
         self.dvc.checkout(force=True)
+        assert not os.path.exists("metrics.json")
 
         with open("metrics.json", "w+") as fd:
             json.dump({"metrics": branch}, fd)
diff --git a/tests/func/test_move.py b/tests/func/test_move.py
--- a/tests/func/test_move.py
+++ b/tests/func/test_move.py
@@ -124,23 +124,31 @@ def test(self):
 
 class TestMoveDirectoryShouldNotOverwriteExisting(TestDvc):
     def test(self):
-        new_dir_name = "data_dir2"
-        os.mkdir(new_dir_name)
+        dir_name = "dir"
+        orig_listdir = set(os.listdir(self.DATA_DIR))
+
         ret = main(["add", self.DATA_DIR])
         self.assertEqual(ret, 0)
 
-        ret = main(["move", self.DATA_DIR, new_dir_name])
+        os.mkdir(dir_name)
+        new_dir_name = os.path.join(dir_name, self.DATA_DIR)
+
+        ret = main(["move", self.DATA_DIR, dir_name])
         self.assertEqual(ret, 0)
 
+        data_dir_stage = self.DATA_DIR + Stage.STAGE_FILE_SUFFIX
         self.assertFalse(os.path.exists(self.DATA_DIR))
-        self.assertFalse(
-            os.path.exists(self.DATA_DIR + Stage.STAGE_FILE_SUFFIX)
+        self.assertFalse(os.path.exists(data_dir_stage))
+
+        self.assertTrue(os.path.exists(dir_name))
+        self.assertEqual(
+            set(os.listdir(dir_name)),
+            set([".gitignore", data_dir_stage, self.DATA_DIR]),
         )
 
         self.assertTrue(os.path.exists(new_dir_name))
         self.assertTrue(os.path.isfile(new_dir_name + Stage.STAGE_FILE_SUFFIX))
-
-        self.assertEqual(os.listdir(new_dir_name), [self.DATA_DIR])
+        self.assertEqual(set(os.listdir(new_dir_name)), orig_listdir)
 
 
 class TestMoveFileBetweenDirectories(TestDvc):
diff --git a/tests/func/test_output.py b/tests/func/test_output.py
--- a/tests/func/test_output.py
+++ b/tests/func/test_output.py
@@ -1,30 +1,29 @@
+import pytest
+
 from dvc.stage import Stage
 from dvc.output import _get, OUTS_MAP
 
-from tests.basic_env import TestDvc
 
+TESTS = [
+    ("s3://bucket/path", "s3"),
+    ("gs://bucket/path", "gs"),
+    ("ssh://example.com:/dir/path", "ssh"),
+    ("hdfs://example.com/dir/path", "hdfs"),
+    ("path/to/file", "local"),
+    ("path\\to\\file", "local"),
+    ("file", "local"),
+    ("./file", "local"),
+    (".\\file", "local"),
+    ("../file", "local"),
+    ("..\\file", "local"),
+    ("unknown://path", "local"),
+]
 
-class TestOutScheme(TestDvc):
-    MAP = OUTS_MAP
 
-    TESTS = {
-        "s3://bucket/path": "s3",
-        "gs://bucket/path": "gs",
-        "ssh://example.com:/dir/path": "ssh",
-        "hdfs://example.com/dir/path": "hdfs",
-        "path/to/file": "local",
-        "path\\to\\file": "local",
-        "file": "local",
-        "./file": "local",
-        ".\\file": "local",
-        "../file": "local",
-        "..\\file": "local",
-        "unknown://path": "local",
-    }
+def _get_out(dvc, path):
+    return _get(Stage(dvc), path, None, None, None, None)
 
-    def _get(self, path):
-        return _get(Stage(self.dvc), path, None, None, None, None)
 
-    def test(self):
-        for path, cls in self.TESTS.items():
-            self.assertIsInstance(self._get(path), self.MAP[cls])
+@pytest.mark.parametrize("url,scheme", TESTS)
+def test_scheme(dvc, url, scheme):
+    assert type(_get_out(dvc, url)) == OUTS_MAP[scheme]
diff --git a/tests/func/test_state.py b/tests/func/test_state.py
--- a/tests/func/test_state.py
+++ b/tests/func/test_state.py
@@ -14,20 +14,27 @@
 class TestState(TestDvc):
     def test_update(self):
         path = os.path.join(self.dvc.root_dir, self.FOO)
+        path_info = {"scheme": "local", "path": path}
         md5 = file_md5(path)[0]
 
         state = State(self.dvc, self.dvc.config.config)
 
         with state:
-            entry_md5 = state.update(path)
+            state.save(path_info, md5)
+            entry_md5 = state.get(path_info)
             self.assertEqual(entry_md5, md5)
 
             os.unlink(path)
             with open(path, "a") as fd:
                 fd.write("1")
 
+            entry_md5 = state.get(path_info)
+            self.assertTrue(entry_md5 is None)
+
             md5 = file_md5(path)[0]
-            entry_md5 = state.update(path)
+            state.save(path_info, md5)
+
+            entry_md5 = state.get(path_info)
             self.assertEqual(entry_md5, md5)
 
 
@@ -66,9 +73,10 @@ def test_transforms_inode(self, get_inode_mock):
         self.assertNotEqual(inode, state._to_sqlite(inode))
 
         path = os.path.join(self.dvc.root_dir, self.FOO)
+        md5 = file_md5(path)[0]
         get_inode_mock.side_effect = self.mock_get_inode(path, inode)
 
         with state:
-            state.update(path)
+            state.save({"scheme": "local", "path": path}, md5)
             ret = state.get_state_record_for_inode(inode)
             self.assertIsNotNone(ret)
diff --git a/tests/unit/output/test_local.py b/tests/unit/output/test_local.py
--- a/tests/unit/output/test_local.py
+++ b/tests/unit/output/test_local.py
@@ -26,22 +26,20 @@ def _get_output(self):
         stage = Stage(self.dvc)
         return OutputLOCAL(stage, "path")
 
-    @patch.object(OutputLOCAL, "cache", None)
     def test_return_0_on_no_cache(self):
         o = self._get_output()
-
+        o.use_cache = False
         self.assertEqual(0, o.get_files_number())
 
-    @patch.object(OutputLOCAL, "cache", "cache")
-    @patch.object(OutputLOCAL, "is_dir_cache", True)
+    @patch.object(OutputLOCAL, "checksum", "12345678.dir")
     @patch.object(OutputLOCAL, "dir_cache", [{"md5": "asdf"}, {"md5": "qwe"}])
     def test_return_mutiple_for_dir(self):
         o = self._get_output()
 
         self.assertEqual(2, o.get_files_number())
 
-    @patch.object(OutputLOCAL, "cache", "cache")
-    @patch.object(OutputLOCAL, "is_dir_cache", False)
+    @patch.object(OutputLOCAL, "checksum", "12345678")
+    @patch.object(OutputLOCAL, "is_dir_checksum", False)
     def test_return_1_on_single_file_cache(self):
         o = self._get_output()
 
diff --git a/tests/unit/remote/ssh/test_connection.py b/tests/unit/remote/ssh/test_connection.py
--- a/tests/unit/remote/ssh/test_connection.py
+++ b/tests/unit/remote/ssh/test_connection.py
@@ -1,6 +1,9 @@
+from __future__ import unicode_literals
+
 import mockssh
 import os
-from pytest import yield_fixture
+import posixpath
+import pytest
 
 from dvc.remote.ssh.connection import SSHConnection
 
@@ -10,35 +13,74 @@
 key_path = os.path.join(here, "{0}.key".format(user))
 
 
-@yield_fixture()
-def server():
+@pytest.yield_fixture()
+def ssh():
     users = {user: key_path}
     with mockssh.Server(users) as s:
-        yield s
+        yield SSHConnection(
+            s.host, username=user, port=s.port, key_filename=key_path
+        )
+
+
+def test_connection(ssh):
+    assert ssh.execute("ls /")
+
+
+def test_isdir(ssh):
+    assert ssh.isdir(here)
+    assert not ssh.isdir(__file__)
+
+
+def test_exists(ssh):
+    assert not ssh.exists("/path/to/non/existent/file")
+    assert ssh.exists(__file__)
+
+
+def test_isfile(ssh):
+    assert ssh.isfile(__file__)
+    assert not ssh.isfile(here)
+
+
+def test_makedirs(tmp_path, ssh):
+    tmp = tmp_path.absolute().as_posix()
+    path = posixpath.join(tmp, "dir", "subdir")
+    ssh.makedirs(path)
+    assert os.path.isdir(path)
+
+
+def test_walk(tmp_path, ssh):
+    root_path = tmp_path
+    dir_path = root_path / "dir"
+    subdir_path = dir_path / "subdir"
 
+    dir_path.mkdir()
+    subdir_path.mkdir()
 
-def _client(server):
-    client = SSHConnection(
-        server.host, username=user, port=server.port, key_filename=key_path
-    )
-    return client
+    root_data_path = root_path / "root_data"
+    dir_data_path = dir_path / "dir_data"
+    subdir_data_path = subdir_path / "subdir_data"
 
+    with root_data_path.open("w+") as fobj:
+        fobj.write("")
 
-def test_connection(server):
-    client = _client(server)
-    ls = client.execute("ls /")
-    assert ls
+    with dir_data_path.open("w+") as fobj:
+        fobj.write("")
 
+    with subdir_data_path.open("w+") as fobj:
+        fobj.write("")
 
-def test_isdir(server):
-    client = _client(server)
-    path = here
-    isdir = client.isdir(path=path)
-    assert isdir is True
+    entries = [
+        dir_path,
+        subdir_path,
+        root_data_path,
+        dir_data_path,
+        subdir_data_path,
+    ]
+    expected = set([entry.absolute().as_posix() for entry in entries])
 
+    paths = set()
+    for root, dirs, files in ssh.walk(root_path.absolute().as_posix()):
+        for entry in dirs + files:
+            paths.add(posixpath.join(root, entry))
 
-def test_file_exists(server):
-    client = _client(server)
-    path = "/path/to/file"
-    file_path = client.file_exists(path=path)
-    assert file_path is False
+    assert paths == expected
diff --git a/tests/unit/remote/test_base.py b/tests/unit/remote/test_base.py
--- a/tests/unit/remote/test_base.py
+++ b/tests/unit/remote/test_base.py
@@ -1,61 +1,68 @@
 import mock
 from unittest import TestCase
 
-from dvc.remote import REMOTES, RemoteLOCAL, RemoteSSH, RemoteHDFS
 from dvc.remote.base import RemoteBase, RemoteCmdError, RemoteMissingDepsError
 
 
-class TestMissingDeps(TestCase):
+class TestRemoteBase(object):
+    REMOTE_CLS = RemoteBase
+
+
+class TestMissingDeps(TestCase, TestRemoteBase):
     def test(self):
-        remotes = REMOTES + [RemoteLOCAL, RemoteBase]
-        for remote_class in remotes:
-            REQUIRES = {"foo": None, "bar": None, "mock": mock}
-            with mock.patch.object(remote_class, "REQUIRES", REQUIRES):
-                with self.assertRaises(RemoteMissingDepsError):
-                    remote_class(None, {})
+        REQUIRES = {"foo": None, "bar": None, "mock": mock}
+        with mock.patch.object(self.REMOTE_CLS, "REQUIRES", REQUIRES):
+            with self.assertRaises(RemoteMissingDepsError):
+                self.REMOTE_CLS(None, {})
 
 
-class TestCmdError(TestCase):
+class TestCmdError(TestCase, TestRemoteBase):
     def test(self):
-        for remote_class in [RemoteSSH, RemoteHDFS]:
-            repo = None
-            config = {}
+        repo = None
+        config = {}
 
-            remote_name = remote_class.scheme
-            cmd = "sed 'hello'"
-            ret = "1"
-            err = "sed: expression #1, char 2: extra characters after command"
+        cmd = "sed 'hello'"
+        ret = "1"
+        err = "sed: expression #1, char 2: extra characters after command"
 
-            with mock.patch.object(
-                remote_class,
-                "remove",
-                side_effect=RemoteCmdError(remote_name, cmd, ret, err),
-            ):
-                with self.assertRaises(RemoteCmdError):
-                    remote_class(repo, config).remove("file")
+        with mock.patch.object(
+            self.REMOTE_CLS,
+            "remove",
+            side_effect=RemoteCmdError("base", cmd, ret, err),
+        ):
+            with self.assertRaises(RemoteCmdError):
+                self.REMOTE_CLS(repo, config).remove("file")
 
 
 class TestCacheExists(TestCase):
     def test(self):
-        for remote_class in REMOTES:
-            config = {
-                "url": remote_class.scheme + "://example/prefix",
-                "connection_string": "1234567",
-            }
-            remote = remote_class(None, config)
-
-            with mock.patch.object(remote, "changed_cache", return_value=True):
-                with mock.patch.object(remote, "copy") as cp:
-                    remote.save(
-                        remote.path_info, {remote.PARAM_CHECKSUM: "1234567890"}
-                    )
-                    cp.assert_called_once()
-
-            with mock.patch.object(
-                remote, "changed_cache", return_value=False
-            ):
-                with mock.patch.object(remote, "copy") as cp:
-                    remote.save(
-                        remote.path_info, {remote.PARAM_CHECKSUM: "1234567890"}
-                    )
-                    cp.assert_not_called()
+        config = {
+            "url": "base://example/prefix",
+            "connection_string": "1234567",
+        }
+        remote = RemoteBase(None, config)
+
+        remote.PARAM_CHECKSUM = "checksum"
+        remote.path_info = {}
+        remote.url = ""
+        remote.prefix = ""
+        path_info = {"scheme": None, "path": "example"}
+        checksum_info = {remote.PARAM_CHECKSUM: "1234567890"}
+
+        with mock.patch.object(remote, "_checkout") as mock_checkout:
+            with mock.patch.object(remote, "_save") as mock_save:
+                with mock.patch.object(
+                    remote, "changed_cache", return_value=True
+                ):
+                    remote.save(path_info, checksum_info)
+                    mock_save.assert_called_once()
+                    mock_checkout.assert_not_called()
+
+        with mock.patch.object(remote, "_checkout") as mock_checkout:
+            with mock.patch.object(remote, "_save") as mock_save:
+                with mock.patch.object(
+                    remote, "changed_cache", return_value=False
+                ):
+                    remote.save(path_info, checksum_info)
+                    mock_save.assert_not_called()
+                    mock_checkout.assert_called_once()