Use removeprefix rather than replace to avoid separator deletion (#2778)

* Use removeprefix rather than replace to avoid separator deletion

* Test list behavior with empty paths

---------

Co-authored-by: Joe Hamman <[email protected]>
Co-authored-by: Davis Bennett <[email protected]>

Author: 3 people

changes/2778.bugfix.rst

@@ -0,0 +1 @@
+Use removeprefix rather than replace when removing filename prefixes in `FsspecStore.list`

src/zarr/storage/_fsspec.py

@@ -341,7 +341,7 @@ async def set_partial_values(
     async def list(self) -> AsyncIterator[str]:
         # docstring inherited
         allfiles = await self.fs._find(self.path, detail=False, withdirs=False)
-        for onefile in (a.replace(self.path + "/", "") for a in allfiles):
+        for onefile in (a.removeprefix(self.path + "/") for a in allfiles):
             yield onefile
 
     async def list_dir(self, prefix: str) -> AsyncIterator[str]:

src/zarr/testing/store.py

@@ -400,6 +400,37 @@ async def test_list_prefix(self, store: S) -> None:
             expected = tuple(sorted(expected))
             assert observed == expected
 
+    async def test_list_empty_path(self, store: S) -> None:
+        """
+        Verify that list and list_prefix work correctly when path is an empty string,
+        i.e. no unwanted replacement occurs.
+        """
+        data = self.buffer_cls.from_bytes(b"")
+        store_dict = {
+            "foo/bar/zarr.json": data,
+            "foo/bar/c/1": data,
+            "foo/baz/c/0": data,
+        }
+        await store._set_many(store_dict.items())
+
+        # Test list()
+        observed_list = await _collect_aiterator(store.list())
+        observed_list_sorted = sorted(observed_list)
+        expected_list_sorted = sorted(store_dict.keys())
+        assert observed_list_sorted == expected_list_sorted
+
+        # Test list_prefix() with an empty prefix
+        observed_prefix_empty = await _collect_aiterator(store.list_prefix(""))
+        observed_prefix_empty_sorted = sorted(observed_prefix_empty)
+        expected_prefix_empty_sorted = sorted(store_dict.keys())
+        assert observed_prefix_empty_sorted == expected_prefix_empty_sorted
+
+        # Test list_prefix() with a non-empty prefix
+        observed_prefix = await _collect_aiterator(store.list_prefix("foo/bar/"))
+        observed_prefix_sorted = sorted(observed_prefix)
+        expected_prefix_sorted = sorted(k for k in store_dict if k.startswith("foo/bar/"))
+        assert observed_prefix_sorted == expected_prefix_sorted
+
     async def test_list_dir(self, store: S) -> None:
         root = "foo"
         store_dict = {