Repository: dask/dask (python)

Issue:
Failing dask/bytes/tests/test_local.py with fsspec master
cc @martindurant

```
========================================================================= FAILURES =========================================================================
______________________________________________________________ test_urlpath_inference_errors _______________________________________________________________

    def test_urlpath_inference_errors():
        # Empty list
        with pytest.raises(ValueError, match="empty"):
            get_fs_token_paths([])

        # Protocols differ
        with pytest.raises(ValueError, match="the same protocol"):
>           get_fs_token_paths(["s3://test/path.csv", "/other/path.csv"])

dask/bytes/tests/test_local.py:74:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

urlpath = ['s3://test/path.csv', '/other/path.csv'], mode = 'rb', num = 1, name_function = None, storage_options = None, protocol = None, expand = True

    def get_fs_token_paths(
        urlpath,
        mode="rb",
        num=1,
        name_function=None,
        storage_options=None,
        protocol=None,
        expand=True,
    ):
        """Filesystem, deterministic token, and paths from a urlpath and options.

        Parameters
        ----------
        urlpath: string or iterable
            Absolute or relative filepath, URL (may include protocols like
            ``s3://``), or globstring pointing to data.
        mode: str, optional
            Mode in which to open files.
        num: int, optional
            If opening in writing mode, number of files we expect to create.
        name_function: callable, optional
            If opening in writing mode, this callable is used to generate path
            names. Names are generated for each partition by
            ``urlpath.replace('*', name_function(partition_index))``.
        storage_options: dict, optional
            Additional keywords to pass to the filesystem class.
        protocol: str or None
            To override the protocol specifier in the URL
        expand: bool
            Expand string paths for writing, assuming the path is a directory
        """
>       chain = _un_chain(urlpath, storage_options or {})

../../Envs/dask-dev/lib/python3.7/site-packages/fsspec/core.py:484:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

path = ['s3://test/path.csv', '/other/path.csv'], kwargs = ({}, {})

    def _un_chain(path, kwargs):
        if isinstance(path, (tuple, list)):
            bits = [_un_chain(p, kwargs) for p in path]
            out = []
            for pbit in zip(*bits):
                paths, protocols, kwargs = zip(*pbit)
                if len(set(protocols)) > 1:
>                   raise ValueError("Protocol mismatch in URL chain")
E                   ValueError: Protocol mismatch in URL chain

../../Envs/dask-dev/lib/python3.7/site-packages/fsspec/core.py:258: ValueError

During handling of the above exception, another exception occurred:

    def test_urlpath_inference_errors():
        # Empty list
        with pytest.raises(ValueError, match="empty"):
            get_fs_token_paths([])

        # Protocols differ
        with pytest.raises(ValueError, match="the same protocol"):
>           get_fs_token_paths(["s3://test/path.csv", "/other/path.csv"])
E           AssertionError: Pattern 'the same protocol' does not match 'Protocol mismatch in URL chain'

dask/bytes/tests/test_local.py:74: AssertionError
_____________________________________________________________________ test_with_paths ______________________________________________________________________

    @pytest.mark.skipif(sys.platform == "win32", reason="pathlib and moto clash on windows")
    def test_with_paths():
        with filetexts(files, mode="b"):
            url = pathlib.Path("./.test.accounts.*")
>           sample, values = read_bytes(url, blocksize=None)

dask/bytes/tests/test_local.py:215:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
dask/bytes/core.py:93: in read_bytes
    fs, fs_token, paths = get_fs_token_paths(urlpath, mode="rb", storage_options=kwargs)
../../Envs/dask-dev/lib/python3.7/site-packages/fsspec/core.py:484: in get_fs_token_paths
    chain = _un_chain(urlpath, storage_options or {})
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

path = PosixPath('.test.accounts.*'), kwargs = {}

    def _un_chain(path, kwargs):
        if isinstance(path, (tuple, list)):
            bits = [_un_chain(p, kwargs) for p in path]
            out = []
            for pbit in zip(*bits):
                paths, protocols, kwargs = zip(*pbit)
                if len(set(protocols)) > 1:
                    raise ValueError("Protocol mismatch in URL chain")
                if len(set(paths)) == 1:
                    paths = paths[0]
                else:
                    paths = list(paths)
                out.append([paths, protocols[0], kwargs[0]])
            return out
        x = re.compile(".*[^a-z]+.*")  # test for non protocol-like single word
        bits = (
            [p if "://" in p or x.match(p) else p + "://" for p in path.split("::")]
>           if "::" in path
            else [path]
        )
E       TypeError: argument of type 'PosixPath' is not iterable

../../Envs/dask-dev/lib/python3.7/site-packages/fsspec/core.py:268: TypeError
================================================================ slowest 10 test durations =================================================================

(0.00 durations hidden.  Use -vv to show these durations.)
==================================================================== 2 failed in 0.49s =====================================================================
```

Pull request description:
Update expected error msg for protocol difference in fsspec
Also remove check for error is a set of filenames is passed in as this
is now supported upstream.

Fixes #6327 

- [x] Tests added / passed
- [x] Passes `black dask` / `flake8 dask`

Open question around whether a `set` of filenames is OK.  I've removed the test for that for now, happy to put it back in and add some checks on the `dask` side of things if that's what we settle on.

Patch:
diff --git a/dask/bytes/core.py b/dask/bytes/core.py
index 76b5a3b31..e6b203ed1 100644
--- a/dask/bytes/core.py
+++ b/dask/bytes/core.py
@@ -1,3 +1,4 @@
+import os
 import copy
 
 from fsspec.core import (  # noqa: F401
@@ -90,6 +91,9 @@ def read_bytes(
         represented in the corresponding block.
 
     """
+    if not isinstance(urlpath, (str, list, tuple, os.PathLike)):
+        raise TypeError("Path should be a string, os.PathLike, list or tuple")
+
     fs, fs_token, paths = get_fs_token_paths(urlpath, mode="rb", storage_options=kwargs)
 
     if len(paths) == 0:


Test patch:
diff --git a/dask/bytes/tests/test_local.py b/dask/bytes/tests/test_local.py
index 2eed2da74..c9047d0cd 100644
--- a/dask/bytes/tests/test_local.py
+++ b/dask/bytes/tests/test_local.py
@@ -12,7 +12,7 @@ from dask import compute
 from dask.utils import filetexts
 from fsspec.implementations.local import LocalFileSystem
 from fsspec.compression import compr
-from dask.bytes.core import read_bytes, open_files, get_fs_token_paths
+from dask.bytes.core import read_bytes, open_files
 from dask.bytes.utils import compress
 
 compute = partial(compute, scheduler="sync")
@@ -44,47 +44,11 @@ def to_uri(path):
     return pathlib.Path(os.path.abspath(path)).as_uri()
 
 
-def test_urlpath_inference_strips_protocol(tmpdir):
-    tmpdir = str(tmpdir)
-    paths = [os.path.join(tmpdir, "test.%02d.csv" % i) for i in range(20)]
-
-    for path in paths:
-        with open(path, "wb") as f:
-            f.write(b"1,2,3\n" * 10)
-
-    # globstring
-    protocol = "file:///" if sys.platform == "win32" else "file://"
-    urlpath = protocol + os.path.join(tmpdir, "test.*.csv")
-    _, _, paths2 = get_fs_token_paths(urlpath)
-    assert "file:" not in paths2[0]
-    assert paths2[0].endswith("/test.00.csv")
-
-    # list of paths
-    _, _, paths3 = get_fs_token_paths([protocol + p for p in paths])
-    assert paths2 == paths3
-
-
-def test_urlpath_inference_errors():
-    # Empty list
-    with pytest.raises(ValueError, match="empty"):
-        get_fs_token_paths([])
-
-    # Protocols differ
-    with pytest.raises(ValueError, match="the same protocol"):
-        get_fs_token_paths(["s3://test/path.csv", "/other/path.csv"])
-
-    # Options differ
-    with pytest.raises(ValueError, match="the same file-system options"):
-        get_fs_token_paths(
-            [
-                "ftp://myuser@node.com/test/path.csv",
-                "ftp://otheruser@node.com/other/path.csv",
-            ]
-        )
+def test_unordered_urlpath_errors():
 
-    # Unknown type
+    # Unordered urlpath argument
     with pytest.raises(TypeError):
-        get_fs_token_paths(
+        read_bytes(
             {
                 "sets/are.csv",
                 "unordered/so/they.csv",
@@ -94,39 +58,6 @@ def test_urlpath_inference_errors():
         )
 
 
-def test_urlpath_expand_read():
-    """Make sure * is expanded in file paths when reading."""
-    # when reading, globs should be expanded to read files by mask
-    with filetexts(csv_files, mode="b"):
-        _, _, paths = get_fs_token_paths(".*.csv")
-        assert len(paths) == 2
-        _, _, paths = get_fs_token_paths([".*.csv"])
-        assert len(paths) == 2
-
-
-def test_recursive_glob_expand():
-    """Make sure * is expanded in file paths when reading."""
-    with filetexts(
-        {"sub1/afile.csv": b"", "sub1/sub2/another.csv": b"", "sub1/twofile.csv": b""},
-        mode="b",
-    ):
-        _, _, paths = get_fs_token_paths(os.path.abspath("**/*.csv"))
-        assert len(paths) == 3
-
-
-def test_urlpath_expand_write():
-    """Make sure * is expanded in file paths when writing."""
-    _, _, paths = get_fs_token_paths("prefix-*.csv", mode="wb", num=2)
-    assert [p.endswith(pa) for p, pa in zip(paths, ["prefix-0.csv", "prefix-1.csv"])]
-    _, _, paths = get_fs_token_paths(["prefix-*.csv"], mode="wb", num=2)
-    assert [p.endswith(pa) for p, pa in zip(paths, ["prefix-0.csv", "prefix-1.csv"])]
-    # we can read with multiple masks, but not write
-    with pytest.raises(ValueError):
-        _, _, paths = get_fs_token_paths(
-            ["prefix1-*.csv", "prefix2-*.csv"], mode="wb", num=2
-        )
-
-
 def test_read_bytes():
     with filetexts(files, mode="b"):
         sample, values = read_bytes(".test.accounts.*")