import: fix importing into subdirectory bug (#4340)

* tests: test importing into subdirectory

* import: fix bug when output subdir doesnt already exist

* use urlparse to determine if output path is local, not os.path.exists

* account for windows drive letters

* import: error out if output subdirectory does not exist

dvc/utils/__init__.py

@@ -363,19 +363,22 @@ def resolve_output(inp, out):
 
 
 def resolve_paths(repo, out):
+    from urllib.parse import urlparse
     from ..dvcfile import DVC_FILE_SUFFIX
     from ..path_info import PathInfo
     from .fs import contains_symlink_up_to
 
-    abspath = os.path.abspath(out)
+    abspath = PathInfo(os.path.abspath(out))
     dirname = os.path.dirname(abspath)
     base = os.path.basename(os.path.normpath(out))
 
-    # NOTE: `out` might not exist yet, so using `dirname`(aka `wdir`) to check
-    # if it is a local path.
+    scheme = urlparse(out).scheme
+    if os.name == "nt" and scheme == abspath.drive[0].lower():
+        # urlparse interprets windows drive letters as URL scheme
+        scheme = ""
     if (
-        os.path.exists(dirname)  # out might not exist yet, so
-        and PathInfo(abspath).isin_or_eq(repo.root_dir)
+        not scheme
+        and abspath.isin_or_eq(repo.root_dir)
         and not contains_symlink_up_to(abspath, repo.root_dir)
     ):
         wdir = dirname

tests/func/test_import.py

@@ -9,6 +9,7 @@ from dvc.cache import Cache
 from dvc.config import NoRemoteError
 from dvc.dvcfile import Dvcfile
 from dvc.exceptions import DownloadError, PathMissingError
+from dvc.stage.exceptions import StagePathNotFoundError
 from dvc.system import System
 from dvc.utils.fs import makedirs, remove
 
@@ -132,6 +133,28 @@ def test_import_file_from_dir(tmp_dir, scm, dvc, erepo_dir):
     assert (tmp_dir / "X.dvc").exists()
 
 
+def test_import_file_from_dir_to_dir(tmp_dir, scm, dvc, erepo_dir):
+    with erepo_dir.chdir():
+        erepo_dir.dvc_gen({"dir": {"foo": "foo"}}, commit="create dir")
+
+    with pytest.raises(StagePathNotFoundError):
+        dvc.imp(
+            os.fspath(erepo_dir),
+            os.path.join("dir", "foo"),
+            out=os.path.join("dir", "foo"),
+        )
+
+    tmp_dir.gen({"dir": {}})
+    dvc.imp(
+        os.fspath(erepo_dir),
+        os.path.join("dir", "foo"),
+        out=os.path.join("dir", "foo"),
+    )
+    assert not (tmp_dir / "foo.dvc").exists()
+    assert (tmp_dir / "dir" / "foo").read_text() == "foo"
+    assert (tmp_dir / "dir" / "foo.dvc").exists()
+
+
 def test_import_non_cached(erepo_dir, tmp_dir, dvc, scm):
     src = "non_cached_output"
     dst = src + "_imported"