fix: Ensure kwargs passed via expand_path (#1911)

---------

Co-authored-by: Martin Durant <[email protected]>

Author: ReubenFrankel

fsspec/spec.py

@@ -892,7 +892,7 @@ def cat(self, path, recursive=False, on_error="raise", **kwargs):
         dict of {path: contents} if there are multiple paths
         or the path has been otherwise expanded
         """
-        paths = self.expand_path(path, recursive=recursive)
+        paths = self.expand_path(path, recursive=recursive, **kwargs)
         if (
             len(paths) > 1
             or isinstance(path, list)
@@ -972,7 +972,9 @@ def get(
             )
 
             source_is_str = isinstance(rpath, str)
-            rpaths = self.expand_path(rpath, recursive=recursive, maxdepth=maxdepth)
+            rpaths = self.expand_path(
+                rpath, recursive=recursive, maxdepth=maxdepth, **kwargs
+            )
             if source_is_str and (not recursive or maxdepth is not None):
                 # Non-recursive glob does not copy directories
                 rpaths = [p for p in rpaths if not (trailing_sep(p) or self.isdir(p))]
@@ -1060,7 +1062,9 @@ def put(
             if source_is_str:
                 lpath = make_path_posix(lpath)
             fs = LocalFileSystem()
-            lpaths = fs.expand_path(lpath, recursive=recursive, maxdepth=maxdepth)
+            lpaths = fs.expand_path(
+                lpath, recursive=recursive, maxdepth=maxdepth, **kwargs
+            )
             if source_is_str and (not recursive or maxdepth is not None):
                 # Non-recursive glob does not copy directories
                 lpaths = [p for p in lpaths if not (trailing_sep(p) or fs.isdir(p))]
@@ -1131,7 +1135,9 @@ def copy(
             from .implementations.local import trailing_sep
 
             source_is_str = isinstance(path1, str)
-            paths1 = self.expand_path(path1, recursive=recursive, maxdepth=maxdepth)
+            paths1 = self.expand_path(
+                path1, recursive=recursive, maxdepth=maxdepth, **kwargs
+            )
             if source_is_str and (not recursive or maxdepth is not None):
                 # Non-recursive glob does not copy directories
                 paths1 = [p for p in paths1 if not (trailing_sep(p) or self.isdir(p))]
@@ -1172,7 +1178,7 @@ def expand_path(self, path, recursive=False, maxdepth=None, **kwargs):
             raise ValueError("maxdepth must be at least 1")
 
         if isinstance(path, (str, os.PathLike)):
-            out = self.expand_path([path], recursive, maxdepth)
+            out = self.expand_path([path], recursive, maxdepth, **kwargs)
         else:
             out = set()
             path = [self._strip_protocol(p) for p in path]

fsspec/tests/test_spec.py

@@ -1375,3 +1375,41 @@ def test_glob_posix_rules(path, expected, glob_fs):
     detailed_output = glob_fs.glob(path=f"mock://{path}", detail=True)
     for name, info in _clean_paths(detailed_output).items():
         assert info == glob_fs[name]
+
+
+@pytest.fixture(scope="function")
+def tmpfs(tmpdir):
+    get_files(tmpdir)
+
+    class CustomLocalFS(LocalFileSystem):
+        def ls(self, *args, **kwargs):
+            files = super().ls(*args, **kwargs)
+            limit = kwargs.pop("limit", None)
+
+            return files[:limit] if limit else files
+
+    return CustomLocalFS(auto_mkdir=True)
+
+
+def test_cat_wildcard_path_passthrough_kwargs_to_ls(tmpfs, tmpdir):
+    file_contents = tmpfs.cat(tmpdir / "*", limit=2)
+    assert len(file_contents) == 2
+
+
+def test_get_wildcard_path_passthrough_kwargs_to_ls(tmpfs, tmpdir):
+    dest = tmpdir / "dest"
+    tmpfs.get(tmpdir / "*", str(dest), limit=2)
+
+    assert len(dest.listdir()) == 2
+
+
+def test_copy_wildcard_path_passthrough_kwargs_to_ls(tmpfs, tmpdir):
+    dest = tmpdir / "dest"
+    tmpfs.copy(tmpdir / "*", str(dest), limit=2)
+
+    assert len(dest.listdir()) == 2
+
+
+def test_expand_path_wildcard_path_passthrough_kwargs_to_ls(tmpfs, tmpdir):
+    expanded_paths = tmpfs.expand_path(tmpdir / "*", limit=2)
+    assert len(expanded_paths) == 2