tests: support windows

tests/func/test_data_cloud.py

@@ -9,7 +9,6 @@ import platform
 import copy
 import logging
 import pytest
-import posixpath
 
 from mock import patch
 
@@ -28,10 +27,8 @@ 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
@@ -452,25 +449,6 @@ 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_ssh.py

@@ -1,15 +1,15 @@
 import getpass
 import os
 import sys
-import posixpath
 
 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.path_info import URLInfo
 from dvc.system import System
 
 
@@ -188,21 +188,27 @@ def test_ssh_gss_auth(mock_file, mock_exists, config, 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": "ssh://localhost",
-        "hostname": ssh_server.test_creds["host"],
-        "port": ssh_server.test_creds["port"],
-        "user": ssh_server.test_creds["username"],
+        "url": get_ssh_url_mocked(user, port),
+        "port": port,
+        "user": user,
         "keyfile": ssh_server.test_creds["key_filename"],
     }
     remote = RemoteSSH(None, config)
 
-    open("empty", "w").close()
+    from_info = remote.path_info / "empty"
+    to_info = remote.path_info / "link"
 
-    def urlpath(path):
-        netloc = remote.path_info.netloc
-        path = posixpath.abspath(path)
-        return URLInfo("ssh://{}/{}".format(netloc, path))
+    with remote.open(from_info, "wb"):
+        pass
 
-    remote.hardlink(urlpath("empty"), urlpath("link"))
-    assert not System.is_hardlink("link")
+    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)