Repository: iterative/dvc

Issue:
`dvc move` fails to move datafile to a directory, when the tracking file has a different name than data file
I was playing with `dvc move` and I got this problem.

Script to reproduce it:
```bash
#!/bin/bash -x

rm -rf test1
mkdir test1
cd test1/

echo 'test 1' > file1.txt
dvc init --no-scm
dvc add file1.txt
mv file1.txt.dvc file1.dvc
dvc status
tree -a

mkdir dir1
dvc move file1.txt dir1
dvc status
tree -a
# file1.txt has not been moved to dir1/
```

I am not sure if this is a glitch or everything is as expected, since the `.dvc` file has a different name from the datafile. When tracking file and data file have the same name everything works fine.

Discussion:

Patch:
diff --git a/dvc/command/run.py b/dvc/command/run.py
index b0761435f..1c3c3f457 100644
--- a/dvc/command/run.py
+++ b/dvc/command/run.py
@@ -51,7 +51,6 @@ class CmdRun(CmdBase):
                 no_commit=self.args.no_commit,
                 outs_persist=self.args.outs_persist,
                 outs_persist_no_cache=self.args.outs_persist_no_cache,
-                always_changed=self.args.always_changed,
             )
         except DvcException:
             logger.exception("failed to run command")
@@ -190,12 +189,6 @@ def add_parser(subparsers, parent_parser):
         help="Declare output file or directory that will not be "
         "removed upon repro (do not put into DVC cache).",
     )
-    run_parser.add_argument(
-        "--always-changed",
-        action="store_true",
-        default=False,
-        help="Always consider this DVC-file as changed.",
-    )
     run_parser.add_argument(
         "command", nargs=argparse.REMAINDER, help="Command to execute."
     )
diff --git a/dvc/command/version.py b/dvc/command/version.py
index 3e2b780f0..ad041b4dd 100644
--- a/dvc/command/version.py
+++ b/dvc/command/version.py
@@ -5,7 +5,6 @@ import platform
 import argparse
 import logging
 import uuid
-import itertools
 
 try:
     import psutil
@@ -70,12 +69,9 @@ class CmdVersion(CmdBaseNoRepo):
     def get_fs_type(path):
         partition = {
             pathlib.Path(part.mountpoint): (part.fstype, part.device)
-            for part in psutil.disk_partitions(all=True)
+            for part in psutil.disk_partitions()
         }
-
-        path = pathlib.Path(path)
-
-        for parent in itertools.chain([path], path.parents):
+        for parent in pathlib.Path(path).parents:
             if parent in partition:
                 return partition[parent]
         return ("unkown", "none")
diff --git a/dvc/remote/base.py b/dvc/remote/base.py
index 73d0710ac..207b94703 100644
--- a/dvc/remote/base.py
+++ b/dvc/remote/base.py
@@ -12,8 +12,6 @@ from operator import itemgetter
 from multiprocessing import cpu_count
 from functools import partial
 from concurrent.futures import ThreadPoolExecutor
-from copy import copy
-from dvc.remote.slow_link_detection import slow_link_guard
 
 import dvc.prompt as prompt
 from dvc.config import Config
@@ -81,7 +79,6 @@ class RemoteBASE(object):
     PARAM_RELPATH = "relpath"
     CHECKSUM_DIR_SUFFIX = ".dir"
     CHECKSUM_JOBS = max(1, min(4, cpu_count() // 2))
-    DEFAULT_CACHE_TYPES = ["copy"]
 
     state = StateNoop()
 
@@ -122,15 +119,6 @@ class RemoteBASE(object):
         self.no_traverse = config.get(Config.SECTION_REMOTE_NO_TRAVERSE, True)
         self._dir_info = {}
 
-        types = config.get(Config.SECTION_CACHE_TYPE, None)
-        if types:
-            if isinstance(types, str):
-                types = [t.strip() for t in types.split(",")]
-            self.cache_types = types
-        else:
-            self.cache_types = copy(self.DEFAULT_CACHE_TYPES)
-        self.cache_type_confirmed = False
-
     def __repr__(self):
         return "{class_name}: '{path_info}'".format(
             class_name=type(self).__name__,
@@ -364,47 +352,7 @@ class RemoteBASE(object):
         return False
 
     def link(self, from_info, to_info):
-        self._link(from_info, to_info, self.cache_types)
-
-    def _link(self, from_info, to_info, link_types):
-        assert self.isfile(from_info)
-
-        self.makedirs(to_info.parent)
-
-        self._try_links(from_info, to_info, link_types)
-
-    @slow_link_guard
-    def _try_links(self, from_info, to_info, link_types):
-        while link_types:
-            link_method = getattr(self, link_types[0])
-            try:
-                self._do_link(from_info, to_info, link_method)
-                self.cache_type_confirmed = True
-                return
-
-            except DvcException as exc:
-                msg = "Cache type '{}' is not supported: {}"
-                logger.debug(msg.format(link_types[0], str(exc)))
-                del link_types[0]
-
-        raise DvcException("no possible cache types left to try out.")
-
-    def _do_link(self, from_info, to_info, link_method):
-        if self.exists(to_info):
-            raise DvcException("Link '{}' already exists!".format(to_info))
-
-        link_method(from_info, to_info)
-
-        if self.protected:
-            self.protect(to_info)
-
-        msg = "Created {}'{}': {} -> {}".format(
-            "protected " if self.protected else "",
-            self.cache_types[0],
-            from_info,
-            to_info,
-        )
-        logger.debug(msg)
+        self.copy(from_info, to_info)
 
     def _save_file(self, path_info, checksum, save_link=True):
         assert checksum
@@ -443,15 +391,9 @@ class RemoteBASE(object):
         return False
 
     def isfile(self, path_info):
-        """Optional: Overwrite only if the remote has a way to distinguish
-        between a directory and a file.
-        """
-        return True
+        raise NotImplementedError
 
     def isdir(self, path_info):
-        """Optional: Overwrite only if the remote has a way to distinguish
-        between a directory and a file.
-        """
         return False
 
     def walk(self, path_info):
@@ -570,15 +512,6 @@ class RemoteBASE(object):
     def copy(self, from_info, to_info):
         raise RemoteActionNotImplemented("copy", self.scheme)
 
-    def symlink(self, from_info, to_info):
-        raise RemoteActionNotImplemented("symlink", self.scheme)
-
-    def hardlink(self, from_info, to_info):
-        raise RemoteActionNotImplemented("hardlink", self.scheme)
-
-    def reflink(self, from_info, to_info):
-        raise RemoteActionNotImplemented("reflink", self.scheme)
-
     def exists(self, path_info):
         raise NotImplementedError
 
@@ -787,10 +720,7 @@ class RemoteBASE(object):
         return True
 
     def makedirs(self, path_info):
-        """Optional: Implement only if the remote needs to create
-        directories before copying/linking/moving data
-        """
-        pass
+        raise NotImplementedError
 
     def _checkout_dir(
         self, path_info, checksum, force, progress_callback=None
diff --git a/dvc/remote/local/__init__.py b/dvc/remote/local/__init__.py
index 1c1256efd..43e679dbb 100644
--- a/dvc/remote/local/__init__.py
+++ b/dvc/remote/local/__init__.py
@@ -1,12 +1,16 @@
 from __future__ import unicode_literals
 
+from copy import copy
+
 from dvc.scheme import Schemes
+from dvc.remote.local.slow_link_detection import slow_link_guard
 from dvc.utils.compat import str, fspath_py35, open
 
 import os
 import stat
 import errno
 from shortuuid import uuid
+import shutil
 import logging
 from functools import partial
 
@@ -47,6 +51,12 @@ class RemoteLOCAL(RemoteBASE):
     UNPACKED_DIR_SUFFIX = ".unpacked"
 
     DEFAULT_CACHE_TYPES = ["reflink", "copy"]
+    CACHE_TYPE_MAP = {
+        "copy": shutil.copyfile,
+        "symlink": System.symlink,
+        "hardlink": System.hardlink,
+        "reflink": System.reflink,
+    }
 
     SHARED_MODE_MAP = {None: (0o644, 0o755), "group": (0o664, 0o775)}
 
@@ -61,6 +71,15 @@ class RemoteLOCAL(RemoteBASE):
             # cache files are set to be read-only for everyone
             self._file_mode = stat.S_IREAD | stat.S_IRGRP | stat.S_IROTH
 
+        types = config.get(Config.SECTION_CACHE_TYPE, None)
+        if types:
+            if isinstance(types, str):
+                types = [t.strip() for t in types.split(",")]
+            self.cache_types = types
+        else:
+            self.cache_types = copy(self.DEFAULT_CACHE_TYPES)
+        self.cache_type_confirmed = False
+
         # A clunky way to detect cache dir
         storagepath = config.get(Config.SECTION_LOCAL_STORAGEPATH, None)
         cache_dir = config.get(Config.SECTION_REMOTE_URL, storagepath)
@@ -107,14 +126,80 @@ class RemoteLOCAL(RemoteBASE):
 
         return self.checksum_to_path_info(md5).url
 
-    @staticmethod
-    def exists(path_info):
+    def exists(self, path_info):
         assert path_info.scheme == "local"
         return os.path.lexists(fspath_py35(path_info))
 
     def makedirs(self, path_info):
         makedirs(path_info, exist_ok=True, mode=self._dir_mode)
 
+    def link(self, from_info, to_info):
+        self._link(from_info, to_info, self.cache_types)
+
+    def _link(self, from_info, to_info, link_types):
+        from_path = from_info.fspath
+        to_path = to_info.fspath
+
+        assert os.path.isfile(from_path)
+
+        dname = os.path.dirname(to_path)
+        if not os.path.exists(dname):
+            os.makedirs(dname)
+
+        # NOTE: just create an empty file for an empty cache
+        if os.path.getsize(from_path) == 0:
+            open(to_path, "w+").close()
+
+            msg = "Created empty file: {} -> {}".format(from_path, to_path)
+            logger.debug(msg)
+            return
+
+        self._try_links(from_info, to_info, link_types)
+
+    @classmethod
+    def _get_link_method(cls, link_type):
+        try:
+            return cls.CACHE_TYPE_MAP[link_type]
+        except KeyError:
+            raise DvcException(
+                "Cache type: '{}' not supported!".format(link_type)
+            )
+
+    def _do_link(self, from_info, to_info, link_method):
+        if self.exists(to_info):
+            raise DvcException("Link '{}' already exists!".format(to_info))
+        else:
+            link_method(from_info.fspath, to_info.fspath)
+
+        if self.protected:
+            self.protect(to_info)
+
+        msg = "Created {}'{}': {} -> {}".format(
+            "protected " if self.protected else "",
+            self.cache_types[0],
+            from_info,
+            to_info,
+        )
+        logger.debug(msg)
+
+    @slow_link_guard
+    def _try_links(self, from_info, to_info, link_types):
+        i = len(link_types)
+        while i > 0:
+            link_method = self._get_link_method(link_types[0])
+            try:
+                self._do_link(from_info, to_info, link_method)
+                self.cache_type_confirmed = True
+                return
+
+            except DvcException as exc:
+                msg = "Cache type '{}' is not supported: {}"
+                logger.debug(msg.format(link_types[0], str(exc)))
+                del link_types[0]
+                i -= 1
+
+        raise DvcException("no possible cache types left to try out.")
+
     def already_cached(self, path_info):
         assert path_info.scheme in ["", "local"]
 
@@ -136,18 +221,12 @@ class RemoteLOCAL(RemoteBASE):
 
         return False
 
-    @staticmethod
-    def isfile(path_info):
+    def isfile(self, path_info):
         return os.path.isfile(fspath_py35(path_info))
 
-    @staticmethod
-    def isdir(path_info):
+    def isdir(self, path_info):
         return os.path.isdir(fspath_py35(path_info))
 
-    @staticmethod
-    def getsize(path_info):
-        return os.path.getsize(fspath_py35(path_info))
-
     def walk(self, path_info):
         return dvc_walk(path_info, self.repo.dvcignore)
 
@@ -174,44 +253,6 @@ class RemoteLOCAL(RemoteBASE):
 
         move(from_info, to_info, mode=mode)
 
-    @staticmethod
-    def copy(from_info, to_info):
-        System.copy(from_info, to_info)
-
-    @staticmethod
-    def symlink(from_info, to_info):
-        System.symlink(from_info, to_info)
-
-    def hardlink(self, from_info, to_info):
-        # If there are a lot of empty files (which happens a lot in datasets),
-        # and the cache type is `hardlink`, we might reach link limits and
-        # will get something like: `too many links error`
-        #
-        # This is because all those empty files will have the same checksum
-        # (i.e. 68b329da9893e34099c7d8ad5cb9c940), therfore, they will be
-        # linked to the same file in the cache.
-        #
-        # From https://en.wikipedia.org/wiki/Hard_link
-        #   * ext4 limits the number of hard links on a file to 65,000
-        #   * Windows with NTFS has a limit of 1024 hard links on a file
-        #
-        # That's why we simply create an empty file rather than a link.
-        if self.getsize(from_info) == 0:
-            self.open(to_info, "w").close()
-
-            logger.debug(
-                "Created empty file: {src} -> {dest}".format(
-                    src=str(from_info), dest=str(to_info)
-                )
-            )
-            return
-
-        System.hardlink(from_info, to_info)
-
-    @staticmethod
-    def reflink(from_info, to_info):
-        System.reflink(from_info, to_info)
-
     def cache_exists(self, checksums, jobs=None, name=None):
         return [
             checksum
@@ -245,8 +286,8 @@ class RemoteLOCAL(RemoteBASE):
             name=name,
         )
 
-    @staticmethod
-    def open(path_info, mode="r", encoding=None):
+    def open(self, path_info, mode="r", encoding=None):
+        assert mode in {"r", "rt", "rb"}
         return open(fspath_py35(path_info), mode=mode, encoding=encoding)
 
     def _group(self, checksum_infos, show_checksums=False):
diff --git a/dvc/remote/local/slow_link_detection.py b/dvc/remote/local/slow_link_detection.py
new file mode 100644
index 000000000..7bd3bb143
--- /dev/null
+++ b/dvc/remote/local/slow_link_detection.py
@@ -0,0 +1,62 @@
+import logging
+import time
+
+import colorama
+from dvc.config import Config
+
+logger = logging.getLogger(__name__)
+
+
+class SlowLinkDetectorDecorator(object):
+    LINKING_TIMEOUT_SECONDS = 10.0
+    was_displayed = False
+
+    @classmethod
+    def should_display(cls):
+        if not cls.was_displayed:
+            cls.was_displayed = True
+            return True
+        return False
+
+    def __init__(self, method):
+        self.method = method
+
+    def __call__(self, *args, **kwargs):
+        start = time.time()
+        result = self.method(*args, **kwargs)
+        execution_time_seconds = time.time() - start
+
+        if (
+            execution_time_seconds >= self.LINKING_TIMEOUT_SECONDS
+            and self.should_display()
+        ):
+            msg = (
+                "You can cut execution time considerably by using a different"
+                " `cache.type` configuration.\nSee "
+                "{blue}https://dvc.org/doc/commands-reference/config#cache{"
+                "reset} for more information.\n"
+                "To disable this message, run:\n"
+                "'dvc config cache.slow_link_warning false'".format(
+                    blue=colorama.Fore.BLUE, reset=colorama.Fore.RESET
+                )
+            )
+            logger.warning(msg)
+
+        return result
+
+
+def slow_link_guard(method):
+    def call(remote_local, *args, **kwargs):
+        cache_config = remote_local.repo.config.config.get(
+            Config.SECTION_CACHE, {}
+        )
+        should_warn = cache_config.get(
+            Config.SECTION_CACHE_SLOW_LINK_WARNING, True
+        ) and not cache_config.get(Config.SECTION_CACHE_TYPE, None)
+
+        if should_warn:
+            decorated = SlowLinkDetectorDecorator(method)
+            return decorated(remote_local, *args, **kwargs)
+        return method(remote_local, *args, **kwargs)
+
+    return call
diff --git a/dvc/remote/slow_link_detection.py b/dvc/remote/slow_link_detection.py
deleted file mode 100644
index 0dd2285d6..000000000
--- a/dvc/remote/slow_link_detection.py
+++ /dev/null
@@ -1,50 +0,0 @@
-import logging
-import time
-import sys
-from functools import wraps
-
-import colorama
-from dvc.config import Config
-
-
-logger = logging.getLogger(__name__)
-this = sys.modules[__name__]
-
-this.timeout_seconds = 10.0
-this.already_displayed = False
-this.message = (
-    "You can cut execution time considerably by using a different "
-    "`cache.type` configuration.\n"
-    "See {blue}https://dvc.org/doc/commands-reference/config#cache{reset} "
-    "for more information.\n"
-    "To disable this message, run:\n"
-    "'dvc config cache.slow_link_warning false'".format(
-        blue=colorama.Fore.BLUE, reset=colorama.Fore.RESET
-    )
-)
-
-
-def slow_link_guard(f):
-    @wraps(f)
-    def wrapper(remote, *args, **kwargs):
-        if this.already_displayed:
-            return f(remote, *args, **kwargs)
-
-        config = remote.repo.config.config.get(Config.SECTION_CACHE, {})
-        cache_type = config.get(Config.SECTION_CACHE_TYPE)
-        should_warn = config.get(Config.SECTION_CACHE_SLOW_LINK_WARNING, True)
-
-        if not should_warn or cache_type:
-            return f(remote, *args, **kwargs)
-
-        start = time.time()
-        result = f(remote, *args, **kwargs)
-        delta = time.time() - start
-
-        if delta >= this.timeout_seconds:
-            logger.warning(this.message)
-            this.already_displayed = True
-
-        return result
-
-    return wrapper
diff --git a/dvc/remote/ssh/__init__.py b/dvc/remote/ssh/__init__.py
index 233f133ee..74280d7bb 100644
--- a/dvc/remote/ssh/__init__.py
+++ b/dvc/remote/ssh/__init__.py
@@ -47,10 +47,9 @@ class RemoteSSH(RemoteBASE):
     # We use conservative setting of 4 instead to not exhaust max sessions.
     CHECKSUM_JOBS = 4
 
-    DEFAULT_CACHE_TYPES = ["copy"]
-
     def __init__(self, repo, config):
         super(RemoteSSH, self).__init__(repo, config)
+
         url = config.get(Config.SECTION_REMOTE_URL)
         if url:
             parsed = urlparse(url)
@@ -165,54 +164,12 @@ class RemoteSSH(RemoteBASE):
         with self.ssh(path_info) as ssh:
             return ssh.isdir(path_info.path)
 
-    def isfile(self, path_info):
-        with self.ssh(path_info) as ssh:
-            return ssh.isfile(path_info.path)
-
-    def getsize(self, path_info):
-        with self.ssh(path_info) as ssh:
-            return ssh.getsize(path_info.path)
-
     def copy(self, from_info, to_info):
-        if not from_info.scheme == to_info.scheme == self.scheme:
-            raise NotImplementedError
-
-        with self.ssh(from_info) as ssh:
-            ssh.copy(from_info.path, to_info.path)
-
-    def symlink(self, from_info, to_info):
-        if not from_info.scheme == to_info.scheme == self.scheme:
-            raise NotImplementedError
-
-        with self.ssh(from_info) as ssh:
-            ssh.symlink(from_info.path, to_info.path)
-
-    def hardlink(self, from_info, to_info):
-        if not from_info.scheme == to_info.scheme == self.scheme:
-            raise NotImplementedError
-
-        # See dvc/remote/local/__init__.py - hardlink()
-        if self.getsize(from_info) == 0:
-
-            with self.ssh(to_info) as ssh:
-                ssh.sftp.open(to_info.path, "w").close()
-
-            logger.debug(
-                "Created empty file: {src} -> {dest}".format(
-                    src=str(from_info), dest=str(to_info)
-                )
-            )
-            return
-
-        with self.ssh(from_info) as ssh:
-            ssh.hardlink(from_info.path, to_info.path)
-
-    def reflink(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.reflink(from_info.path, to_info.path)
+            ssh.cp(from_info.path, to_info.path)
 
     def remove(self, path_info):
         if path_info.scheme != self.scheme:
@@ -250,12 +207,12 @@ class RemoteSSH(RemoteBASE):
 
     @contextmanager
     def open(self, path_info, mode="r", encoding=None):
-        assert mode in {"r", "rt", "rb", "wb"}
+        assert mode in {"r", "rt", "rb"}
 
         with self.ssh(path_info) as ssh, closing(
-            ssh.sftp.open(path_info.path, mode)
+            ssh.sftp.file(path_info.path, mode="r")
         ) as fd:
-            if "b" in mode:
+            if mode == "rb":
                 yield fd
             else:
                 yield io.TextIOWrapper(fd, encoding=encoding)
diff --git a/dvc/remote/ssh/connection.py b/dvc/remote/ssh/connection.py
index 4e1e52a43..43b380ada 100644
--- a/dvc/remote/ssh/connection.py
+++ b/dvc/remote/ssh/connection.py
@@ -60,13 +60,7 @@ class SSHConnection:
 
     def st_mode(self, path):
         with ignore_file_not_found():
-            return self.sftp.lstat(path).st_mode
-
-        return 0
-
-    def getsize(self, path):
-        with ignore_file_not_found():
-            return self.sftp.lstat(path).st_size
+            return self.sftp.stat(path).st_mode
 
         return 0
 
@@ -273,7 +267,8 @@ class SSHConnection:
         assert len(md5) == 32
         return md5
 
-    def copy(self, src, dest):
+    def cp(self, src, dest):
+        self.makedirs(posixpath.dirname(dest))
         self.execute("cp {} {}".format(src, dest))
 
     def open_max_sftp_channels(self):
@@ -287,23 +282,3 @@ class SSHConnection:
                         raise
                     break
         return self._sftp_channels
-
-    def open(self, *args, **kwargs):
-        return self.sftp.open(*args, **kwargs)
-
-    def symlink(self, src, dest):
-        self.sftp.symlink(src, dest)
-
-    def reflink(self, src, dest):
-        if self.uname == "Linux":
-            return self.execute("cp --reflink {} {}".format(src, dest))
-
-        if self.uname == "Darwin":
-            return self.execute("cp -c {} {}".format(src, dest))
-
-        raise DvcException(
-            "'{}' is not supported as a SSH remote".format(self.uname)
-        )
-
-    def hardlink(self, src, dest):
-        self.execute("ln {} {}".format(src, dest))
diff --git a/dvc/repo/move.py b/dvc/repo/move.py
index 6cd8b915c..84c506ca7 100644
--- a/dvc/repo/move.py
+++ b/dvc/repo/move.py
@@ -59,9 +59,9 @@ def move(self, from_path, to_path):
             os.path.join(os.curdir, os.path.dirname(to_path))
         )
 
-    to_out = Output.loads_from(
-        stage, [os.path.basename(to_path)], out.use_cache, out.metric
-    )[0]
+    to_path = os.path.relpath(to_path, stage.wdir)
+
+    to_out = Output.loads_from(stage, [to_path], out.use_cache, out.metric)[0]
 
     with self.state:
         out.move(to_out)
diff --git a/dvc/scm/base.py b/dvc/scm/base.py
index d7ad9d2a5..a1b2bd4fa 100644
--- a/dvc/scm/base.py
+++ b/dvc/scm/base.py
@@ -23,6 +23,11 @@ class FileNotInCommitError(SCMError):
     """
 
 
+class FileNotInTargetSubdirError(SCMError):
+    """Thrown when trying to place .gitignore for a file that not in
+    the file subdirectory."""
+
+
 class CloneError(SCMError):
     def __init__(self, url, path, cause):
         super(CloneError, self).__init__(
diff --git a/dvc/scm/git/__init__.py b/dvc/scm/git/__init__.py
index 2b9a67938..6afe2973b 100644
--- a/dvc/scm/git/__init__.py
+++ b/dvc/scm/git/__init__.py
@@ -11,6 +11,7 @@ from dvc.scm.base import (
     Base,
     SCMError,
     FileNotInRepoError,
+    FileNotInTargetSubdirError,
     CloneError,
     RevError,
 )
@@ -103,12 +104,22 @@ class Git(Base):
     def ignore_file(self):
         return self.GITIGNORE
 
-    def _get_gitignore(self, path):
-        ignore_file_dir = os.path.dirname(path)
+    def _get_gitignore(self, path, ignore_file_dir=None):
+        if not ignore_file_dir:
+            ignore_file_dir = os.path.dirname(os.path.realpath(path))
 
         assert os.path.isabs(path)
         assert os.path.isabs(ignore_file_dir)
 
+        if not path.startswith(ignore_file_dir):
+            msg = (
+                "{} file has to be located in one of '{}' subdirectories"
+                ", not outside '{}'"
+            )
+            raise FileNotInTargetSubdirError(
+                msg.format(self.GITIGNORE, path, ignore_file_dir)
+            )
+
         entry = relpath(path, ignore_file_dir).replace(os.sep, "/")
         # NOTE: using '/' prefix to make path unambiguous
         if len(entry) > 0 and entry[0] != "/":
@@ -132,7 +143,8 @@ class Git(Base):
         return False
 
     def ignore(self, path):
-        entry, gitignore = self._get_gitignore(path)
+        base_dir = os.path.dirname(path)
+        entry, gitignore = self._get_gitignore(path, base_dir)
 
         if self._ignored(entry, gitignore):
             return
diff --git a/dvc/stage.py b/dvc/stage.py
index 999e68ac9..cc91c81e0 100644
--- a/dvc/stage.py
+++ b/dvc/stage.py
@@ -139,7 +139,6 @@ class Stage(object):
     PARAM_OUTS = "outs"
     PARAM_LOCKED = "locked"
     PARAM_META = "meta"
-    PARAM_ALWAYS_CHANGED = "always_changed"
 
     SCHEMA = {
         Optional(PARAM_MD5): Or(str, None),
@@ -149,7 +148,6 @@ class Stage(object):
         Optional(PARAM_OUTS): Or(And(list, Schema([output.SCHEMA])), None),
         Optional(PARAM_LOCKED): bool,
         Optional(PARAM_META): object,
-        Optional(PARAM_ALWAYS_CHANGED): bool,
     }
 
     TAG_REGEX = r"^(?P<path>.*)@(?P<tag>[^\\/@:]*)$"
@@ -166,7 +164,6 @@ class Stage(object):
         locked=False,
         tag=None,
         state=None,
-        always_changed=False,
     ):
         if deps is None:
             deps = []
@@ -182,7 +179,6 @@ class Stage(object):
         self.md5 = md5
         self.locked = locked
         self.tag = tag
-        self.always_changed = always_changed
         self._state = state or {}
 
     def __repr__(self):
@@ -248,9 +244,6 @@ class Stage(object):
             )
             return True
 
-        if self.always_changed:
-            return True
-
         for dep in self.deps:
             status = dep.status()
             if status:
@@ -464,7 +457,6 @@ class Stage(object):
             wdir=wdir,
             cmd=kwargs.get("cmd", None),
             locked=kwargs.get("locked", False),
-            always_changed=kwargs.get("always_changed", False),
         )
 
         Stage._fill_stage_outputs(stage, **kwargs)
@@ -523,7 +515,6 @@ class Stage(object):
                 not ignore_build_cache
                 and stage.is_cached
                 and not stage.is_callback
-                and not stage.always_changed
             ):
                 logger.info("Stage is cached, skipping.")
                 return None
@@ -628,7 +619,6 @@ class Stage(object):
             md5=d.get(Stage.PARAM_MD5),
             locked=d.get(Stage.PARAM_LOCKED, False),
             tag=tag,
-            always_changed=d.get(Stage.PARAM_ALWAYS_CHANGED, False),
             state=state,
         )
 
@@ -639,21 +629,16 @@ class Stage(object):
 
     def dumpd(self):
         rel_wdir = relpath(self.wdir, os.path.dirname(self.path))
-
-        wdir = pathlib.PurePath(rel_wdir).as_posix()
-        wdir = wdir if wdir != "." else None
-
         return {
             key: value
             for key, value in {
                 Stage.PARAM_MD5: self.md5,
                 Stage.PARAM_CMD: self.cmd,
-                Stage.PARAM_WDIR: wdir,
+                Stage.PARAM_WDIR: pathlib.PurePath(rel_wdir).as_posix(),
                 Stage.PARAM_LOCKED: self.locked,
                 Stage.PARAM_DEPS: [d.dumpd() for d in self.deps],
                 Stage.PARAM_OUTS: [o.dumpd() for o in self.outs],
                 Stage.PARAM_META: self._state.get("meta"),
-                Stage.PARAM_ALWAYS_CHANGED: self.always_changed,
             }.items()
             if value
         }
@@ -854,7 +839,6 @@ class Stage(object):
                 if (
                     not force
                     and not self.is_callback
-                    and not self.always_changed
                     and self._already_cached()
                 ):
                     self.checkout()
@@ -901,7 +885,7 @@ class Stage(object):
         if self.changed_md5():
             ret.append("changed checksum")
 
-        if self.is_callback or self.always_changed:
+        if self.is_callback:
             ret.append("always changed")
 
         if ret:
diff --git a/dvc/system.py b/dvc/system.py
index c7c7cf8e4..b445ec01a 100644
--- a/dvc/system.py
+++ b/dvc/system.py
@@ -4,7 +4,6 @@ from dvc.utils.compat import str, open, fspath
 
 import os
 import errno
-import shutil
 
 
 class System(object):
@@ -12,11 +11,6 @@ class System(object):
     def is_unix():
         return os.name != "nt"
 
-    @staticmethod
-    def copy(src, dest):
-        src, dest = fspath(src), fspath(dest)
-        return shutil.copyfile(src, dest)
-
     @staticmethod
     def hardlink(source, link_name):
         import ctypes
diff --git a/dvc/version.py b/dvc/version.py
index 035914130..57c3c4234 100644
--- a/dvc/version.py
+++ b/dvc/version.py
@@ -7,7 +7,7 @@ import os
 import subprocess
 
 
-_BASE_VERSION = "0.59.2"
+_BASE_VERSION = "0.59.1"
 
 
 def _generate_version(base_version):
diff --git a/setup.py b/setup.py
index d718bc8c9..83db8b3fb 100644
--- a/setup.py
+++ b/setup.py
@@ -113,11 +113,10 @@ tests_requirements = [
     "Pygments",  # required by collective.checkdocs,
     "collective.checkdocs",
     "flake8",
-    "psutil",
     "flake8-docstrings",
     "pydocstyle<4.0",
     "jaraco.windows==3.9.2",
-    "mock-ssh-server>=0.6.0",
+    "mock-ssh-server>=0.5.0",
     "moto>=1.3.14.dev55",
     "rangehttpserver==1.2.0",
 ]


Test patch:
diff --git a/tests/func/test_add.py b/tests/func/test_add.py
index 82e474d05..4566351f3 100644
--- a/tests/func/test_add.py
+++ b/tests/func/test_add.py
@@ -565,12 +565,12 @@ def test_should_not_checkout_when_adding_cached_copy(repo_dir, dvc_repo):
 
     shutil.copy(repo_dir.BAR, repo_dir.FOO)
 
-    copy_spy = spy(dvc_repo.cache.local.copy)
+    copy_spy = spy(shutil.copyfile)
 
-    with patch.object(dvc_repo.cache.local, "copy", copy_spy):
-        dvc_repo.add(repo_dir.FOO)
+    RemoteLOCAL.CACHE_TYPE_MAP["copy"] = copy_spy
+    dvc_repo.add(repo_dir.FOO)
 
-        assert copy_spy.mock.call_count == 0
+    assert copy_spy.mock.call_count == 0
 
 
 @pytest.mark.parametrize(
@@ -585,13 +585,13 @@ def test_should_not_checkout_when_adding_cached_copy(repo_dir, dvc_repo):
 def test_should_relink_on_repeated_add(
     link, new_link, link_test_func, repo_dir, dvc_repo
 ):
-    dvc_repo.config.set("cache", "type", link)
+    dvc_repo.cache.local.cache_types = [link]
 
     dvc_repo.add(repo_dir.FOO)
     dvc_repo.add(repo_dir.BAR)
 
     os.remove(repo_dir.FOO)
-    getattr(dvc_repo.cache.local, link)(repo_dir.BAR, repo_dir.FOO)
+    RemoteLOCAL.CACHE_TYPE_MAP[link](repo_dir.BAR, repo_dir.FOO)
 
     dvc_repo.cache.local.cache_types = [new_link]
 
@@ -615,11 +615,10 @@ def test_should_relink_single_file_in_dir(link, link_func, dvc_repo, repo_dir):
     dvc_repo.unprotect(repo_dir.DATA_SUB)
 
     link_spy = spy(link_func)
+    RemoteLOCAL.CACHE_TYPE_MAP[link] = link_spy
+    dvc_repo.add(repo_dir.DATA_DIR)
 
-    with patch.object(dvc_repo.cache.local, link, link_spy):
-        dvc_repo.add(repo_dir.DATA_DIR)
-
-        assert link_spy.mock.call_count == 1
+    assert link_spy.mock.call_count == 1
 
 
 @pytest.mark.parametrize("link", ["hardlink", "symlink", "copy"])
diff --git a/tests/func/test_move.py b/tests/func/test_move.py
index 3004c7b29..83484e542 100644
--- a/tests/func/test_move.py
+++ b/tests/func/test_move.py
@@ -200,3 +200,13 @@ def test_move_should_save_stage_info(dvc_repo, repo_dir):
     dvc_repo.move(repo_dir.DATA_DIR, "new_name")
 
     assert dvc_repo.status() == {}
+
+
+def test_should_move_to_dir_on_non_default_stage_file(dvc_repo, repo_dir):
+    stage_file_name = "stage.dvc"
+
+    dvc_repo.add(repo_dir.FOO, fname=stage_file_name)
+
+    dvc_repo.move(repo_dir.FOO, repo_dir.DATA_DIR)
+
+    assert os.path.exists(os.path.join(repo_dir.DATA_DIR, repo_dir.FOO))
diff --git a/tests/func/test_repro.py b/tests/func/test_repro.py
index f9b0f43a4..f1a006ede 100644
--- a/tests/func/test_repro.py
+++ b/tests/func/test_repro.py
@@ -808,10 +808,6 @@ class TestReproExternalBase(TestDvc):
     def cache_scheme(self):
         return self.scheme
 
-    @property
-    def cache_type(self):
-        return "copy"
-
     @property
     def scheme(self):
         return None
@@ -866,7 +862,7 @@ class TestReproExternalBase(TestDvc):
         self.assertEqual(ret, 0)
         ret = main(["remote", "add", "myrepo", cache])
         self.assertEqual(ret, 0)
-        ret = main(["remote", "modify", "myrepo", "type", self.cache_type])
+        ret = main(["remote", "modify", "myrepo", "type", "hardlink"])
         self.assertEqual(ret, 0)
 
         remote_name = "myremote"
@@ -877,7 +873,7 @@ class TestReproExternalBase(TestDvc):
 
         ret = main(["remote", "add", remote_name, remote])
         self.assertEqual(ret, 0)
-        ret = main(["remote", "modify", remote_name, "type", self.cache_type])
+        ret = main(["remote", "modify", remote_name, "type", "hardlink"])
         self.assertEqual(ret, 0)
 
         self.dvc = DvcRepo(".")
@@ -1104,10 +1100,6 @@ class TestReproExternalLOCAL(TestReproExternalBase):
         self.assertEqual(ret, 0)
         self.dvc = DvcRepo(".")
 
-    @property
-    def cache_type(self):
-        return "hardlink"
-
     def should_test(self):
         return True
 
diff --git a/tests/func/test_run.py b/tests/func/test_run.py
index 567f30be0..383a4204a 100644
--- a/tests/func/test_run.py
+++ b/tests/func/test_run.py
@@ -560,18 +560,18 @@ class TestCmdRunCliMetrics(TestDvc):
 
 
 class TestCmdRunWorkingDirectory(TestDvc):
-    def test_default_wdir_is_not_written(self):
+    def test_default_wdir_is_written(self):
         stage = self.dvc.run(
             cmd="echo test > {}".format(self.FOO), outs=[self.FOO], wdir="."
         )
         d = load_stage_file(stage.relpath)
-        self.assertNotIn(Stage.PARAM_WDIR, d.keys())
+        self.assertEqual(d[Stage.PARAM_WDIR], ".")
 
         stage = self.dvc.run(
             cmd="echo test > {}".format(self.BAR), outs=[self.BAR]
         )
         d = load_stage_file(stage.relpath)
-        self.assertNotIn(Stage.PARAM_WDIR, d.keys())
+        self.assertEqual(d[Stage.PARAM_WDIR], ".")
 
     def test_fname_changes_path_and_wdir(self):
         dname = "dir"
diff --git a/tests/func/test_scm.py b/tests/func/test_scm.py
index 6c8d60276..ada5b4759 100644
--- a/tests/func/test_scm.py
+++ b/tests/func/test_scm.py
@@ -2,11 +2,12 @@ from __future__ import unicode_literals
 
 import os
 
+import pytest
 from git import Repo
 
-from dvc.system import System
 from dvc.utils.compat import str  # noqa: F401
 from dvc.scm import SCM, NoSCM, Git
+from dvc.scm.base import FileNotInTargetSubdirError
 
 from tests.basic_env import TestDir, TestGit, TestGitSubmodule
 from tests.utils import get_gitignore_content
@@ -96,14 +97,6 @@ class TestIgnore(object):
         assert entry == "/dir"
         assert gitignore == os.path.join(repo_dir._root_dir, Git.GITIGNORE)
 
-    def test_get_gitignore_symlink(self, git, repo_dir):
-        link = os.path.join(repo_dir.root_dir, "link")
-        target = os.path.join(repo_dir.root_dir, repo_dir.DATA_SUB)
-        System.symlink(target, link)
-        entry, gitignore = Git(repo_dir._root_dir)._get_gitignore(link)
-        assert entry == "/link"
-        assert gitignore == os.path.join(repo_dir.root_dir, Git.GITIGNORE)
-
     def test_get_gitignore_subdir(self, git, repo_dir):
         data_dir = os.path.join(
             repo_dir._root_dir, os.path.join("dir1", "file1")
@@ -123,6 +116,35 @@ class TestIgnore(object):
             repo_dir._root_dir, "dir1", Git.GITIGNORE
         )
 
+    def test_get_gitignore_ignorefile_dir(self, git, repo_dir):
+        git = Git(repo_dir._root_dir)
+
+        file_double_dir = os.path.join("dir1", "dir2", "file1")
+        data_dir1 = os.path.join(repo_dir._root_dir, file_double_dir)
+        dir1_real1 = os.path.realpath("dir1")
+        entry, gitignore = git._get_gitignore(data_dir1, dir1_real1)
+        assert entry == "/dir2/file1"
+        gitignore1 = os.path.join(repo_dir._root_dir, "dir1", Git.GITIGNORE)
+        assert gitignore == gitignore1
+
+        triple_dir = os.path.join("dir1", "dir2", "dir3")
+        data_dir2 = os.path.join(repo_dir._root_dir, triple_dir)
+        dir1_real2 = os.path.realpath("dir1")
+        entry, gitignore = git._get_gitignore(data_dir2, dir1_real2)
+        assert entry == "/dir2/dir3"
+        gitignore2 = os.path.join(repo_dir._root_dir, "dir1", Git.GITIGNORE)
+        assert gitignore == gitignore2
+
+    def test_get_gitignore_ignorefile_dir_upper_level(self, git, repo_dir):
+        git = Git(repo_dir._root_dir)
+
+        file_double_dir = os.path.join("dir1", "dir2", "file1")
+        data_dir1 = os.path.join(repo_dir._root_dir, file_double_dir)
+        ignore_file_dir = os.path.realpath(os.path.join("aa", "bb"))
+
+        with pytest.raises(FileNotInTargetSubdirError):
+            git._get_gitignore(data_dir1, ignore_file_dir)
+
     def test_gitignore_should_end_with_newline(self, git, repo_dir):
         git = Git(repo_dir._root_dir)
 
diff --git a/tests/func/test_stage.py b/tests/func/test_stage.py
index 4d4a92795..fa18160d8 100644
--- a/tests/func/test_stage.py
+++ b/tests/func/test_stage.py
@@ -102,10 +102,16 @@ class TestDefaultWorkingDirectory(TestDvc):
         )
 
         d = stage.dumpd()
-        self.assertNotIn(Stage.PARAM_WDIR, d.keys())
+        self.assertEqual(d[stage.PARAM_WDIR], ".")
 
         d = load_stage_file(stage.relpath)
-        self.assertNotIn(Stage.PARAM_WDIR, d.keys())
+        self.assertEqual(d[stage.PARAM_WDIR], ".")
+
+        del d[stage.PARAM_WDIR]
+        dump_stage_file(stage.relpath, d)
+
+        d = load_stage_file(stage.relpath)
+        self.assertIsNone(d.get(stage.PARAM_WDIR))
 
         with self.dvc.state:
             stage = Stage.load(self.dvc, stage.relpath)
diff --git a/tests/unit/command/test_run.py b/tests/unit/command/test_run.py
index 6e1f3ba77..4fde85a72 100644
--- a/tests/unit/command/test_run.py
+++ b/tests/unit/command/test_run.py
@@ -32,7 +32,6 @@ def test_run(mocker, dvc_repo):
             "outs-persist",
             "--outs-persist-no-cache",
             "outs-persist-no-cache",
-            "--always-changed",
             "command",
         ]
     )
@@ -59,7 +58,6 @@ def test_run(mocker, dvc_repo):
         ignore_build_cache=True,
         remove_outs=True,
         no_commit=True,
-        always_changed=True,
         cmd="command",
     )
 
@@ -85,7 +83,6 @@ def test_run_args_from_cli(mocker, dvc_repo):
         ignore_build_cache=False,
         remove_outs=False,
         no_commit=False,
-        always_changed=False,
         cmd="echo foo",
     )
 
@@ -111,6 +108,5 @@ def test_run_args_with_spaces(mocker, dvc_repo):
         ignore_build_cache=False,
         remove_outs=False,
         no_commit=False,
-        always_changed=False,
         cmd='echo "foo bar"',
     )
diff --git a/tests/unit/remote/ssh/test_connection.py b/tests/unit/remote/ssh/test_connection.py
index f0c12ec21..bdd1a98b4 100644
--- a/tests/unit/remote/ssh/test_connection.py
+++ b/tests/unit/remote/ssh/test_connection.py
@@ -2,14 +2,6 @@ from __future__ import unicode_literals
 
 import os
 import posixpath
-import platform
-import pytest
-import filecmp
-import tempfile
-
-from dvc.system import System
-from dvc.command.version import CmdVersion
-
 
 here = os.path.abspath(os.path.dirname(__file__))
 
@@ -72,42 +64,3 @@ def test_walk(tmp_path, ssh):
             paths.add(posixpath.join(root, entry))
 
     assert paths == expected
-
-
-@pytest.mark.skipif(
-    CmdVersion.get_fs_type(tempfile.gettempdir())[0]
-    not in ["xfs", "apfs", "btrfs"],
-    reason="Reflinks only work in specified file systems",
-)
-def test_reflink(repo_dir, ssh):
-    ssh.reflink("foo", "link")
-    assert filecmp.cmp("foo", "link")
-    assert not System.is_symlink("link")
-    assert not System.is_hardlink("link")
-
-
-@pytest.mark.skipif(
-    platform.system() == "Windows",
-    reason="sftp symlink is not supported on Windows",
-)
-def test_symlink(repo_dir, ssh):
-    ssh.symlink("foo", "link")
-    assert System.is_symlink("link")
-
-
-@pytest.mark.skipif(
-    platform.system() == "Windows",
-    reason="hardlink is temporarily not supported on Windows",
-)
-def test_hardlink(repo_dir, ssh):
-    ssh.hardlink("foo", "link")
-    assert System.is_hardlink("link")
-
-
-@pytest.mark.skipif(
-    platform.system() == "Windows",
-    reason="copy is temporarily not supported on Windows",
-)
-def test_copy(repo_dir, ssh):
-    ssh.copy("foo", "link")
-    assert filecmp.cmp("foo", "link")
diff --git a/tests/unit/remote/ssh/test_ssh.py b/tests/unit/remote/ssh/test_ssh.py
index 6e6e101aa..0dee1ac3e 100644
--- a/tests/unit/remote/ssh/test_ssh.py
+++ b/tests/unit/remote/ssh/test_ssh.py
@@ -7,10 +7,7 @@ from unittest import TestCase
 from mock import patch, mock_open
 import pytest
 
-from tests.func.test_data_cloud import get_ssh_url_mocked
-
 from dvc.remote.ssh import RemoteSSH
-from dvc.system import System
 
 
 class TestRemoteSSH(TestCase):
@@ -185,30 +182,3 @@ def test_ssh_gss_auth(mock_file, mock_exists, config, expected_gss_auth):
     mock_exists.assert_called_with(RemoteSSH.ssh_config_filename())
     mock_file.assert_called_with(RemoteSSH.ssh_config_filename())
     assert remote.gss_auth == expected_gss_auth
-
-
-def test_hardlink_optimization(repo_dir, ssh_server):
-    port = ssh_server.test_creds["port"]
-    user = ssh_server.test_creds["username"]
-
-    config = {
-        "url": get_ssh_url_mocked(user, port),
-        "port": port,
-        "user": user,
-        "keyfile": ssh_server.test_creds["key_filename"],
-    }
-    remote = RemoteSSH(None, config)
-
-    from_info = remote.path_info / "empty"
-    to_info = remote.path_info / "link"
-
-    with remote.open(from_info, "wb"):
-        pass
-
-    if os.name == "nt":
-        link_path = "c:" + to_info.path.replace("/", "\\")
-    else:
-        link_path = to_info.path
-
-    remote.hardlink(from_info, to_info)
-    assert not System.is_hardlink(link_path)
diff --git a/tests/unit/remote/test_slow_link_detection.py b/tests/unit/remote/test_slow_link_detection.py
index 141b4ff95..3b981e69e 100644
--- a/tests/unit/remote/test_slow_link_detection.py
+++ b/tests/unit/remote/test_slow_link_detection.py
@@ -1,47 +1,114 @@
-import pytest
-import mock
+import logging
 
-import dvc.remote.slow_link_detection
-from dvc.remote.slow_link_detection import slow_link_guard
+import colorama
+import pytest
 from dvc.config import Config
+from dvc.remote.local.slow_link_detection import (
+    slow_link_guard,
+    SlowLinkDetectorDecorator,
+)
+from mock import Mock, patch
+
+
+@patch(
+    "dvc.remote.local.slow_link_detection.SlowLinkDetectorDecorator",
+    autospec=True,
+)
+class TestSlowLinkGuard(object):
+    @pytest.fixture(autouse=True)
+    def setUp(self):
+        self.method = Mock()
+        self.remote_local_mock = Mock()
+        self.config_mock = Mock()
+        self.remote_local_mock.repo.config.config.get.return_value = (
+            self.config_mock
+        )
+
+    def _cache_config(self, slow_link_warning, cache_type):
+        def config_side_effect(section_name, _):
+            if section_name == Config.SECTION_CACHE_SLOW_LINK_WARNING:
+                return slow_link_warning
+            elif section_name == Config.SECTION_CACHE_TYPE:
+                return cache_type
+
+        self.config_mock.get.side_effect = config_side_effect
+
+    def test_should_decorate_on_slow_link_warning_and_no_cache_type(
+        self, SlowLinkDetectorDecoratorClassMock
+    ):
+        self._cache_config(slow_link_warning=True, cache_type=None)
+
+        slow_link_guard(self.method)(self.remote_local_mock)
+
+        assert 1 == SlowLinkDetectorDecoratorClassMock.call_count
+        assert 0 == self.method.call_count
 
+    @pytest.mark.parametrize(
+        "slow_link_warning, cache_type",
+        [(True, "any_cache_type"), (False, None)],
+    )
+    def test_should_not_decorate(
+        self, SlowLinkDetectorDecoratorClassMock, slow_link_warning, cache_type
+    ):
+        self._cache_config(
+            slow_link_warning=slow_link_warning, cache_type=cache_type
+        )
 
-@pytest.fixture(autouse=True)
-def timeout_immediately(monkeypatch):
-    monkeypatch.setattr(dvc.remote.slow_link_detection, "timeout_seconds", 0.0)
+        slow_link_guard(self.method)(self.remote_local_mock)
 
+        assert 0 == SlowLinkDetectorDecoratorClassMock.call_count
+        assert 1 == self.method.call_count
 
-@pytest.fixture
-def make_remote():
-    def _make_remote(cache_type=None, should_warn=True):
-        remote = mock.Mock()
-        remote.repo.config.config.get.return_value = {
-            Config.SECTION_CACHE_TYPE: cache_type,
-            Config.SECTION_CACHE_SLOW_LINK_WARNING: should_warn,
-        }
-        return remote
 
-    return _make_remote
+class TestSlowLinkDetectorDecorator(object):
+    @pytest.fixture(autouse=True)
+    def setUp(self):
+        self.args_mock = Mock()
+        self.method_result_mock = Mock()
 
+        def method_effect(arg):
+            assert self.args_mock == arg
+            return self.method_result_mock
 
-def test_show_warning_once(caplog, make_remote):
-    remote = make_remote()
-    slow_link_guard(lambda x: None)(remote)
-    slow_link_guard(lambda x: None)(remote)
+        self.method_mock = Mock(side_effect=method_effect)
 
-    assert caplog.records[0].message == dvc.remote.slow_link_detection.message
-    assert len(caplog.records) == 1
+    @patch.object(SlowLinkDetectorDecorator, "LINKING_TIMEOUT_SECONDS", 0.0)
+    def test_should_log_only_once(self, caplog):
 
+        with caplog.at_level(logging.WARNING, logger="dvc"):
+            test_instance = SlowLinkDetectorDecorator(self.method_mock)
+            result = test_instance(self.args_mock)
+            assert self.method_result_mock == result
 
-def test_dont_warn_when_cache_type_is_set(caplog, make_remote):
-    remote = make_remote(cache_type="copy")
-    slow_link_guard(lambda x: None)(remote)
+            # slow link guard keeps creating new SlowLinkDetectorDecorators
+            # each time it is called, so we need to make sure that we are
+            # preventing every new instance of this class from displaying the
+            # message
+            test_instance2 = SlowLinkDetectorDecorator(self.method_mock)
+            result = test_instance2(self.args_mock)
+            assert self.method_result_mock == result
 
-    assert len(caplog.records) == 0
+        msg = (
+            "You can cut execution time considerably by using a different"
+            " `cache.type` configuration.\nSee "
+            "{blue}https://dvc.org/doc/commands-reference/config#cache{"
+            "reset} for more information.\n"
+            "To disable this message, run:\n"
+            "'dvc config cache.slow_link_warning false'".format(
+                blue=colorama.Fore.BLUE, reset=colorama.Fore.RESET
+            )
+        )
+        assert 1 == len(caplog.messages)
+        assert msg == caplog.messages[0]
 
+    @patch.object(
+        SlowLinkDetectorDecorator, "LINKING_TIMEOUT_SECONDS", 99999.0
+    )
+    def test_should_not_log(self, caplog):
+        test_instance = SlowLinkDetectorDecorator(self.method_mock)
 
-def test_dont_warn_when_warning_is_disabled(caplog, make_remote):
-    remote = make_remote(should_warn=False)
-    slow_link_guard(lambda x: None)(remote)
+        with caplog.at_level(logging.WARNING, logger="dvc"):
+            result = test_instance(self.args_mock)
 
-    assert len(caplog.records) == 0
+        assert self.method_result_mock == result
+        assert 0 == len(caplog.messages)
diff --git a/tests/unit/test_stage.py b/tests/unit/test_stage.py
index f87fbe0f6..41460f9e6 100644
--- a/tests/unit/test_stage.py
+++ b/tests/unit/test_stage.py
@@ -107,10 +107,3 @@ def test_stage_run_ignore_sigint(mocker):
     assert communicate.called_once_with()
     signal_mock.assert_any_call(signal.SIGINT, signal.SIG_IGN)
     assert signal.getsignal(signal.SIGINT) == signal.default_int_handler
-
-
-def test_always_changed():
-    stage = Stage(None, "path", always_changed=True)
-    stage.save()
-    assert stage.changed()
-    assert stage.status()["path"] == ["always changed"]