s3/gs: do not try to make directory on top level path

dvc/remote/gs.py

@@ -143,6 +143,9 @@ def walk_files(self, path_info):
             yield path_info.replace(fname)
 
     def makedirs(self, path_info):
+        if not path_info.path:
+            return
+
         self.gs.bucket(path_info.bucket).blob(
             (path_info / "").path
         ).upload_from_string("")

dvc/remote/s3.py

@@ -238,6 +238,9 @@ def makedirs(self, path_info):
         #
         # We are not creating directory objects for every parent prefix,
         # as it is not required.
+        if not path_info.path:
+            return
+
         dir_path = path_info / ""
         self.s3.put_object(Bucket=path_info.bucket, Key=dir_path.path, Body="")
 

tests/func/test_s3.py

@@ -2,6 +2,7 @@
 
 import boto3
 import moto.s3.models as s3model
+import pytest
 from moto import mock_s3
 
 from dvc.remote.s3 import RemoteS3
@@ -45,6 +46,30 @@ def test_copy_singlepart_preserve_etag():
     RemoteS3._copy(s3, from_info, to_info, {})
 
 
+@mock_s3
+@pytest.mark.parametrize(
+    "base_info",
+    [RemoteS3.path_cls("s3://bucket/"), RemoteS3.path_cls("s3://bucket/ns/")],
+)
+def test_link_created_on_non_nested_path(base_info, tmp_dir, dvc, scm):
+    remote = RemoteS3(dvc, {"url": str(base_info.parent)})
+    remote.s3.create_bucket(Bucket=base_info.bucket)
+    remote.s3.put_object(
+        Bucket=base_info.bucket, Key=(base_info / "from").path, Body="data"
+    )
+    remote.link(base_info / "from", base_info / "to")
+
+    assert remote.exists(base_info / "from")
+    assert remote.exists(base_info / "to")
+
+
+@mock_s3
+def test_makedirs_doesnot_try_on_top_level_paths(tmp_dir, dvc, scm):
+    base_info = RemoteS3.path_cls("s3://bucket/")
+    remote = RemoteS3(dvc, {"url": str(base_info)})
+    remote.makedirs(base_info)
+
+
 def _upload_multipart(s3, Bucket, Key):
     mpu = s3.create_multipart_upload(Bucket=Bucket, Key=Key)
     mpu_id = mpu["UploadId"]

tests/unit/remote/test_remote.py

@@ -1,4 +1,6 @@
-from dvc.remote import Remote
+import pytest
+
+from dvc.remote import Remote, RemoteS3, RemoteGS
 
 
 def test_remote_with_checksum_jobs(dvc):
@@ -25,3 +27,15 @@ def test_remote_without_checksum_jobs_default(dvc):
 
     remote = Remote(dvc, name="without_checksum_jobs")
     assert remote.checksum_jobs == remote.CHECKSUM_JOBS
+
+
+@pytest.mark.parametrize("remote_cls", [RemoteGS, RemoteS3])
+def test_makedirs_not_create_for_top_level_path(remote_cls, mocker):
+    url = "{.scheme}://bucket/".format(remote_cls)
+    remote = remote_cls(None, {"url": url})
+    mocked_client = mocker.PropertyMock()
+    # we use remote clients with same name as scheme to interact with remote
+    mocker.patch.object(remote_cls, remote.scheme, mocked_client)
+
+    remote.makedirs(remote.path_info)
+    assert not mocked_client.called