Test the fsspec support against a real S3 endpoint

memory:// cannot see the class of bug that lives on a real backend, and has
now hidden two: aget_chunk awaiting an async filesystem's coroutine (fine on
memory://, which is not async, and broken on every chunk against s3fs), and
a cache stamp that degraded to size-only because memory:// exposes no mtime.

tests/test_fsspec_s3.py runs moto in-process (ThreadedMotoServer on a free
port, so xdist workers do not collide) with s3fs in front of it: a real S3
protocol, real range requests, a real async backend, and still offline -- no
credentials, no network, so no `network` marker.  Eight tests in ~3 s.
Reverting the aget_chunk fix fails exactly the two that cover it.

moto[server] and s3fs go into the test group as well as dev, so this runs on
push rather than only when someone remembers.  That is a heavier dependency
tree on every job; if it ever churns badly enough to break installs, moving
both to dev-only and running them nightly is the fallback.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Author: FrancescAlted

pyproject.toml

@@ -77,6 +77,7 @@ dev = [
     "matplotlib",
     "pandas",
     "plotly",
+    "moto[server]",
     "pre-commit",
     "pyarrow",
     "ruff",
@@ -87,10 +88,15 @@ dev = [
 test = [
     "pytest",
     # tests/test_fsspec.py importorskips fsspec, so without this the whole fsspec
-    # feature silently skips in CI.  Protocol backends (s3fs, gcsfs...) stay out:
-    # the tests run on memory:// and a local zip, which is also the configuration
-    # most users installing [fsspec] are in.
+    # feature silently skips in CI.  memory:// covers the protocol-generic paths,
+    # and is also the configuration most users installing [fsspec] are in.
     "fsspec; platform_machine != 'wasm32'",
+    # tests/test_fsspec_s3.py needs a real S3 endpoint (moto, served locally, so
+    # still offline) and a real *async* backend (s3fs).  memory:// is neither, and
+    # cannot see the class of bug that lives there: awaiting an async filesystem's
+    # own coroutine fails on s3fs and passes silently on memory://.  Runs in ~3 s.
+    "moto[server]; platform_machine != 'wasm32'",
+    "s3fs; platform_machine != 'wasm32'",
     # pytest.ini defaults to `-n auto`; where xdist is absent (wasm32, or a
     # bare `pip install pytest`) the root conftest.py degrades it to a serial run
     "pytest-xdist; platform_machine != 'wasm32'",

tests/test_fsspec_s3.py

@@ -0,0 +1,117 @@
+#######################################################################
+# Copyright (c) 2019-present, Blosc Development Team <blosc@blosc.org>
+# All rights reserved.
+#
+# This source code is licensed under a BSD-style license (found in the
+# LICENSE file in the root directory of this source tree)
+#######################################################################
+
+"""fsspec reads against a real S3 endpoint, served locally by moto.
+
+Everything else about the fsspec support is tested over ``memory://``, which is
+protocol-generic and needs no dependencies.  Two things it structurally cannot
+cover, both of which have already hidden a bug:
+
+- it is not an *async* backend, so ``aget_chunk`` always took its blocking
+  fallback there, while against s3fs it raised "got Future attached to a
+  different loop" on every chunk;
+- it is poorer in metadata than any real store (no mtime), which let a
+  size-only cache stamp serve a stale chunk cache.
+
+These run offline -- moto is a local server, no credentials, no network -- so
+they are not marked ``network``.
+"""
+
+import asyncio
+
+import numpy as np
+import pytest
+
+import blosc2
+
+pytest.importorskip("s3fs")
+pytest.importorskip("moto")
+fsspec = pytest.importorskip("fsspec")
+
+BUCKET = "blosc2-test"
+
+
+@pytest.fixture(scope="module")
+def s3_endpoint():
+    """A local S3 server, and fsspec configured to reach it."""
+    import fsspec.config
+    from moto.server import ThreadedMotoServer
+
+    server = ThreadedMotoServer(ip_address="127.0.0.1", port=0, verbose=False)
+    server.start()
+    host, port = server.get_host_and_port()
+    endpoint = f"http://{host}:{port}"
+
+    # blosc2.open() has no storage_options passthrough, so the endpoint and the
+    # dummy credentials go through fsspec's own per-protocol defaults
+    previous = fsspec.config.conf.get("s3")
+    fsspec.config.conf["s3"] = {
+        "endpoint_url": endpoint,
+        "key": "testing",
+        "secret": "testing",
+        # Not us-east-1: creating a bucket there must carry no location
+        # constraint, and s3fs sends one whenever it knows the region
+        "client_kwargs": {"region_name": "eu-west-1"},
+    }
+    fsspec.filesystem("s3", **fsspec.config.conf["s3"]).mkdir(BUCKET)
+    yield endpoint
+
+    fsspec.config.conf.pop("s3", None)
+    if previous is not None:
+        fsspec.config.conf["s3"] = previous
+    server.stop()
+
+
+@pytest.fixture(scope="module")
+def stored(s3_endpoint):
+    """A 10-chunk array in the bucket, plus the array it was made from."""
+    a = blosc2.arange(0, 1000, dtype=np.int32, chunks=(100,))
+    urlpath = f"s3://{BUCKET}/ds.b2nd"
+    a.save(urlpath)
+    return urlpath, a
+
+
+def test_save_and_open_whole(stored):
+    urlpath, a = stored
+    assert np.array_equal(blosc2.open(urlpath)[:], a[:])
+
+
+def test_cache_storage(stored, tmp_path):
+    urlpath, a = stored
+    b = blosc2.open(urlpath, cache_storage=tmp_path, mmap_mode="r")
+    assert np.array_equal(b[:], a[:])
+
+
+def test_lazy_range_reads(stored):
+    urlpath, a = stored
+    p = blosc2.open(urlpath, lazy=True)
+    assert np.array_equal(p[150:250], a[150:250])
+    assert np.array_equal(p[:], a[:])
+
+
+@pytest.mark.parametrize("max_concurrency", [1, 8])
+def test_lazy_concurrency(stored, max_concurrency):
+    urlpath, a = stored
+    p = blosc2.open(urlpath, lazy=True, max_concurrency=max_concurrency)
+    assert np.array_equal(p[:], a[:])
+
+
+@pytest.mark.parametrize("max_concurrency", [1, 8])
+def test_afetch_on_an_async_backend(stored, max_concurrency):
+    # The regression this file exists for: s3fs runs its coroutines on a private
+    # event loop, so awaiting one from the caller's loop fails outright
+    urlpath, a = stored
+    p = blosc2.open(urlpath, lazy=True)
+    cache = asyncio.run(p.afetch(slice(150, 250), max_concurrency=max_concurrency))
+    assert np.array_equal(cache[150:250], a[150:250])
+
+
+def test_lazy_expression(stored):
+    urlpath, a = stored
+    p = blosc2.open(urlpath, lazy=True)
+    assert np.array_equal((p * 2)[150:250], a[150:250] * 2)