tests: functional ssh symlink

tests/func/test_data_cloud.py

@@ -9,6 +9,7 @@ import platform
 import copy
 import logging
 import pytest
+import posixpath
 
 from mock import patch
 
@@ -27,8 +28,10 @@ from dvc.data_cloud import (
     RemoteHTTP,
 )
 from dvc.remote.base import STATUS_OK, STATUS_NEW, STATUS_DELETED
+from dvc.path_info import URLInfo
 from dvc.utils import file_md5
 from dvc.utils.stage import load_stage_file, dump_stage_file
+from dvc.system import System
 
 from tests.basic_env import TestDvc
 from tests.utils import spy
@@ -449,6 +452,25 @@ class TestRemoteSSH(TestDataCloudBase):
     def _get_cloud_class(self):
         return RemoteSSH
 
+    def test_symlinks(self):
+        self._setup_cloud()
+
+        remote = self.cloud.get_remote()
+
+        os.remove("bar")
+
+        def urlinfo(path):
+            cwd = os.getcwd()
+            path = posixpath.abspath(os.path.join(cwd, path))
+            netloc = remote.path_info.netloc
+            return URLInfo("ssh://{}{}".format(netloc, path))
+
+        from_info = urlinfo("foo")
+        to_info = urlinfo("bar")
+
+        remote.symlink(from_info, to_info)
+        assert System.is_symlink("bar")
+
 
 @pytest.mark.usefixtures("ssh_server")
 class TestRemoteSSHMocked(TestDataCloudBase):

tests/unit/remote/ssh/test_connection.py

@@ -77,23 +77,22 @@ def test_walk(tmp_path, ssh):
 
     assert paths == expected
 
-@pytest.mark.skipif(platform.system() != "Darwin",
-                    reason="Only works on OSX (because of APFS)")
+
+@pytest.mark.skipif(
+    platform.system() != "Darwin", reason="Only works on OSX (because of APFS)"
+)
 def test_reflink(chdir_tmp, foo, ssh):
     ssh.reflink("foo", "bar")
     assert filecmp.cmp("foo", "bar")
     assert not System.is_symlink("bar")
     assert not System.is_hardlink("bar")
 
-@pytest.mark.skip(reason="MockSSH doesn't support symlinks")
-def test_symlink(chdir_tmp, foo, ssh):
-    ssh.symlink("foo", "bar")
-    assert System.is_symlink("bar")
 
 def test_hardlink(chdir_tmp, foo, ssh):
     ssh.hardlink("foo", "bar")
     assert System.is_hardlink("bar")
 
+
 def test_copy(chdir_tmp, foo, ssh):
     ssh.copy("foo", "bar")
     assert filecmp.cmp("foo", "bar")

tests/unit/remote/ssh/test_ssh.py

@@ -186,6 +186,7 @@ def test_ssh_gss_auth(mock_file, mock_exists, config, expected_gss_auth):
     mock_file.assert_called_with(RemoteSSH.ssh_config_filename())
     assert remote.gss_auth == expected_gss_auth
 
+
 def test_hardlink_optimization(chdir_tmp, ssh_server):
     config = {
         "url": "ssh://localhost",