Propagate storage_options to all backends resolved by GenericFileSystem (#1856)

fsspec/generic.py

@@ -208,7 +208,7 @@ async def _find(self, path, maxdepth=None, withdirs=False, detail=False, **kwarg
         return list(result)
 
     async def _info(self, url, **kwargs):
-        fs = _resolve_fs(url, self.method)
+        fs = _resolve_fs(url, self.method, storage_options=self.st_opts)
         if fs.async_impl:
             out = await fs._info(url, **kwargs)
         else:
@@ -223,7 +223,7 @@ async def _ls(
         detail=True,
         **kwargs,
     ):
-        fs = _resolve_fs(url, self.method)
+        fs = _resolve_fs(url, self.method, storage_options=self.st_opts)
         if fs.async_impl:
             out = await fs._ls(url, detail=True, **kwargs)
         else:
@@ -241,7 +241,7 @@ async def _cat_file(
         url,
         **kwargs,
     ):
-        fs = _resolve_fs(url, self.method)
+        fs = _resolve_fs(url, self.method, storage_options=self.st_opts)
         if fs.async_impl:
             return await fs._cat_file(url, **kwargs)
         else:
@@ -263,7 +263,7 @@ async def _rm(self, url, **kwargs):
         urls = url
         if isinstance(urls, str):
             urls = [urls]
-        fs = _resolve_fs(urls[0], self.method)
+        fs = _resolve_fs(urls[0], self.method, storage_options=self.st_opts)
         if fs.async_impl:
             await fs._rm(urls, **kwargs)
         else:
@@ -293,8 +293,8 @@ async def _cp_file(
         tempdir: str | None = None,
         **kwargs,
     ):
-        fs = _resolve_fs(url, self.method)
-        fs2 = _resolve_fs(url2, self.method)
+        fs = _resolve_fs(url, self.method, storage_options=self.st_opts)
+        fs2 = _resolve_fs(url2, self.method, storage_options=self.st_opts)
         if fs is fs2:
             # pure remote
             if fs.async_impl:
@@ -304,7 +304,7 @@ async def _cp_file(
         await copy_file_op(fs, [url], fs2, [url2], tempdir, 1, on_error="raise")
 
     async def _make_many_dirs(self, urls, exist_ok=True):
-        fs = _resolve_fs(urls[0], self.method)
+        fs = _resolve_fs(urls[0], self.method, storage_options=self.st_opts)
         if fs.async_impl:
             coros = [fs._makedirs(u, exist_ok=exist_ok) for u in urls]
             await _run_coros_in_chunks(coros)
@@ -332,8 +332,8 @@ async def _copy(
         path1 = [path1] if isinstance(path1, str) else path1
         path2 = [path2] if isinstance(path2, str) else path2
 
-        fs = _resolve_fs(path1, self.method)
-        fs2 = _resolve_fs(path2, self.method)
+        fs = _resolve_fs(path1, self.method, storage_options=self.st_opts)
+        fs2 = _resolve_fs(path2, self.method, storage_options=self.st_opts)
 
         if fs is fs2:
             if fs.async_impl:

fsspec/tests/test_generic.py

@@ -1,6 +1,9 @@
 import pytest
 
 import fsspec
+from fsspec.generic import rsync
+from fsspec.implementations.memory import MemoryFileSystem
+from fsspec.registry import _registry, register_implementation
 from fsspec.tests.conftest import data, server  # noqa: F401
 
 
@@ -116,3 +119,57 @@ def test_rsync(tmpdir, m):
         allfiles[f"file://{pos_tmpdir}/deep/path/afile"]
         == allfiles2[f"file://{pos_tmpdir}/deep/path/afile"]
     )
+
+
+class _OptionsRequiredFS(MemoryFileSystem):
+    """Backend that raises unless its storage_options reached the constructor."""
+
+    protocol = "optsrequired"
+
+    def __init__(self, *args, token=None, **kwargs):
+        if token is None:
+            raise ValueError("storage_options were not propagated")
+        super().__init__(*args, **kwargs)
+
+    @classmethod
+    def _strip_protocol(cls, path):
+        if isinstance(path, str):
+            path = path.removeprefix(f"{cls.protocol}://")
+        return MemoryFileSystem._strip_protocol(path)
+
+
+@pytest.fixture()
+def opts_fs(m):
+    register_implementation(
+        _OptionsRequiredFS.protocol, _OptionsRequiredFS, clobber=True
+    )
+    m.pipe_file("/afile", b"hello")
+    m.makedirs("/adir", exist_ok=True)
+    m.pipe_file("/adir/nested", b"world")
+    yield fsspec.filesystem(
+        "generic",
+        default_method="options",
+        storage_options={_OptionsRequiredFS.protocol: {"token": "set"}},
+    )
+    _registry.pop(_OptionsRequiredFS.protocol, None)
+    _OptionsRequiredFS.clear_instance_cache()
+
+
+def test_storage_options_propagated_to_backend(opts_fs):
+    opts_fs.info("optsrequired:///afile")
+    opts_fs.ls("optsrequired:///adir")
+    opts_fs.cat_file("optsrequired:///afile")
+    opts_fs.isdir("optsrequired:///adir")
+    opts_fs.find("optsrequired:///adir")
+
+
+def test_storage_options_propagated_cross_protocol(opts_fs, tmpdir):
+    rsync(
+        "optsrequired:///adir",
+        f"file://{tmpdir}",
+        inst_kwargs={
+            "default_method": "options",
+            "storage_options": {_OptionsRequiredFS.protocol: {"token": "set"}},
+        },
+    )
+    assert (tmpdir / "nested").read_binary() == b"world"