test: refactor ssh related functions in test helper classes

tests/func/test_data_cloud.py

@@ -36,6 +36,7 @@
     Azure,
     GDrive,
     S3,
+    SSHMocked,
     OSS,
     TEST_CONFIG,
     TEST_SECTION,
@@ -46,7 +47,6 @@
     get_gcp_url,
     get_hdfs_url,
     get_local_url,
-    get_ssh_url_mocked,
 )
 
 
@@ -318,13 +318,13 @@ def _setup_cloud(self):
 
     def _get_url(self):
         user = self.ssh_server.test_creds["username"]
-        return get_ssh_url_mocked(user, self.ssh_server.port)
+        return SSHMocked.get_url(user, self.ssh_server.port)
 
     def _get_keyfile(self):
         return self.ssh_server.test_creds["key_filename"]
 
     def _should_test(self):
-        return True
+        return SSHMocked.should_test()
 
     def _get_cloud_class(self):
         return RemoteSSH

tests/func/test_repro.py

@@ -37,9 +37,9 @@
 from tests.remotes import (
     _should_test_gcp,
     _should_test_hdfs,
-    _should_test_ssh,
-    get_ssh_url_mocked,
     S3,
+    SSH,
+    SSHMocked,
     TEST_AWS_REPO_BUCKET,
     TEST_GCP_REPO_BUCKET,
 )
@@ -1058,7 +1058,7 @@ class TestReproExternalSSH(TestReproExternalBase):
     _dir = None
 
     def should_test(self):
-        return _should_test_ssh()
+        return SSH.should_test()
 
     @property
     def scheme(self):
@@ -1623,11 +1623,11 @@ def test_ssh_dir_out(tmp_dir, dvc, ssh_server):
     port = ssh_server.port
     keyfile = ssh_server.test_creds["key_filename"]
 
-    remote_url = get_ssh_url_mocked(user, port)
+    remote_url = SSHMocked.get_url(user, port)
     assert main(["remote", "add", "upstream", remote_url]) == 0
     assert main(["remote", "modify", "upstream", "keyfile", keyfile]) == 0
 
-    cache_url = get_ssh_url_mocked(user, port)
+    cache_url = SSHMocked.get_url(user, port)
     assert main(["remote", "add", "sshcache", cache_url]) == 0
     assert main(["config", "cache.ssh", "sshcache"]) == 0
     assert main(["remote", "modify", "sshcache", "keyfile", keyfile]) == 0

tests/remotes.py

@@ -66,23 +66,6 @@ def _should_test_gcp():
     return True
 
 
-def _should_test_ssh():
-    do_test = env2bool("DVC_TEST_SSH", undefined=None)
-    if do_test is not None:
-        return do_test
-
-    # FIXME: enable on windows
-    if os.name == "nt":
-        return False
-
-    try:
-        check_output(["ssh", "-o", "BatchMode=yes", "127.0.0.1", "ls"])
-    except (CalledProcessError, IOError):
-        return False
-
-    return True
-
-
 def _should_test_hdfs():
     if platform.system() != "Linux":
         return False
@@ -114,31 +97,6 @@ def get_local_url():
     return get_local_storagepath()
 
 
-def get_ssh_url():
-    return "ssh://{}@127.0.0.1:22{}".format(
-        getpass.getuser(), get_local_storagepath()
-    )
-
-
-def get_ssh_url_mocked(user, port):
-    path = get_local_storagepath()
-    if os.name == "nt":
-        # NOTE: On Windows get_local_storagepath() will return an ntpath
-        # that looks something like `C:\some\path`, which is not compatible
-        # with SFTP paths [1], so we need to convert it to a proper posixpath.
-        # To do that, we should construct a posixpath that would be relative
-        # to the server's root. In our case our ssh server is running with
-        # `c:/` as a root, and our URL format requires absolute paths, so the
-        # resulting path would look like `/some/path`.
-        #
-        # [1]https://tools.ietf.org/html/draft-ietf-secsh-filexfer-13#section-6
-        drive, path = os.path.splitdrive(path)
-        assert drive.lower() == "c:"
-        path = path.replace("\\", "/")
-    url = "ssh://{}@127.0.0.1:{}{}".format(user, port, path)
-    return url
-
-
 def get_hdfs_url():
     return "hdfs://{}@127.0.0.1{}".format(
         getpass.getuser(), get_local_storagepath()
@@ -269,8 +227,53 @@ def get_url():
 
 
 class SSH:
-    should_test = _should_test_ssh
-    get_url = get_ssh_url
+    @staticmethod
+    def should_test():
+        do_test = env2bool("DVC_TEST_SSH", undefined=None)
+        if do_test is not None:
+            return do_test
+
+        # FIXME: enable on windows
+        if os.name == "nt":
+            return False
+
+        try:
+            check_output(["ssh", "-o", "BatchMode=yes", "127.0.0.1", "ls"])
+        except (CalledProcessError, IOError):
+            return False
+
+        return True
+
+    @staticmethod
+    def get_url():
+        return "ssh://{}@127.0.0.1:22{}".format(
+            getpass.getuser(), get_local_storagepath()
+        )
+
+
+class SSHMocked:
+    should_test = lambda: True  # noqa: E731
+
+    @staticmethod
+    def get_url(user, port):
+        path = get_local_storagepath()
+        if os.name == "nt":
+            # NOTE: On Windows get_local_storagepath() will return an
+            # ntpath that looks something like `C:\some\path`, which is not
+            # compatible with SFTP paths [1], so we need to convert it to
+            # a proper posixpath.
+            # To do that, we should construct a posixpath that would be
+            # relative to the server's root.
+            # In our case our ssh server is running with `c:/` as a root,
+            # and our URL format requires absolute paths, so the
+            # resulting path would look like `/some/path`.
+            #
+            # [1]https://tools.ietf.org/html/draft-ietf-secsh-filexfer-13#section-6
+            drive, path = os.path.splitdrive(path)
+            assert drive.lower() == "c:"
+            path = path.replace("\\", "/")
+        url = "ssh://{}@127.0.0.1:{}{}".format(user, port, path)
+        return url
 
 
 class HDFS:

tests/unit/remote/ssh/test_ssh.py

@@ -9,7 +9,7 @@
 
 from dvc.remote.ssh import RemoteSSH
 from dvc.system import System
-from tests.remotes import get_ssh_url_mocked
+from tests.remotes import SSHMocked
 
 
 class TestRemoteSSH(TestCase):
@@ -191,7 +191,7 @@ def test_hardlink_optimization(repo_dir, ssh_server):
     user = ssh_server.test_creds["username"]
 
     config = {
-        "url": get_ssh_url_mocked(user, port),
+        "url": SSHMocked.get_url(user, port),
         "port": port,
         "user": user,
         "keyfile": ssh_server.test_creds["key_filename"],