tests: add GCSPath test for copy_into (xfail)

Author: ap--

upath/tests/implementations/test_gcs.py

@@ -1,3 +1,5 @@
+import warnings
+
 import fsspec
 import pytest
 
@@ -8,6 +10,7 @@
 from ..utils import OverrideMeta
 from ..utils import extends_base
 from ..utils import overrides_base
+from ..utils import posixify
 from ..utils import skip_on_windows
 
 
@@ -49,3 +52,56 @@ def test_mkdir_in_empty_bucket(docker_gcs):
         endpoint_url=docker_gcs,
         token="anon",
     ).parent.mkdir(parents=True, exist_ok=True)
+
+
+@skip_on_windows
+@pytest.mark.xfail(reason="gcsfs returns isdir false")
+def test_copy__object_key_collides_with_dir_prefix(docker_gcs, tmp_path):
+    gcs = fsspec.filesystem(
+        "gcs",
+        endpoint_url=docker_gcs,
+        token="anon",
+        use_listings_cache=False,
+    )
+    bucket = "copy_into_collision_bucket"
+    gcs.mkdir(bucket)
+    # gcs.mkdir(bucket + "/src" + "/common_prefix/")
+    # object under common prefix as key
+    gcs.pipe_file(f"{bucket}/src/common_prefix", b"hello world")
+    # store more objects with same prefix
+    gcs.pipe_file(f"{bucket}/src/common_prefix/file1.txt", b"1")
+    gcs.pipe_file(f"{bucket}/src/common_prefix/file2.txt", b"2")
+    gcs.invalidate_cache()
+
+    # make sure the sources have a collision
+    assert gcs.isfile(f"{bucket}/src/common_prefix")
+    assert gcs.isdir(f"{bucket}/src/common_prefix")  # BROKEN in gcsfs
+    assert gcs.isfile(f"{bucket}/src/common_prefix/file1.txt")
+    assert gcs.isfile(f"{bucket}/src/common_prefix/file2.txt")
+    # prepare source and destination
+    src = UPath(f"gs://{bucket}/src", endpoint_url=docker_gcs, token="anon")
+    dst = UPath(tmp_path)
+
+    def on_collision_rename_file(src, dst):
+        warnings.warn(
+            f"{src!s} collides with prefix. Renaming target file object to {dst!s}",
+            UserWarning,
+            stacklevel=3,
+        )
+        return (
+            dst.with_suffix(dst.suffix + ".COLLISION"),
+            dst,
+        )
+
+    # perform copy
+    src.copy_into(dst, on_name_collision=on_collision_rename_file)
+
+    # check results
+    dst_files = sorted(posixify(x.relative_to(tmp_path)) for x in dst.glob("**/*"))
+    assert dst_files == [
+        "src",
+        "src/common_prefix",
+        "src/common_prefix.COLLISION",
+        "src/common_prefix/file1.txt",
+        "src/common_prefix/file2.txt",
+    ]