diff --git a/changes/4192.feature.md b/changes/4192.feature.md new file mode 100644 index 0000000000..9ecd24aa17 --- /dev/null +++ b/changes/4192.feature.md @@ -0,0 +1,7 @@ +Added core support for URL pipelines (https://github.com/jbms/url-pipeline), `|`-chained +URLs such as `s3://bucket/data.zip|zip:|zarr3:`. Parse one with +`zarr.storage.URLPipeline.from_url` and pass it to any function that accepts a `StoreLike`; +plain strings are never read as pipelines. A trailing `zarr:`, `zarr2:` or `zarr3:` segment +gives the node path and selects (or detects) the format. Packages provide adapters for other +schemes by subclassing `zarr.abc.url_pipeline.URLPipelineAdapter` and registering it under the +`zarr.url_adapters` entry-point group or with `zarr.registry.register_url_adapter`. diff --git a/docs/api/zarr/abc/index.md b/docs/api/zarr/abc/index.md index 7e15cb2a51..d155f60a5a 100644 --- a/docs/api/zarr/abc/index.md +++ b/docs/api/zarr/abc/index.md @@ -11,3 +11,4 @@ Abstract base classes for extending Zarr-Python. - **[zarr.abc.metadata](./metadata.md)** - Creating metadata classes compatible with the Zarr API - **[zarr.abc.numcodec](./numcodec.md)** - Protocols and classes for modeling codec interface used by numcodecs - **[zarr.abc.store](./store.md)** - ABC for implementing Zarr stores and managing getting and setting bytes in a store +- **[zarr.abc.url_pipeline](./url_pipeline.md)** - ABC for implementing [URL pipeline](https://github.com/jbms/url-pipeline) adapters diff --git a/docs/api/zarr/abc/url_pipeline.md b/docs/api/zarr/abc/url_pipeline.md new file mode 100644 index 0000000000..2606d4d5d4 --- /dev/null +++ b/docs/api/zarr/abc/url_pipeline.md @@ -0,0 +1,5 @@ +--- +title: url_pipeline +--- + +::: zarr.abc.url_pipeline diff --git a/docs/user-guide/extending.md b/docs/user-guide/extending.md index 507afedea7..bf0eb442c9 100644 --- a/docs/user-guide/extending.md +++ b/docs/user-guide/extending.md @@ -85,6 +85,72 @@ implementation. Custom stores can be created by implementing the [`zarr.abc.store.Store`][] interface. See [developing custom stores](storage.md#developing-custom-stores) for more information. +## Custom URL pipeline adapters + +In a [URL pipeline](storage.md#user-guide-url-pipelines) such as +`s3://bucket/data.zip|zip:|zarr3:`, each sub-URL after the first names an *adapter* +that reinterprets everything to its left. A package provides an adapter by subclassing +[`zarr.abc.url_pipeline.URLPipelineAdapter`][] and implementing one classmethod, which +returns an open store and the path addressed inside it as a +[`StorePath`][zarr.storage.StorePath]: + +```python test="true" session="url-adapter" +from zarr.abc.url_pipeline import URLPipelineAdapter +from zarr.storage import PipelineContext, PipelineSegment, StorePath, WrapperStore + + +class MyAdapter(URLPipelineAdapter): + @classmethod + async def open_pipeline_segment( + cls, segment: PipelineSegment, context: PipelineContext + ) -> StorePath: + # A wrapper adapter opens the resource to its left and wraps it, joining the + # preceding residual path with this segment's own path. + preceding = await context.resolve_preceding() + store = WrapperStore(preceding.store) + path = "/".join(part.strip("/") for part in (preceding.path, segment.body) if part) + return StorePath(store, path) +``` + +To resolve the preceding pipeline with another mode or options, use a derived context: +`await dataclasses.replace(context, mode="r").resolve_preceding()`. Wrapper adapters +compose: `memory://x|mypackage.myscheme:a|mypackage.myscheme:b` resolves to `a/b`. + +```python test="true" session="url-adapter" +import zarr +from zarr.registry import register_url_adapter +from zarr.storage import URLPipeline + +register_url_adapter("mypackage.myscheme", MyAdapter) +url = "memory://x|mypackage.myscheme:a|mypackage.myscheme:b" +# keep a reference: in-memory data lives only while a store refers to it +created = zarr.create_group(URLPipeline.from_url(url)) +group = zarr.open_group(URLPipeline.from_url(url), mode="r") +assert group.path == "a/b" +``` + +Register the class under the `zarr.url_adapters` entry-point group with the scheme as +the entry-point name, vendor-prefixed for nonstandard schemes. Adapters are imported +only when a pipeline naming their scheme is resolved. + +```toml +[project.entry-points."zarr.url_adapters"] +"mypackage.myscheme" = "mypackage.zarr_adapter:MyAdapter" +``` + +An adapter can also be a *root* scheme (`mypackage.myscheme://org/repo`), with an empty +`context.preceding`. Installing a root adapter for a scheme fsspec also handles takes +that scheme over at the root of every pipeline. The schemes zarr handles itself +(`file`, `memory`, `zarr`, `zarr2`, `zarr3`) cannot be registered, and entry points +with those names are ignored. + +Every adapter sees all of the caller's `storage_options`, including the root's +credentials: consume only keys namespaced for your adapter, strip them before resolving +the rest of the pipeline, and never log the options. The returned store must honor +`context.read_only`, and `open_pipeline_segment` must not block or call zarr's +synchronous API. See [`PipelineContext`][zarr.storage.PipelineContext] for the full +contract. + ## Custom array buffers Zarr-python provides control over where and how arrays are stored in memory through diff --git a/docs/user-guide/storage.md b/docs/user-guide/storage.md index 88ee2f1f32..13149da198 100644 --- a/docs/user-guide/storage.md +++ b/docs/user-guide/storage.md @@ -100,9 +100,56 @@ print(group) group = zarr.open_group(UPath('s3://noaa-nwm-retro-v2-zarr-pds', anon=True), mode='r') ``` +- a [`URLPipeline`][zarr.storage.URLPipeline], the parsed form of a + [URL pipeline](#user-guide-url-pipelines) such as `s3://bucket/data.zip|zip:`, which + is resolved through registered adapters. + - a [`Store`][zarr.abc.store.Store] or [`StorePath`][zarr.storage.StorePath] - see explicit store creation below. +## URL Pipelines {#user-guide-url-pipelines} + +Zarr supports [URL pipelines](https://github.com/jbms/url-pipeline): `|`-chained URLs +read left to right. The first sub-URL locates a resource with a conventional URL; each +later sub-URL names an *adapter* that reinterprets everything to its left (e.g. +`s3://bucket/data.zip|zip:|zarr3:`). Adapters are provided by packages; see +[custom URL pipeline adapters](extending.md#custom-url-pipeline-adapters). + +Parse a pipeline with [`URLPipeline.from_url`][zarr.storage.URLPipeline.from_url] and +pass it to any function that accepts a `StoreLike`. At present, this is the only way a string is +read as a pipeline: a plain string store specification currently maintains its existing meaning, so a +local path containing `|` still names a directory. + +```python exec="true" session="storage" source="above" result="ansi" +from zarr.storage import URLPipeline + +pipeline = URLPipeline.from_url("memory://pipeline-demo|zarr2:") +group = zarr.open_group(pipeline, mode="w") +print(zarr.open_group(pipeline, mode="r").metadata.zarr_format) +``` + +A trailing `zarr2:` or `zarr3:` *format segment* selects the zarr format, +and its body (`zarr3:path/to/node`) is the node path; `zarr:` gives a node path and +detects the format. A conflicting `zarr_format` argument raises `ValueError`, including +the default `zarr_format=3` of `zarr.create_array`, `Group.open`, `Group.from_store` and +`Array.open`. + +The root sub-URL must carry a scheme, so a local path is spelled as an absolute `file:` +URL (`file:/data/x.zarr`, with at most a `localhost` authority). Every `memory:` root +addresses one managed in-memory store, and its body is a path within it (`memory:a` ≡ +`memory:/a` ≡ `memory://a`); data lives only while some store object refers to it. +Neither root accepts `.` or `..` path components, and drive letters are only special on +Windows. No percent-escape is decoded anywhere in a pipeline. Other roots, such as +`s3://...`, are opened exactly as the same plain string would be. + +`storage_options` configure the root sub-URL, and every adapter in the pipeline can +see them. When a pipeline may come from an untrusted source, keep credentials out of +`storage_options` unless you trust every installed adapter, and don't accept `memory:` +roots in a process that serves several users: the in-memory store is process-wide. + +For complex store configurations, explicit store creation, as described below, may be +required. + ## Explicit Store Creation In some cases, it may be helpful to create a store instance directly. Zarr-Python offers diff --git a/mkdocs.yml b/mkdocs.yml index 4d06701a87..58b5d3344a 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -43,6 +43,7 @@ nav: - ' zarr.abc.metadata': api/zarr/abc/metadata.md - ' zarr.abc.numcodec': api/zarr/abc/numcodec.md - ' zarr.abc.store': api/zarr/abc/store.md + - ' zarr.abc.url_pipeline': api/zarr/abc/url_pipeline.md - ' zarr.api': - api/zarr/api/index.md - ' zarr.api.asynchronous': api/zarr/api/asynchronous.md diff --git a/src/zarr/abc/url_pipeline.py b/src/zarr/abc/url_pipeline.py new file mode 100644 index 0000000000..d00db407e8 --- /dev/null +++ b/src/zarr/abc/url_pipeline.py @@ -0,0 +1,55 @@ +""" +Abstract base class for URL pipeline adapters +(https://github.com/jbms/url-pipeline). See +[`URLPipelineAdapter`][zarr.abc.url_pipeline.URLPipelineAdapter]. +""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from zarr.storage import PipelineContext, PipelineSegment, StorePath + +__all__ = ["URLPipelineAdapter"] + + +class URLPipelineAdapter(ABC): + """ + Handler for one URL pipeline scheme. + + In a pipeline such as `s3://bucket/data.zip|zip:inner|zarr3:`, each sub-URL + after the first names an adapter that reinterprets everything to its left. + Subclasses implement + [`open_pipeline_segment`][zarr.abc.url_pipeline.URLPipelineAdapter.open_pipeline_segment] + and are registered under the `zarr.url_adapters` entry-point group, with + the URL scheme as the entry-point name: + + [project.entry-points."zarr.url_adapters"] + "mypackage.myscheme" = "mypackage.zarr_adapter:MyAdapter" + + Nonstandard schemes should be vendor-prefixed (`vendor.scheme`), per the + URL pipeline specification. An adapter is used either as an *adapter + segment* (`s3://bucket/repo|icechunk://tag.v1`), where the context carries + the preceding sub-URLs, or as a *root scheme* (`gh://org/repo`), where + `context.preceding` is empty. + """ + + @classmethod + @abstractmethod + async def open_pipeline_segment( + cls, segment: PipelineSegment, context: PipelineContext + ) -> StorePath: + """ + Resolve `segment`, in the context of the pipeline to its left, into an + open store and the path the pipeline addresses inside it (empty for + the store root). + + The store must honor `context.read_only`; the resolver also enforces + it. This coroutine runs on zarr's I/O event loop, so it must not + block, and must not call zarr's synchronous API (which raises + `SyncError`). Open the preceding pipeline with + [`resolve_preceding`][zarr.storage.PipelineContext.resolve_preceding]. + """ + ... diff --git a/src/zarr/api/asynchronous.py b/src/zarr/api/asynchronous.py index 39e7cdd9d4..588ce5850a 100644 --- a/src/zarr/api/asynchronous.py +++ b/src/zarr/api/asynchronous.py @@ -48,7 +48,7 @@ ZarrUserWarning, ) from zarr.storage import StorePath -from zarr.storage._common import make_store_path +from zarr.storage._common import make_store_path, resolve_zarr_format if TYPE_CHECKING: from collections.abc import Iterable @@ -234,6 +234,7 @@ async def consolidate_metadata( consolidated metadata, this function raises a `TypeError`. See `Store.supports_consolidated_metadata`. """ + zarr_format = resolve_zarr_format(store, zarr_format) store_path = await make_store_path(store, path=path) if not store_path.store.supports_consolidated_metadata: @@ -403,6 +404,7 @@ async def open( mode = "r" else: mode = "a" + zarr_format = resolve_zarr_format(store, zarr_format) store_path = await make_store_path(store, mode=mode, path=path, storage_options=storage_options) # TODO: the mode check below seems wrong! @@ -526,6 +528,7 @@ async def save_array( **kwargs Passed through to [`create`][zarr.api.asynchronous.create], e.g., compressor. """ + zarr_format = resolve_zarr_format(store, zarr_format) if zarr_format is None: zarr_format = _default_zarr_format() if not isinstance(arr, NDArrayLike): @@ -581,6 +584,7 @@ async def save_group( NumPy arrays with data to save. """ + zarr_format = resolve_zarr_format(store, zarr_format) if zarr_format is None: zarr_format = _default_zarr_format() @@ -796,6 +800,7 @@ async def create_group( The new group. """ + zarr_format = resolve_zarr_format(store, zarr_format) if zarr_format is None: zarr_format = _default_zarr_format() @@ -891,6 +896,7 @@ async def open_group( } ) + zarr_format = resolve_zarr_format(store, zarr_format) store_path = await make_store_path(store, mode=mode, storage_options=storage_options, path=path) if attributes is None: attributes = {} @@ -1098,6 +1104,7 @@ async def create( z : array The array. """ + zarr_format = resolve_zarr_format(store, zarr_format) if zarr_format is None: zarr_format = _default_zarr_format() @@ -1333,6 +1340,7 @@ async def open_array( """ mode = kwargs.pop("mode", None) + zarr_format = resolve_zarr_format(store, zarr_format) store_path = await make_store_path(store, path=path, mode=mode, storage_options=storage_options) if "write_empty_chunks" in kwargs: diff --git a/src/zarr/core/array.py b/src/zarr/core/array.py index 06f7338755..46a5d27fc7 100644 --- a/src/zarr/core/array.py +++ b/src/zarr/core/array.py @@ -156,7 +156,7 @@ _parse_bytes_bytes_codec, get_pipeline_class, ) -from zarr.storage._common import StorePath, make_store_path +from zarr.storage._common import StorePath, make_store_path, resolve_zarr_format from zarr.storage._memory import MemoryStore from zarr.storage._utils import _relativize_path @@ -469,6 +469,7 @@ async def _create( Deprecated in favor of [`zarr.api.asynchronous.create_array`][]. """ + zarr_format = resolve_zarr_format(store, zarr_format) dtype_parsed = parse_dtype(dtype, zarr_format=zarr_format) store_path = await make_store_path(store) @@ -821,6 +822,7 @@ async def example(): # ``` """ + zarr_format = resolve_zarr_format(store, zarr_format) store_path = await make_store_path(store) metadata_dict = await get_array_metadata(store_path, zarr_format=zarr_format) # TODO: remove this cast when we have better type hints @@ -4399,6 +4401,7 @@ async def from_array( """ mode: Literal["a"] = "a" config_parsed = parse_array_config(config) + zarr_format = resolve_zarr_format(store, zarr_format) store_path = await make_store_path(store, path=name, mode=mode, storage_options=storage_options) ( @@ -4902,6 +4905,7 @@ async def create_array( else: mode: Literal["a"] = "a" + zarr_format = resolve_zarr_format(store, zarr_format) store_path = await make_store_path( store, path=name, mode=mode, storage_options=storage_options ) diff --git a/src/zarr/core/group.py b/src/zarr/core/group.py index 7792fa58de..c888c3bc85 100644 --- a/src/zarr/core/group.py +++ b/src/zarr/core/group.py @@ -67,7 +67,7 @@ ZarrUserWarning, ) from zarr.storage import StoreLike, StorePath -from zarr.storage._common import make_store_path +from zarr.storage._common import make_store_path, resolve_zarr_format from zarr.storage._utils import _join_paths, _normalize_path_keys, normalize_path if TYPE_CHECKING: @@ -532,6 +532,7 @@ async def from_store( overwrite: bool = False, zarr_format: ZarrFormat = 3, ) -> AsyncGroup: + zarr_format = resolve_zarr_format(store, zarr_format) store_path = await make_store_path(store) attributes = attributes or {} group = cls( @@ -574,6 +575,7 @@ async def open( (``.zmetadata`` by default). Specify the custom key as ``use_consolidated`` to load consolidated metadata from a non-default key. """ + zarr_format = resolve_zarr_format(store, zarr_format) store_path = await make_store_path(store) if not store_path.store.supports_consolidated_metadata: # Fail if consolidated metadata was requested but the Store doesn't support it diff --git a/src/zarr/errors.py b/src/zarr/errors.py index 3e445de2e9..d9029a7430 100644 --- a/src/zarr/errors.py +++ b/src/zarr/errors.py @@ -12,6 +12,7 @@ "MetadataValidationError", "NegativeStepError", "NodeTypeValidationError", + "URLPipelineError", "UnknownCodecError", "UnstableSpecificationWarning", "VindexInvalidSelectionError", @@ -101,6 +102,12 @@ class UnknownCodecError(BaseZarrError): """ +class URLPipelineError(BaseZarrError): + """ + Raised when a URL pipeline cannot be parsed or resolved. + """ + + class NodeTypeValidationError(MetadataValidationError): """ Specialized exception when the node_type of the metadata document is incorrect. diff --git a/src/zarr/registry.py b/src/zarr/registry.py index 307c323153..9ebfed3af9 100644 --- a/src/zarr/registry.py +++ b/src/zarr/registry.py @@ -1,5 +1,7 @@ from __future__ import annotations +import re +import threading import warnings from collections import defaultdict from collections.abc import Mapping @@ -8,7 +10,7 @@ from zarr.core.config import BadConfigError, config from zarr.core.dtype import data_type_registry -from zarr.errors import UnknownCodecError, ZarrUserWarning +from zarr.errors import UnknownCodecError, URLPipelineError, ZarrUserWarning if TYPE_CHECKING: from importlib.metadata import EntryPoint @@ -22,6 +24,7 @@ CodecPipeline, ) from zarr.abc.numcodec import Numcodec + from zarr.abc.url_pipeline import URLPipelineAdapter from zarr.core.buffer import Buffer, NDBuffer from zarr.core.chunk_key_encodings import ChunkKeyEncoding from zarr.core.common import JSON, ZarrFormat @@ -38,6 +41,7 @@ "register_codec", "register_ndbuffer", "register_pipeline", + "register_url_adapter", ] _ZARR_CODEC_DOCS_URL = "https://zarr.readthedocs.io/en/stable/user-guide/extending/#custom-codecs" @@ -257,6 +261,7 @@ def register(self, cls: type[T], qualname: str | None = None) -> None: _buffer_registry: Registry[Buffer] = Registry() _ndbuffer_registry: Registry[NDBuffer] = Registry() _chunk_key_encoding_registry: Registry[ChunkKeyEncoding] = Registry() +_url_adapter_registry: Registry[URLPipelineAdapter] = Registry() """ The registry module is responsible for managing implementations of codecs, @@ -303,6 +308,17 @@ def _collect_entrypoints() -> list[Registry[Any]]: entry_points.select(group="zarr", name="chunk_key_encoding") ) + for entry_point in entry_points.select(group="zarr.url_adapters"): + if _is_adapter_scheme(entry_point.name): + _url_adapter_registry.lazy_load_list.append(entry_point) + else: + warnings.warn( + f"ignoring the 'zarr.url_adapters' entry point {entry_point.name!r} " + f"({entry_point.value!r}): not a valid adapter scheme", + category=ZarrUserWarning, + stacklevel=2, + ) + _pipeline_registry.lazy_load_list.extend(entry_points.select(group="zarr.codec_pipeline")) _pipeline_registry.lazy_load_list.extend( entry_points.select(group="zarr", name="codec_pipeline") @@ -319,6 +335,7 @@ def _collect_entrypoints() -> list[Registry[Any]]: _buffer_registry, _ndbuffer_registry, _chunk_key_encoding_registry, + _url_adapter_registry, ] @@ -504,6 +521,140 @@ def get_chunk_key_encoding_class(key: str) -> type[ChunkKeyEncoding]: return _chunk_key_encoding_registry[key] +# An RFC 3986 URL scheme (ASCII only), e.g. "s3" or "foo.myscheme". +# Use with fullmatch. +_URL_SCHEME_RE = re.compile(r"[a-zA-Z][a-zA-Z0-9+.\-]*") + +# Schemes zarr-python handles itself: the native `file:` and `memory:` roots +# and the `zarr:`/`zarr2:`/`zarr3:` format segments. +_RESERVED_URL_SCHEMES = frozenset({"file", "memory", "zarr", "zarr2", "zarr3"}) + + +def _is_adapter_scheme(scheme: str) -> bool: + return bool(_URL_SCHEME_RE.fullmatch(scheme)) and scheme.lower() not in _RESERVED_URL_SCHEMES + + +def register_url_adapter(scheme: str, cls: type[URLPipelineAdapter]) -> None: + """ + Register a [`URLPipelineAdapter`][zarr.abc.url_pipeline.URLPipelineAdapter] + class for a URL scheme (case-insensitive). + + Registering a scheme that already has an adapter replaces it and emits a + [`ZarrUserWarning`][zarr.errors.ZarrUserWarning]. + + Raises + ------ + ValueError + If `scheme` is not a valid URL scheme, or is one zarr-python handles + itself (`file`, `memory`, `zarr`, `zarr2`, `zarr3`). + """ + if not _URL_SCHEME_RE.fullmatch(scheme): + raise ValueError(f"{scheme!r} is not a valid URL scheme") + if not _is_adapter_scheme(scheme): + raise ValueError( + f"the URL scheme {scheme!r} is handled by zarr-python itself and cannot be " + "given to a URL pipeline adapter" + ) + key = scheme.lower() + previous = _url_adapter_registry.get(key) + if previous is not None and previous is not cls: + warnings.warn( + f"URL pipeline adapter for scheme {scheme!r} is being replaced: " + f"{fully_qualified_name(previous)} -> {fully_qualified_name(cls)}", + category=ZarrUserWarning, + stacklevel=2, + ) + _url_adapter_registry.register(cls, key) + + +def _list_url_adapter_schemes() -> set[str]: + """ + The set of URL schemes with a registered URL pipeline adapter. + + Includes adapters advertised via not-yet-loaded `zarr.url_adapters` + entry points; consulting this does not import any adapter code. + Schemes are case-insensitive and reported lowercased. + """ + schemes = set(_url_adapter_registry) | { + e.name.lower() for e in _url_adapter_registry.lazy_load_list + } + return {scheme for scheme in schemes if _is_adapter_scheme(scheme)} + + +_url_adapter_lock = threading.Lock() + + +def _get_url_adapter(scheme: str) -> type[URLPipelineAdapter]: + """ + Get the URL pipeline adapter class registered for `scheme`. + + Loads pending `zarr.url_adapters` entry points for this scheme only, so + resolving one scheme never imports other providers' packages. When the + scheme is already registered (e.g. with `register_url_adapter`), a + same-named entry point is discarded with a + [`ZarrUserWarning`][zarr.errors.ZarrUserWarning]; when several entry + points share the name, the first one wins and a warning names the rest. + + Raises + ------ + URLPipelineError + If no adapter is registered for the scheme, or if the entry point + providing it fails to import. + """ + key = scheme.lower() + if not _is_adapter_scheme(key): + raise URLPipelineError( + f"'{key}:' is handled by zarr-python itself and cannot be an adapter segment" + ) + # Claim this scheme's entry points under the lock so concurrent loads of + # different schemes can't clobber the list; import outside it, since an + # import may be slow or itself resolve URL adapters. + with _url_adapter_lock: + registered = _url_adapter_registry.get(key) + pending = [e for e in _url_adapter_registry.lazy_load_list if e.name.lower() == key] + if pending: + _url_adapter_registry.lazy_load_list[:] = [ + e for e in _url_adapter_registry.lazy_load_list if e.name.lower() != key + ] + if pending and registered is not None: + warnings.warn( + f"URL pipeline adapter for scheme {scheme!r} is already registered " + f"({fully_qualified_name(registered)}); ignoring entry point(s) " + f"{[e.value for e in pending]} of the same name", + category=ZarrUserWarning, + stacklevel=2, + ) + elif pending: + entry_point, *duplicates = pending + if duplicates: + warnings.warn( + f"multiple 'zarr.url_adapters' entry points are named {entry_point.name!r}; " + f"using {entry_point.value!r} and ignoring {[e.value for e in duplicates]}", + category=ZarrUserWarning, + stacklevel=2, + ) + try: + cls = entry_point.load() + except Exception as exc: + # leave the entry points discoverable so a transient import + # failure is not permanent + with _url_adapter_lock: + _url_adapter_registry.lazy_load_list.extend(pending) + raise URLPipelineError( + f"the 'zarr.url_adapters' entry point {entry_point.name!r} could not be loaded" + ) from exc + with _url_adapter_lock: + if key not in _url_adapter_registry: + _url_adapter_registry.register(cls, qualname=key) + try: + return _url_adapter_registry[key] + except KeyError: + raise URLPipelineError( + f"no URL pipeline adapter is registered for scheme {scheme!r}. Adapters are " + "provided by packages via the 'zarr.url_adapters' entry-point group." + ) from None + + _collect_entrypoints() diff --git a/src/zarr/storage/__init__.py b/src/zarr/storage/__init__.py index f1bd1724af..11839ad9b9 100644 --- a/src/zarr/storage/__init__.py +++ b/src/zarr/storage/__init__.py @@ -10,6 +10,7 @@ from zarr.storage._logging import LoggingStore from zarr.storage._memory import GpuMemoryStore, ManagedMemoryStore, MemoryStore from zarr.storage._obstore import ObjectStore +from zarr.storage._url_pipeline import PipelineContext, PipelineSegment, URLPipeline from zarr.storage._wrapper import WrapperStore from zarr.storage._zip import ZipStore @@ -21,8 +22,11 @@ "ManagedMemoryStore", "MemoryStore", "ObjectStore", + "PipelineContext", + "PipelineSegment", "StoreLike", "StorePath", + "URLPipeline", "WrapperStore", "ZipStore", ] diff --git a/src/zarr/storage/_common.py b/src/zarr/storage/_common.py index 8591447449..cd9b2f40db 100644 --- a/src/zarr/storage/_common.py +++ b/src/zarr/storage/_common.py @@ -2,7 +2,7 @@ import importlib.util from pathlib import Path -from typing import TYPE_CHECKING, Any, Literal, Self +from typing import TYPE_CHECKING, Any, Literal, Self, overload from zarr.abc.store import ( ByteRequest, @@ -21,9 +21,15 @@ AccessModeLiteral, ZarrFormat, ) -from zarr.errors import ContainsArrayAndGroupError, ContainsArrayError, ContainsGroupError +from zarr.errors import ( + ContainsArrayAndGroupError, + ContainsArrayError, + ContainsGroupError, + URLPipelineError, +) from zarr.storage._local import LocalStore from zarr.storage._memory import ManagedMemoryStore, MemoryStore +from zarr.storage._url_pipeline import _FORMAT_SCHEMES, URLPipeline from zarr.storage._utils import UPath, _join_paths, normalize_path, parse_store_url _has_fsspec = importlib.util.find_spec("fsspec") @@ -299,7 +305,7 @@ def __eq__(self, other: object) -> bool: return False -type StoreLike = Store | StorePath | FSMap | Path | UPath | str | dict[str, Buffer] +type StoreLike = Store | StorePath | URLPipeline | FSMap | Path | UPath | str | dict[str, Buffer] async def make_store( @@ -320,6 +326,8 @@ async def make_store( - `None` = `MemoryStore` object. - `FSMap` = `FsspecStore` object. - `UPath` = `FsspecStore` object, or `LocalStore` for a local `UPath`. + - `URLPipeline` = the store a [URL pipeline][user-guide-url-pipelines] resolves to; + the pipeline must address the store root. Parameters ---------- @@ -346,6 +354,18 @@ async def make_store( """ from zarr.storage._fsspec import FsspecStore # circular import + if isinstance(store_like, URLPipeline): + if mode is not None and mode not in ANY_ACCESS_MODE: + raise ValueError(f"Invalid mode: {mode}, expected one of {ANY_ACCESS_MODE}") + result = await store_like._resolve_store_path(mode=mode, storage_options=storage_options) + if result.path: + result.store.close() + raise URLPipelineError( + f"{store_like!r} resolves to a path inside a store; " + "use zarr.open() or make_store_path() instead of make_store()" + ) + return result.store + # Parse URL early so we can reuse the result for both validation and routing parsed = parse_store_url(store_like) if isinstance(store_like, str) else None @@ -455,14 +475,30 @@ async def make_store_path( TypeError If the StoreLike object is not one of the supported types, or if storage_options is provided but not used. ValueError - If path is provided for a store that does not support it. + If mode is invalid, or path is provided for a store that does not support it. + URLPipelineError + If `store_like` is a [URL pipeline][user-guide-url-pipelines] that cannot be + resolved. See Also -------- make_store """ + if mode is not None and mode not in ANY_ACCESS_MODE: + raise ValueError(f"Invalid mode: {mode}, expected one of {ANY_ACCESS_MODE}") path_normalized = normalize_path(path) + if isinstance(store_like, URLPipeline): + result = await store_like._resolve_store_path(mode=mode, storage_options=storage_options) + try: + return await StorePath.open( + result.store, path=_join_paths([result.path, path_normalized]), mode=mode + ) + except BaseException: + # the store was opened for this call; don't leak it + result.store.close() + raise + if isinstance(store_like, StorePath): # Already a StorePath if storage_options: @@ -482,6 +518,43 @@ async def make_store_path( return await StorePath.open(store, path=path_normalized, mode=mode) +@overload +def resolve_zarr_format(store_like: StoreLike | None, zarr_format: ZarrFormat) -> ZarrFormat: ... + + +@overload +def resolve_zarr_format( + store_like: StoreLike | None, zarr_format: ZarrFormat | None +) -> ZarrFormat | None: ... + + +def resolve_zarr_format( + store_like: StoreLike | None, zarr_format: ZarrFormat | None +) -> ZarrFormat | None: + """ + Merge the format selected by a `URLPipeline`'s trailing `zarr2:`/`zarr3:` + segment with the caller's `zarr_format`. Other `StoreLike`s, and a trailing + `zarr:` segment, leave `zarr_format` unchanged. + + Raises + ------ + ValueError + If `zarr_format` is not None and differs from the pipeline's format. + """ + if not isinstance(store_like, URLPipeline) or store_like._format_segment is None: + return zarr_format + pipeline_format = _FORMAT_SCHEMES[store_like._format_segment.scheme] + if pipeline_format is None: + return zarr_format + if zarr_format is not None and zarr_format != pipeline_format: + raise ValueError( + f"zarr_format={zarr_format} conflicts with the 'zarr{pipeline_format}:' segment " + f"of {store_like!r}; pass zarr_format={pipeline_format} (or None where accepted), " + "or open it with zarr.open_group / zarr.open_array, which follow the pipeline" + ) + return pipeline_format + + async def ensure_no_existing_node( store_path: StorePath, zarr_format: ZarrFormat, diff --git a/src/zarr/storage/_url_pipeline.py b/src/zarr/storage/_url_pipeline.py new file mode 100644 index 0000000000..4d89d2ba2a --- /dev/null +++ b/src/zarr/storage/_url_pipeline.py @@ -0,0 +1,438 @@ +""" +Parsing and resolution of URL pipelines (https://github.com/jbms/url-pipeline). + +Only [`URLPipeline.from_url`][zarr.storage.URLPipeline.from_url] reads a string +as a pipeline. Adapters are imported lazily, and no percent-escape is decoded. +""" + +from __future__ import annotations + +import dataclasses +import re +import sys +from pathlib import Path, PureWindowsPath +from typing import TYPE_CHECKING, Any + +from zarr.errors import URLPipelineError +from zarr.registry import _URL_SCHEME_RE, _get_url_adapter, _list_url_adapter_schemes +from zarr.storage._memory import ManagedMemoryStore +from zarr.storage._utils import _join_paths, normalize_path + +if TYPE_CHECKING: + from zarr.abc.store import Store + from zarr.core.common import AccessModeLiteral, ZarrFormat + from zarr.storage._common import StorePath + +__all__ = ["PipelineContext", "PipelineSegment", "URLPipeline"] + +# Format segments are read by zarr-python itself, not by an adapter: they +# select the zarr format (`zarr:` detects it), and their body is a node path. +_FORMAT_SCHEMES: dict[str, ZarrFormat | None] = {"zarr": None, "zarr2": 2, "zarr3": 3} + +# Roots zarr resolves itself; an adapter can never claim them. +_NATIVE_ROOT_SCHEMES = frozenset({"file", "memory"}) + +# More segments than any real pipeline needs; bounds resolution recursion. +_MAX_SEGMENTS = 32 + +_CONTROL_CHAR_RE = re.compile(r"[\x00-\x1f\x7f]") +_WINDOWS_DRIVE_RE = re.compile(r"[A-Za-z]:/") + + +def _redact(text: str) -> str: + """Hide query strings and URL userinfo, which may carry credentials.""" + text = re.sub(r"//[^/@|?]*@", "//…@", text) + return re.sub(r"\?[^|]*", "?…", text) + + +def _check_scheme(scheme: str, raw: str) -> None: + if not _URL_SCHEME_RE.fullmatch(scheme): + raise URLPipelineError(f"invalid URL scheme {scheme!r} in the sub-URL {_redact(raw)!r}") + + +@dataclasses.dataclass(frozen=True) +class PipelineSegment: + """ + One `|`-delimited sub-URL of a URL pipeline. + + Attributes + ---------- + scheme : str + The lowercased URL scheme. + body : str + The text after `scheme:` and before any `?`, verbatim (case-significant + content such as Icechunk snapshot IDs is preserved). + query : str | None + The raw query string after `?`, or None. + raw : str + The original sub-URL text. + """ + + scheme: str + body: str + query: str | None + raw: str + + def __post_init__(self) -> None: + for name in ("scheme", "body", "raw"): + if not isinstance(getattr(self, name), str): + raise TypeError( + f"PipelineSegment.{name} must be a str, " + f"not {type(getattr(self, name)).__name__}" + ) + if self.query is not None and not isinstance(self.query, str): + raise TypeError( + f"PipelineSegment.query must be a str or None, not {type(self.query).__name__}" + ) + _check_scheme(self.scheme, self.raw) + if self.scheme != self.scheme.lower(): + raise URLPipelineError(f"PipelineSegment.scheme must be lowercase: {self.scheme!r}") + + def __repr__(self) -> str: + return f"PipelineSegment({_redact(self.raw)!r})" + + +@dataclasses.dataclass(frozen=True, eq=False) +class PipelineContext: + """ + Context handed to a [`URLPipelineAdapter`][zarr.abc.url_pipeline.URLPipelineAdapter] + describing the pipeline to the left of its segment. + + Instances are immutable and compare by identity; use `dataclasses.replace` + to derive one with other values. + + Attributes + ---------- + preceding : tuple[PipelineSegment, ...] + The sub-URLs to the left of the adapter's segment, outer to inner. + Empty when the adapter's segment is the pipeline root. + mode : AccessModeLiteral | None + The caller's access mode, or None. Adapters for read-only resources + should raise for every mode except `"r"`. + storage_options : dict[str, Any] | None + The caller's options. They configure the root sub-URL (e.g. fsspec + options), and every adapter in the pipeline sees all of them: an + adapter should consume only keys namespaced for it (e.g. + `myscheme_token`), and strip them before resolving the rest of the + pipeline. + """ + + preceding: tuple[PipelineSegment, ...] + mode: AccessModeLiteral | None + storage_options: dict[str, Any] | None = dataclasses.field(repr=False) + + @property + def read_only(self) -> bool: + """ + True when the caller requires a read-only store (`mode == "r"`). + The resolver enforces this on the store the adapter returns. + """ + return self.mode == "r" + + @property + def preceding_url(self) -> str: + """ + The pipeline to the left of this segment, reconstructed exactly. + + An adapter that consumes this string instead of calling + `resolve_preceding` must validate every preceding segment itself, + and raise `URLPipelineError` for any it does not understand. + """ + return "|".join(segment.raw for segment in self.preceding) + + async def resolve_preceding(self) -> StorePath: + """ + Resolve the preceding pipeline, with this context's `mode` and + `storage_options`, into a store and a residual path. + + Wrapper adapters (e.g. `zip:`) that only read the preceding resource + should resolve it read-only: + `await dataclasses.replace(context, mode="r").resolve_preceding()`. + A local *file* root can currently only be resolved in mode `"r"`. + """ + if not self.preceding: + raise URLPipelineError( + "this adapter segment is at the pipeline root; " + "there is no preceding sub-URL to resolve" + ) + return await _resolve(self.preceding, mode=self.mode, storage_options=self.storage_options) + + +def _split_query(sub_url: str) -> tuple[str, str | None]: + if "#" in sub_url: + raise URLPipelineError( + f"URL pipeline sub-URLs do not support fragments ('#'): {_redact(sub_url)!r}" + ) + body, sep, query = sub_url.partition("?") + return body, query if sep else None + + +def _parse_pipeline(url: str) -> tuple[PipelineSegment, ...]: + """ + Split a URL pipeline into segments. Adapter segments are `scheme:body`, + where the colon is optional when the body is empty (`zip` ≡ `zip:`). + """ + if not isinstance(url, str): + raise TypeError(f"a URL pipeline must be a str, not {type(url).__name__}") + if (match := _CONTROL_CHAR_RE.search(url)) is not None: + raise URLPipelineError(f"URL pipeline contains the control character {match.group()!r}") + parts = url.split("|") + if len(parts) > _MAX_SEGMENTS: + raise URLPipelineError(f"URL pipeline has more than {_MAX_SEGMENTS} segments") + if any(not part for part in parts): + raise URLPipelineError(f"URL pipeline contains an empty sub-URL: {_redact(url)!r}") + + segments = [] + for index, part in enumerate(parts): + if part != part.strip(): + raise URLPipelineError(f"sub-URL {_redact(part)!r} has leading or trailing whitespace") + scheme_and_body, query = _split_query(part) + scheme, sep, body = scheme_and_body.partition(":") + if index == 0: + if not sep or not scheme or "/" in scheme or "\\" in scheme: + raise URLPipelineError( + f"the root sub-URL {_redact(part)!r} has no URL scheme; spell a local path " + "as an absolute 'file:' URL, or pass a path containing '|' as a plain string" + ) + _check_scheme(scheme, part) + if len(scheme) == 1: + raise URLPipelineError( + f"the root sub-URL {_redact(part)!r} looks like a Windows drive path; " + "spell it as a 'file:' URL (e.g. 'file:/C:/data')" + ) + else: + _check_scheme(scheme, part) + segments.append(PipelineSegment(scheme=scheme.lower(), body=body, query=query, raw=part)) + return tuple(segments) + + +@dataclasses.dataclass(frozen=True) +class URLPipeline: + """ + A parsed URL pipeline, accepted wherever a `StoreLike` is. + + Build one with [`from_url`][zarr.storage.URLPipeline.from_url] and pass it to + any `StoreLike`-taking function, e.g. + `zarr.open_group(URLPipeline.from_url("s3://bucket/data.zip|zip:|zarr3:"))`. + Instances are immutable and hashable values; nothing is resolved until the + pipeline is opened. + + Attributes + ---------- + segments : tuple[PipelineSegment, ...] + The sub-URLs in order: the root, any adapter segments, and an optional + trailing format segment (`zarr:`, `zarr2:` or `zarr3:`). + + Raises + ------ + TypeError + If `segments` is not a tuple of `PipelineSegment`. + URLPipelineError + If there are no segments or more than 32, if a format segment is + anywhere but last (including the root), or if a format segment carries + a query or a node path with `.` or `..` components. + """ + + segments: tuple[PipelineSegment, ...] + + def __post_init__(self) -> None: + segments: object = self.segments # runtime check of a hand-built pipeline + if not isinstance(segments, tuple) or not all( + isinstance(segment, PipelineSegment) for segment in segments + ): + raise TypeError( + "URLPipeline.segments must be a tuple of PipelineSegment; " + "use URLPipeline.from_url to parse a string" + ) + if not 1 <= len(self.segments) <= _MAX_SEGMENTS: + raise URLPipelineError(f"a URL pipeline needs 1 to {_MAX_SEGMENTS} segments") + if self.segments[0].scheme in _FORMAT_SCHEMES: + raise URLPipelineError(f"a format segment cannot be the root of {self!r}") + if any(segment.scheme in _FORMAT_SCHEMES for segment in self.segments[1:-1]): + raise URLPipelineError(f"a format segment must be the last segment of {self!r}") + format_segment = self._format_segment + if format_segment is not None: + if format_segment.query is not None: + raise URLPipelineError(f"format segments do not accept a query: {self!r}") + try: + normalize_path(format_segment.body) + except ValueError: + raise URLPipelineError( + f"the node path of a format segment must not contain '.' or '..' components: {self!r}" + ) from None + + @classmethod + def from_url(cls, url: str) -> URLPipeline: + """ + Parse a URL pipeline string. A URL without `|` is a pipeline of its + root alone. Parsing loads no adapter. + + Raises + ------ + TypeError + If `url` is not a `str`. + URLPipelineError + If the string cannot be parsed: an empty sub-URL, a root without a + URL scheme (including a local or Windows drive path), an invalid + scheme, a fragment, an ASCII control character, or whitespace at + either end of a sub-URL, or the pipeline is invalid (see the class + docstring). + """ + return cls(_parse_pipeline(url)) + + def __str__(self) -> str: + return "|".join(segment.raw for segment in self.segments) + + def __repr__(self) -> str: + return f"URLPipeline({_redact(str(self))!r})" + + @property + def _format_segment(self) -> PipelineSegment | None: + last = self.segments[-1] + if len(self.segments) > 1 and last.scheme in _FORMAT_SCHEMES: + return last + return None + + async def _resolve_store_path( + self, + *, + mode: AccessModeLiteral | None = None, + storage_options: dict[str, Any] | None = None, + ) -> StorePath: + """ + Resolve the pipeline into a store and a path: the adapters' residual + path joined with the node path of a trailing format segment. + """ + from zarr.storage._common import StorePath # circular import + + format_segment = self._format_segment + if format_segment is None: + return await _resolve(self.segments, mode=mode, storage_options=storage_options) + resolved = await _resolve(self.segments[:-1], mode=mode, storage_options=storage_options) + node_path = normalize_path(format_segment.body) + return StorePath(resolved.store, _join_paths([resolved.path, node_path])) + + +async def _resolve( + segments: tuple[PipelineSegment, ...], + *, + mode: AccessModeLiteral | None, + storage_options: dict[str, Any] | None, +) -> StorePath: + from zarr.storage._common import StorePath # circular import + + *preceding, last = segments + if not preceding and ( + last.scheme in _NATIVE_ROOT_SCHEMES or last.scheme not in _list_url_adapter_schemes() + ): + return StorePath(await _resolve_root(last, mode=mode, storage_options=storage_options)) + + adapter_cls = _get_url_adapter(last.scheme) + context = PipelineContext( + preceding=tuple(preceding), mode=mode, storage_options=storage_options + ) + resolved = await adapter_cls.open_pipeline_segment(last, context) + if mode == "r" and not resolved.store.read_only: + # enforce read-only rather than trusting the adapter + try: + read_only_store = resolved.store.with_read_only(True) + await read_only_store._ensure_open() + except NotImplementedError as exc: + resolved.store.close() + raise URLPipelineError( + f"adapter {last.scheme!r} returned a writable store for mode 'r', " + "and the store does not support read-only conversion via 'with_read_only()'" + ) from exc + except Exception: + resolved.store.close() + raise + resolved = StorePath(read_only_store, resolved.path) + return resolved + + +def _reject_dot_segments(path: str, segment: PipelineSegment) -> None: + if any(part in (".", "..") for part in re.split(r"[/\\]", path)): + raise URLPipelineError( + f"'{segment.scheme}:' pipeline root paths must not contain '.' or '..' " + f"components: {_redact(segment.raw)!r}" + ) + + +async def _resolve_root( + segment: PipelineSegment, + *, + mode: AccessModeLiteral | None, + storage_options: dict[str, Any] | None, +) -> Store: + """ + Resolve a root sub-URL. `memory:` and `file:` follow the URL pipeline + spec; every other scheme goes to the plain-string machinery (fsspec). + """ + from zarr.storage._common import make_store # circular import + + if segment.scheme in _NATIVE_ROOT_SCHEMES and segment.query is not None: + raise URLPipelineError( + f"'{segment.scheme}:' pipeline roots do not accept a query: {_redact(segment.raw)!r}" + ) + + if segment.scheme == "memory": + # one in-memory store; the body is a path within it + # (memory:a ≡ memory:/a ≡ memory://a) + if storage_options is not None: + raise TypeError( + "'storage_options' was provided but unused. " + "'storage_options' is only used when the store is passed as an FSSpec URI string.", + ) + path = segment.body.lstrip("/") + _reject_dot_segments(path, segment) + return ManagedMemoryStore(name="", path=path, read_only=mode == "r") + + if segment.scheme == "file": + # file://localhost/p ≡ file:///p ≡ file:/p. On Windows, check the + # '/'-spelled path, which also rejects UNC paths like /\host. + body = segment.body.replace("\\", "/") if sys.platform == "win32" else segment.body + if body.startswith("//"): + authority, sep, rest = body[2:].partition("/") + if authority.lower() not in ("", "localhost"): + raise URLPipelineError( + f"unsupported authority in the 'file:' pipeline root {_redact(segment.raw)!r}; " + "only an empty authority or 'localhost' is allowed" + ) + body = f"/{rest}" if sep else "" + if body.startswith("//"): + raise URLPipelineError( + f"'file:' pipeline root paths must not start with '//': " + f"{_redact(segment.raw)!r}" + ) + if sys.platform == "win32": + # a file URL spells a drive path /C:/...; drop the leading slash + if body.startswith("/") and _WINDOWS_DRIVE_RE.match(body[1:]): + body = body[1:] + drive = PureWindowsPath(body).drive + is_absolute = (body.startswith("/") and not drive) or bool( + _WINDOWS_DRIVE_RE.match(body) + ) + if ":" in body[len(drive) :]: + # elsewhere ':' names an NTFS alternate data stream + raise URLPipelineError( + f"':' is only allowed in the drive of a 'file:' pipeline root on " + f"Windows: {_redact(segment.raw)!r}" + ) + else: + is_absolute = body.startswith("/") + if not is_absolute: + raise URLPipelineError( + f"'file:' pipeline roots must carry an absolute local path: " + f"{_redact(segment.raw)!r}" + ) + _reject_dot_segments(body, segment) + return await make_store(Path(body), mode=mode, storage_options=storage_options) + + # fsspec matches protocols case-sensitively; URL schemes are not + url = segment.scheme + segment.raw[len(segment.scheme) :] + try: + return await make_store(url, mode=mode, storage_options=storage_options) + except ValueError as exc: + raise URLPipelineError( + f"could not resolve the pipeline root {_redact(segment.raw)!r}: {_redact(str(exc))}" + ) from exc diff --git a/tests/package_with_entrypoint-0.1.dist-info/entry_points.txt b/tests/package_with_entrypoint-0.1.dist-info/entry_points.txt index 7eb0eb7c86..456c725b4e 100644 --- a/tests/package_with_entrypoint-0.1.dist-info/entry_points.txt +++ b/tests/package_with_entrypoint-0.1.dist-info/entry_points.txt @@ -13,4 +13,6 @@ another_ndbuffer = package_with_entrypoint:TestEntrypointGroup.NDBuffer [zarr.codec_pipeline] another_pipeline = package_with_entrypoint:TestEntrypointGroup.Pipeline [zarr.data_type] -new_data_type = package_with_entrypoint:TestDataType \ No newline at end of file +new_data_type = package_with_entrypoint:TestDataType +[zarr.url_adapters] +example-pkg.entrypoint-scheme = package_with_entrypoint:TestEntrypointURLAdapter diff --git a/tests/package_with_entrypoint/__init__.py b/tests/package_with_entrypoint/__init__.py index 23afcf1dc2..521cb8a968 100644 --- a/tests/package_with_entrypoint/__init__.py +++ b/tests/package_with_entrypoint/__init__.py @@ -7,6 +7,7 @@ import zarr.core.buffer from zarr.abc.codec import ArrayBytesCodec, CodecInput, CodecPipeline +from zarr.abc.url_pipeline import URLPipelineAdapter from zarr.codecs import BytesCodec from zarr.core.buffer import Buffer, NDBuffer from zarr.core.dtype.npy.bool import Bool @@ -19,6 +20,7 @@ from zarr.core.array_spec import ArraySpec from zarr.core.common import ZarrFormat from zarr.core.dtype.common import DTypeJSON, DTypeSpec_V2 + from zarr.storage import PipelineContext, PipelineSegment, StorePath class TestEntrypointCodec(ArrayBytesCodec): @@ -100,3 +102,16 @@ def to_json(self, zarr_format: ZarrFormat) -> str | DTypeSpec_V2: # type: ignor if zarr_format == 3: return self._zarr_v3_name raise ValueError("zarr_format must be 2 or 3") + + +class TestEntrypointURLAdapter(URLPipelineAdapter): + """URL pipeline adapter discovered via the zarr.url_adapters entry point.""" + + @classmethod + async def open_pipeline_segment( + cls, segment: PipelineSegment, context: PipelineContext + ) -> StorePath: + from zarr.storage import MemoryStore, StorePath + + store = await MemoryStore.open(read_only=context.read_only) + return StorePath(store, path=segment.body) diff --git a/tests/test_url_pipeline/__init__.py b/tests/test_url_pipeline/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/test_url_pipeline/conftest.py b/tests/test_url_pipeline/conftest.py new file mode 100644 index 0000000000..13c717e04e --- /dev/null +++ b/tests/test_url_pipeline/conftest.py @@ -0,0 +1,22 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + +import pytest + +import zarr.registry + +if TYPE_CHECKING: + from collections.abc import Generator + + +@pytest.fixture +def clean_url_adapter_registry() -> Generator[None, None, None]: + """Snapshot and restore the URL adapter registry around a test.""" + registry = zarr.registry._url_adapter_registry + saved = dict(registry) + saved_lazy = list(registry.lazy_load_list) + yield + registry.clear() + registry.update(saved) + registry.lazy_load_list[:] = saved_lazy diff --git a/tests/test_url_pipeline/test_open_path.py b/tests/test_url_pipeline/test_open_path.py new file mode 100644 index 0000000000..e429588f3e --- /dev/null +++ b/tests/test_url_pipeline/test_open_path.py @@ -0,0 +1,410 @@ +""" +URL pipelines through the public open path: `zarr.open_*`, `make_store_path`, +`make_store` and `consolidate_metadata`, rather than the private resolver. +""" + +from __future__ import annotations + +import sys +from typing import TYPE_CHECKING, ClassVar + +import pytest + +import zarr +import zarr.registry +from zarr.abc.url_pipeline import URLPipelineAdapter +from zarr.errors import GroupNotFoundError, URLPipelineError, ZarrUserWarning +from zarr.registry import register_url_adapter +from zarr.storage import MemoryStore, PipelineSegment, StorePath, URLPipeline +from zarr.storage._common import make_store, make_store_path + +if TYPE_CHECKING: + from pathlib import Path + + from zarr.storage import PipelineContext + +pytestmark = pytest.mark.usefixtures("clean_url_adapter_registry") + + +class ClosingStore(MemoryStore): + """A memory store that records how often it is closed.""" + + closes: ClassVar[list[ClosingStore]] = [] + + def close(self) -> None: + ClosingStore.closes.append(self) + super().close() + + +class RecordingAdapter(URLPipelineAdapter): + """Records each call; returns a fresh `ClosingStore` honoring `read_only`.""" + + calls: ClassVar[list[PipelineContext]] = [] + read_only_store: ClassVar[bool] = False + + @classmethod + async def open_pipeline_segment( + cls, segment: PipelineSegment, context: PipelineContext + ) -> StorePath: + cls.calls.append(context) + store = await ClosingStore.open(read_only=cls.read_only_store or context.read_only) + return StorePath(store, segment.body) + + +@pytest.fixture +def recording() -> type[RecordingAdapter]: + RecordingAdapter.calls = [] + RecordingAdapter.read_only_store = False + ClosingStore.closes = [] + register_url_adapter("rec", RecordingAdapter) + return RecordingAdapter + + +class TestModeValidation: + @pytest.mark.parametrize("mode", ["bogus", "R", "rw"]) + async def test_invalid_mode_rejected_before_adapters_run( + self, recording: type[RecordingAdapter], mode: str + ) -> None: + with pytest.raises(ValueError, match="Invalid mode"): + zarr.open_group(URLPipeline.from_url("memory:|rec:"), mode=mode) # type: ignore[arg-type] + with pytest.raises(ValueError, match="Invalid mode"): + await make_store_path(URLPipeline.from_url("memory:|rec:"), mode=mode) # type: ignore[arg-type] + assert recording.calls == [] + + async def test_invalid_mode_rejected_for_plain_stores(self) -> None: + with pytest.raises(ValueError, match="Invalid mode"): + await make_store_path({}, mode="bogus") # type: ignore[arg-type] + + +class TestStoresClosedOnError: + def test_read_only_store_in_mode_a(self, recording: type[RecordingAdapter]) -> None: + recording.read_only_store = True + with pytest.raises(ValueError, match="Store is read-only but mode is 'a'"): + zarr.open_group(URLPipeline.from_url("memory:|rec:"), mode="a") + assert len(ClosingStore.closes) == 1 + + async def test_mode_w_minus_on_existing_data(self, recording: type[RecordingAdapter]) -> None: + class NonEmptyAdapter(RecordingAdapter): + @classmethod + async def open_pipeline_segment( + cls, segment: PipelineSegment, context: PipelineContext + ) -> StorePath: + store_path = await super().open_pipeline_segment(segment, context) + zarr.create_group(store_path.store) + return store_path + + register_url_adapter("full", NonEmptyAdapter) + with pytest.raises(FileExistsError): + await make_store_path(URLPipeline.from_url("memory:|full:"), mode="w-") + assert len(ClosingStore.closes) == 1 + + async def test_make_store_residual_path(self, recording: type[RecordingAdapter]) -> None: + with pytest.raises(URLPipelineError, match="path inside a store"): + await make_store(URLPipeline.from_url("memory:|rec:sub")) + assert len(ClosingStore.closes) == 1 + + def test_success_does_not_close(self, recording: type[RecordingAdapter]) -> None: + zarr.open_group(URLPipeline.from_url("memory:|rec:"), mode="w") + assert ClosingStore.closes == [] + + +class TestConsolidateMetadata: + def test_honors_format_segment(self, tmp_path: Path) -> None: + root = tmp_path.as_posix() + zarr.create_group(tmp_path, zarr_format=3).create_array("x", shape=(2,), dtype="i4") + # a zarr2: pipeline over v3 data finds no v2 group + with pytest.raises(FileNotFoundError): + zarr.consolidate_metadata(URLPipeline.from_url(f"file:{root}|zarr2:")) + with pytest.warns(ZarrUserWarning, match="Consolidated metadata"): + group = zarr.consolidate_metadata(URLPipeline.from_url(f"file:{root}|zarr3:")) + assert group.metadata.consolidated_metadata is not None + + def test_conflicting_format_raises(self, tmp_path: Path) -> None: + zarr.create_group(tmp_path, zarr_format=2) + with pytest.raises(ValueError, match="conflicts with"): + zarr.consolidate_metadata( + URLPipeline.from_url(f"file:{tmp_path.as_posix()}|zarr2:"), zarr_format=3 + ) + + +class TestFormatSegments: + @pytest.mark.parametrize("url", ["zarr3:", "zarr2:/data|zarr3:", "zarr:x|rec:", "ZARR3:"]) + def test_format_scheme_cannot_be_root(self, url: str) -> None: + with pytest.raises(URLPipelineError, match="cannot be the root"): + URLPipeline.from_url(url) + + @pytest.mark.parametrize("body", ["../x", "a/../../x", "./x", "a/./b"]) + def test_node_path_dot_segments_rejected_at_parse(self, body: str) -> None: + with pytest.raises(URLPipelineError, match="node path of a format segment"): + URLPipeline.from_url(f"memory:|zarr3:{body}") + + def test_node_path_traversal_creates_nothing(self, tmp_path: Path) -> None: + target = tmp_path / "newdir" + with pytest.raises(URLPipelineError): + zarr.open_group(URLPipeline.from_url(f"file:{target.as_posix()}|zarr3:../x"), mode="w") + assert not target.exists() + + def test_zarr_segment_detects_format(self, tmp_path: Path) -> None: + root = tmp_path.as_posix() + zarr.create_group(URLPipeline.from_url(f"file:{root}|zarr2:g"), zarr_format=2) + group = zarr.open_group(URLPipeline.from_url(f"file:{root}|zarr:g"), mode="r") + assert group.metadata.zarr_format == 2 + assert group.path == "g" + # no format is selected, so a default zarr_format=3 does not conflict + zarr.create_array( + URLPipeline.from_url(f"file:{root}|zarr:"), name="a", shape=(2,), dtype="i4" + ) + assert ( + zarr.open_array(URLPipeline.from_url(f"file:{root}|zarr:a")).metadata.zarr_format == 3 + ) + + def test_zarr_segment_must_be_last(self) -> None: + with pytest.raises(URLPipelineError, match="must be the last segment"): + URLPipeline.from_url("memory:|zarr:|rec:") + + +class TestFileRoots: + @pytest.mark.parametrize( + "url", ["file:////data", "file:////nonexistent/data", "file://localhost//data"] + ) + def test_double_slash_path_rejected( + self, url: str, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.chdir(tmp_path) + with pytest.raises(URLPipelineError, match="must not start with '//'"): + zarr.open_group(URLPipeline.from_url(url), mode="w") + assert list(tmp_path.iterdir()) == [] + + @pytest.mark.parametrize("suffix", ["/../x", "/./x", "/a/../../x"]) + def test_dot_segments_rejected(self, suffix: str, tmp_path: Path) -> None: + with pytest.raises(URLPipelineError, match=r"'\.' or '\.\.'"): + zarr.open_group(URLPipeline.from_url(f"file:{tmp_path.as_posix()}{suffix}"), mode="w") + + def test_root_opens_absolute_local_path(self, tmp_path: Path) -> None: + group = zarr.open_group(URLPipeline.from_url(f"file:{tmp_path.as_posix()}/g"), mode="w") + group.attrs["x"] = 1 + assert (tmp_path / "g" / "zarr.json").exists() + + def test_storage_options_rejected(self, tmp_path: Path) -> None: + with pytest.raises(TypeError, match="'storage_options' was provided but unused"): + zarr.open_group( + URLPipeline.from_url(f"file:{tmp_path.as_posix()}"), + mode="w", + storage_options={"anon": True}, + ) + + def test_authority_is_case_insensitive(self, tmp_path: Path) -> None: + # spell the path URL-style: on Windows C:/... gains a leading slash + posix = tmp_path.as_posix() + url_path = posix if posix.startswith("/") else f"/{posix}" + zarr.open_group(URLPipeline.from_url(f"file://LOCALHOST{url_path}"), mode="w") + assert (tmp_path / "zarr.json").exists() + + +class TestMemoryRoot: + async def test_single_store_body_is_a_path(self) -> None: + """The body of a `memory:` root is a path, so it can be split across adapter segments.""" + + class PathAdapter(URLPipelineAdapter): + @classmethod + async def open_pipeline_segment( + cls, segment: PipelineSegment, context: PipelineContext + ) -> StorePath: + preceding = await context.resolve_preceding() + return StorePath(preceding.store, f"{preceding.path}/{segment.body}".strip("/")) + + register_url_adapter("sub", PathAdapter) + # keep the writer alive: managed memory stores drop their data once unreferenced + writer = zarr.open_group(URLPipeline.from_url("memory:mem-one/a/b|sub:"), mode="w") + writer.attrs["k"] = 1 + for url in ("memory:mem-one|sub:a/b", "memory:|sub:mem-one/a/b", "memory://mem-one/a/b"): + assert zarr.open_group(URLPipeline.from_url(url), mode="r").attrs["k"] == 1 + + @pytest.mark.parametrize("body", ["../x", "a/../b", "./a"]) + async def test_dot_segments_rejected(self, body: str) -> None: + with pytest.raises(URLPipelineError, match=r"'\.' or '\.\.'"): + await make_store_path(URLPipeline.from_url(f"memory:{body}")) + + @pytest.mark.parametrize("storage_options", [{}, {"anon": True}]) + async def test_storage_options_rejected(self, storage_options: dict[str, object]) -> None: + with pytest.raises(TypeError, match="'storage_options' was provided but unused"): + await make_store_path(URLPipeline.from_url("memory:x"), storage_options=storage_options) + + +class TestRootSchemeCase: + def test_uppercase_native_root(self, tmp_path: Path) -> None: + group = zarr.open_group(URLPipeline.from_url(f"FILE:{tmp_path.as_posix()}"), mode="w") + assert isinstance(group, zarr.Group) + + async def test_uppercase_fsspec_root(self) -> None: + """fsspec matches protocols case-sensitively, so the root gets the lowercased scheme.""" + pytest.importorskip("fsspec") + pytest.importorskip("aiohttp") + from zarr.storage import FsspecStore + + store_path = await make_store_path(URLPipeline.from_url("HTTPS://example.com/d")) + assert isinstance(store_path.store, FsspecStore) + assert store_path.store.path == "https://example.com/d" + + +class TestRegisterURLAdapter: + @pytest.mark.parametrize("scheme", ["", "a b", "1abc", "a_b", "zip:"]) + def test_invalid_scheme_rejected(self, scheme: str) -> None: + with pytest.raises(ValueError, match="not a valid URL scheme"): + register_url_adapter(scheme, RecordingAdapter) + + @pytest.mark.parametrize("scheme", ["file", "memory", "zarr", "zarr2", "ZARR3"]) + def test_reserved_scheme_rejected(self, scheme: str) -> None: + with pytest.raises(ValueError, match="handled by zarr-python itself"): + register_url_adapter(scheme, RecordingAdapter) + + def test_vendor_scheme_accepted(self) -> None: + register_url_adapter("Vendor.My-Scheme+x", RecordingAdapter) + assert zarr.registry._get_url_adapter("vendor.my-scheme+x") is RecordingAdapter + + @pytest.mark.parametrize("scheme", ["foo\n", "\u212aelvin"]) + def test_register_rejects_scheme_tricks(self, scheme: str) -> None: + with pytest.raises(ValueError, match="not a valid URL scheme"): + register_url_adapter(scheme, RecordingAdapter) + + +class TestSegmentValidation: + @pytest.mark.parametrize( + ("fields", "error"), + [ + ((1, "b", None, "r"), TypeError), + (("s", 2, None, "r"), TypeError), + (("s", "b", 3, "r"), TypeError), + (("s", "b", None, None), TypeError), + (("S3", "b", None, "r"), URLPipelineError), + (("", "b", None, "r"), URLPipelineError), + (("a b", "b", None, "r"), URLPipelineError), + ], + ) + def test_invalid_fields(self, fields: tuple[object, ...], error: type[Exception]) -> None: + with pytest.raises(error): + PipelineSegment(*fields) # type: ignore[arg-type] + + +class TestMessages: + def test_repr_is_the_url(self) -> None: + assert repr(URLPipeline.from_url("memory:|zarr3:a")) == "URLPipeline('memory:|zarr3:a')" + + def test_invalid_root_scheme_message(self) -> None: + with pytest.raises(URLPipelineError, match="invalid URL scheme 'a_b'"): + URLPipeline.from_url("a_b://x") + + @pytest.mark.parametrize("url", ["/a|b", "file:/a#b", " s3://b"]) + def test_no_percent_encoding_advice(self, url: str) -> None: + with pytest.raises(URLPipelineError) as info: + URLPipeline.from_url(url) + assert "%7C" not in str(info.value) + assert "%23" not in str(info.value) + assert "%20" not in str(info.value) + + @pytest.mark.skipif(sys.platform == "win32", reason="'|' is not allowed in Windows file names") + def test_group_not_found_message_is_readable(self, tmp_path: Path) -> None: + with pytest.raises(GroupNotFoundError) as info: + zarr.open_group(URLPipeline.from_url(f"file:{tmp_path.as_posix()}|zarr3:"), mode="r") + assert "PipelineSegment(" not in str(info.value) + + def test_no_adapter_error_does_not_list_installed_schemes( + self, recording: type[RecordingAdapter] + ) -> None: + with pytest.raises(URLPipelineError) as info: + zarr.core.sync.sync(make_store_path(URLPipeline.from_url("memory:|nope:"))) + assert "rec" not in str(info.value) + + +class TestSecretsAreRedacted: + URL = "https://user:pw@h/x?token=SECRET|rec:|zarr2:" + + def test_reprs(self) -> None: + pipeline = URLPipeline.from_url(self.URL) + context = zarr.storage.PipelineContext( + preceding=pipeline.segments[:1], mode="r", storage_options={"secret": "SECRET"} + ) + for text in (repr(pipeline), repr(pipeline.segments[0]), repr(context)): + assert "SECRET" not in text + assert "pw" not in text + assert str(pipeline) == self.URL # str() stays lossless + + @pytest.mark.parametrize( + "url", + [ + "https://h/x?token=SECRET| zip:", + "https://h/x?token=SECRET||zip:", + "https://h/x?token=SECRET#frag", + "https://h/x?token=SECRET|zarr3:|zip:", + "/x?token=SECRET", + "a_b://x?token=SECRET", + ], + ) + def test_parse_errors(self, url: str) -> None: + with pytest.raises(URLPipelineError) as info: + URLPipeline.from_url(url) + assert "SECRET" not in str(info.value) + + async def test_resolution_errors(self, recording: type[RecordingAdapter]) -> None: + with pytest.raises(ValueError, match="conflicts with") as conflict: + zarr.open_group(URLPipeline.from_url(self.URL), mode="w", zarr_format=3) + with pytest.raises(URLPipelineError) as residual: + await make_store(URLPipeline.from_url("memory:?token=SECRET|rec:sub")) + with pytest.raises(URLPipelineError) as root: + await make_store_path(URLPipeline.from_url("vendor.x://[a]/p?token=SECRET")) + for info in (conflict, residual, root): + assert "SECRET" not in str(info.value) + + +class TestEntryPointNames: + @pytest.mark.parametrize("name", ["file", "Memory", "zarr3", "a b", "x\n"]) + def test_unusable_entry_point_names_ignored( + self, name: str, monkeypatch: pytest.MonkeyPatch + ) -> None: + from importlib.metadata import EntryPoint, EntryPoints + + fake = EntryPoints( + [EntryPoint(name=name, value="nowhere:Adapter", group="zarr.url_adapters")] + ) + monkeypatch.setattr(zarr.registry, "get_entry_points", lambda: fake) + zarr.registry._url_adapter_registry.lazy_load_list.clear() + with pytest.warns(ZarrUserWarning, match="ignoring the 'zarr.url_adapters' entry point"): + zarr.registry._collect_entrypoints() + assert zarr.registry._url_adapter_registry.lazy_load_list == [] + + +class TestLimits: + def test_hand_built_pipeline_segment_limit(self) -> None: + segment = PipelineSegment("w", "", None, "w") + root = PipelineSegment("memory", "", None, "memory:") + URLPipeline((root, *[segment] * 31)) + with pytest.raises(URLPipelineError, match="1 to 32 segments"): + URLPipeline((root, *[segment] * 32)) + + +class TestWindowsFileRoots: + """Windows path rules, checked on any platform by patching `sys.platform`.""" + + @pytest.mark.parametrize( + "url", + [ + r"file:/\attacker\share\data", + r"file:///\attacker\share", + r"file://localhost/\attacker\share", + r"file:\\attacker\share", + ], + ) + async def test_unc_paths_rejected(self, url: str, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(sys, "platform", "win32") + with pytest.raises(URLPipelineError, match="authority|'//'|absolute local path"): + await make_store_path(URLPipeline.from_url(url), mode="r") + + @pytest.mark.parametrize( + "url", [r"file:C:x", r"file:/C:..\x", "file:C:/a/../b", "file:/C:/a:stream"] + ) + async def test_drive_relative_and_dot_paths_rejected( + self, url: str, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr(sys, "platform", "win32") + with pytest.raises(URLPipelineError, match=r"absolute local path|'\.' or '\.\.'|':'"): + await make_store_path(URLPipeline.from_url(url), mode="r") diff --git a/tests/test_url_pipeline/test_parser.py b/tests/test_url_pipeline/test_parser.py new file mode 100644 index 0000000000..195a27d13b --- /dev/null +++ b/tests/test_url_pipeline/test_parser.py @@ -0,0 +1,238 @@ +from __future__ import annotations + +import pytest +from hypothesis import given +from hypothesis import strategies as st + +from zarr.errors import URLPipelineError +from zarr.storage._url_pipeline import _parse_pipeline + + +def test_single_root_url() -> None: + (segment,) = _parse_pipeline("s3://bucket/key") + assert segment.scheme == "s3" + assert segment.body == "//bucket/key" + assert segment.query is None + assert segment.raw == "s3://bucket/key" + + +@pytest.mark.parametrize("url", ["/local/path", "/local/path|zip:", "relative/path|zip:"]) +def test_schemeless_root_rejected(url: str) -> None: + """The root sub-URL must carry a URL scheme; a bare path is an error.""" + with pytest.raises(URLPipelineError, match="URL scheme"): + _parse_pipeline(url) + + +def test_adapter_chain() -> None: + segments = _parse_pipeline("s3://bucket/data.zip|zip:inner/path|zarr3:") + assert [s.scheme for s in segments] == ["s3", "zip", "zarr3"] + assert segments[1].body == "inner/path" + assert segments[2].body == "" + + +def test_trailing_colon_optional() -> None: + with_colon = _parse_pipeline("file:/tmp/x.zip|zip:") + without_colon = _parse_pipeline("file:/tmp/x.zip|zip") + assert with_colon[1].scheme == without_colon[1].scheme == "zip" + assert with_colon[1].body == without_colon[1].body == "" + + +def test_scheme_case_insensitive() -> None: + segments = _parse_pipeline("FILE:/tmp/x.zip|ZIP:Inner/Path") + assert segments[0].scheme == "file" + assert segments[1].scheme == "zip" + # bodies are case-preserved + assert segments[1].body == "Inner/Path" + + +def test_case_preserved_in_raw() -> None: + # e.g. icechunk snapshot IDs are case-significant + segments = _parse_pipeline("file:/tmp/repo|icechunk://ABCDEFGH12345678ABCD/x") + assert segments[1].raw == "icechunk://ABCDEFGH12345678ABCD/x" + assert segments[1].body == "//ABCDEFGH12345678ABCD/x" + + +def test_vendor_prefixed_scheme() -> None: + segments = _parse_pipeline("file:/data|vendor-1.custom+adapter:sub/path") + assert segments[1].scheme == "vendor-1.custom+adapter" + assert segments[1].body == "sub/path" + + +def test_query_strings() -> None: + segments = _parse_pipeline("https://example.com/d.zip?token=abc|zip:x?opt=1") + assert segments[0].query == "token=abc" + assert segments[0].body == "//example.com/d.zip" + assert segments[1].query == "opt=1" + assert segments[1].body == "x" + + +def test_empty_query() -> None: + (segment,) = _parse_pipeline("https://example.com/d?") + assert segment.query == "" + + +@pytest.mark.parametrize( + "url", [r"C:\data\store", "C:/data/store|zip:", "c:\\x", "c:data", "C:..\\x", "x://host/p"] +) +def test_single_letter_root_scheme_rejected(url: str) -> None: + """A one-letter root scheme reads as a drive path on Windows, so it is rejected everywhere.""" + with pytest.raises(URLPipelineError, match="Windows drive path"): + _parse_pipeline(url) + + +@pytest.mark.parametrize("scheme", ["\u212aelvin", "s\u0131", "zi\u0440"]) +def test_non_ascii_scheme_rejected(scheme: str) -> None: + """Validation happens before lowercasing, so a look-alike cannot alias an ASCII scheme.""" + with pytest.raises(URLPipelineError, match="invalid URL scheme"): + _parse_pipeline(f"{scheme}://x") + with pytest.raises(URLPipelineError, match="invalid URL scheme"): + _parse_pipeline(f"s3://b|{scheme}:x") + + +def test_segment_limit() -> None: + _parse_pipeline("memory:" + "|w" * 31) + with pytest.raises(URLPipelineError, match="more than 32 segments"): + _parse_pipeline("memory:" + "|w" * 32) + + +@pytest.mark.parametrize("url", [None, 42, b"s3://b|zip:", ["s3://b"]]) +def test_non_string_rejected(url: object) -> None: + with pytest.raises(TypeError, match="must be a str"): + _parse_pipeline(url) # type: ignore[arg-type] + + +@pytest.mark.parametrize( + "url", + ["s3://b/\x00a", "s3://b/a\n|zip:", "s3://b|zip:\r\n", "s3://b|zip:\x7f", "s3://b\t|zip:"], +) +def test_control_characters_rejected(url: str) -> None: + with pytest.raises(URLPipelineError, match="control character"): + _parse_pipeline(url) + + +@pytest.mark.parametrize("url", [" s3://b", "s3://b |zip:", "s3://b| zip:", "s3://b|zip:x\u00a0"]) +def test_surrounding_whitespace_rejected(url: str) -> None: + with pytest.raises(URLPipelineError, match="leading or trailing whitespace"): + _parse_pipeline(url) + + +def test_interior_whitespace_is_preserved() -> None: + """zarr does not percent-decode, so a key with a space can only be spelled with one.""" + segments = _parse_pipeline("s3://b/My Data.zip|zip:a b") + assert segments[0].body == "//b/My Data.zip" + assert segments[1].body == "a b" + + +@pytest.mark.parametrize("url", ["a||b:", "|zip:", "file:/tmp|", ""]) +def test_empty_sub_url_rejected(url: str) -> None: + with pytest.raises(URLPipelineError, match="empty sub-URL"): + _parse_pipeline(url) + + +def test_fragment_rejected() -> None: + with pytest.raises(URLPipelineError, match="fragment"): + _parse_pipeline("file:/tmp/x.zip|zip:inner#frag") + + +@pytest.mark.parametrize("segment", ["1zip:", "zi p:x", "zip@:x"]) +def test_invalid_adapter_scheme_rejected(segment: str) -> None: + with pytest.raises(URLPipelineError, match="invalid URL scheme"): + _parse_pipeline(f"file:/tmp/x|{segment}") + + +def test_scheme_detection_ignores_authority_syntax() -> None: + """Scheme detection works on the raw text, so a malformed authority does not affect it.""" + segments = _parse_pipeline("xy://[authority]/path|zip:") + assert segments[0].scheme == "xy" + assert segments[0].raw == "xy://[authority]/path" + + +def test_round_trip() -> None: + url = "s3://bucket/a.zip?v=2|zip:b/inner.zip|zip:c|zarr3:" + segments = _parse_pipeline(url) + assert "|".join(s.raw for s in segments) == url + + +# --- property-based tests ----------------------------------------------- +# +# These check the parser's splitting invariants, not spec validity. Schemes are +# drawn from the full scheme grammar; bodies and queries from printable ASCII +# minus the characters the parser splits on ("|", "#", and "?" for bodies). + +_ADAPTER_SCHEME = st.from_regex(r"[a-zA-Z][a-zA-Z0-9+.\-]{0,15}", fullmatch=True) +# two or more characters, so Windows drive-letter handling cannot reclassify +# the root scheme as a local path on one platform but not another +_ROOT_SCHEME = st.from_regex(r"[a-zA-Z]{2}[a-zA-Z0-9+.\-]{0,14}", fullmatch=True) +# no trailing space: whitespace at either end of a sub-URL is rejected +_BODY = st.text( + st.characters(min_codepoint=32, max_codepoint=126, exclude_characters="|#?") +).filter(lambda t: not t.endswith(" ")) +_QUERY = st.text( + st.characters(min_codepoint=32, max_codepoint=126, exclude_characters="|#") +).filter(lambda t: not t.endswith(" ")) + + +@st.composite +def pipelines( + draw: st.DrawFn, min_depth: int = 1 +) -> tuple[str, list[tuple[str, str, str | None, str]]]: + """A parseable pipeline URL up to depth 8, with its expected split.""" + depth = draw(st.integers(min_value=min_depth, max_value=8)) + expected = [] + parts = [] + for index in range(depth): + scheme = draw(_ROOT_SCHEME if index == 0 else _ADAPTER_SCHEME) + body = draw(_BODY) + query = draw(st.none() | _QUERY) + raw = f"{scheme}:{body}" + (f"?{query}" if query is not None else "") + parts.append(raw) + expected.append((scheme.lower(), body, query, raw)) + return "|".join(parts), expected + + +@given(pipelines()) +def test_valid_pipeline_invariants( + case: tuple[str, list[tuple[str, str, str | None, str]]], +) -> None: + url, expected = case + segments = _parse_pipeline(url) + # schemes lowercased; bodies and queries preserved verbatim + assert [(s.scheme, s.body, s.query, s.raw) for s in segments] == expected + # lossless round trip + assert "|".join(s.raw for s in segments) == url + + +@given(pipelines(), st.data()) +def test_property_empty_sub_url_rejected( + case: tuple[str, list[tuple[str, str, str | None, str]]], data: st.DataObject +) -> None: + url, _ = case + parts = url.split("|") + parts.insert(data.draw(st.integers(0, len(parts))), "") + with pytest.raises(URLPipelineError, match="empty sub-URL"): + _parse_pipeline("|".join(parts)) + + +@given(pipelines(min_depth=2), st.data()) +def test_property_invalid_adapter_scheme_rejected( + case: tuple[str, list[tuple[str, str, str | None, str]]], data: st.DataObject +) -> None: + url, _ = case + parts = url.split("|") + # corrupt one adapter segment's scheme with a leading character the + # scheme grammar forbids + index = data.draw(st.integers(1, len(parts) - 1)) + parts[index] = data.draw(st.sampled_from(["1", "@", "~"])) + parts[index] + with pytest.raises(URLPipelineError, match="invalid URL scheme"): + _parse_pipeline("|".join(parts)) + + +@given(pipelines(), st.data()) +def test_property_fragment_rejected( + case: tuple[str, list[tuple[str, str, str | None, str]]], data: st.DataObject +) -> None: + url, _ = case + parts = url.split("|") + parts[data.draw(st.integers(0, len(parts) - 1))] += "#frag" + with pytest.raises(URLPipelineError, match="fragment"): + _parse_pipeline("|".join(parts)) diff --git a/tests/test_url_pipeline/test_pipeline_object.py b/tests/test_url_pipeline/test_pipeline_object.py new file mode 100644 index 0000000000..8db1d65744 --- /dev/null +++ b/tests/test_url_pipeline/test_pipeline_object.py @@ -0,0 +1,69 @@ +from __future__ import annotations + +import pytest + +from zarr.errors import URLPipelineError +from zarr.storage import PipelineSegment, URLPipeline +from zarr.storage._url_pipeline import _parse_pipeline + + +@pytest.mark.parametrize( + "url", + [ + "s3://bucket/data.zip|zip:inner|zarr3:", + "file:///data/node.zarr|zarr2", + "file:///repo|icechunk:|zarr3:/path/to/array/", + "memory://x|wrap:a", + "s3://bucket/key", + ], +) +def test_from_url(url: str) -> None: + """`from_url` holds every parsed segment; `str()` reproduces the URL exactly.""" + pipeline = URLPipeline.from_url(url) + assert pipeline.segments == _parse_pipeline(url) + assert str(pipeline) == url + assert URLPipeline(pipeline.segments) == pipeline + + +def test_format_segment_rejects_query() -> None: + """A `zarr2:`/`zarr3:` segment takes a node path only; a query is an error.""" + with pytest.raises(URLPipelineError, match="do not accept a query"): + URLPipeline.from_url("file:/data|zarr3:x?opt=1") + + +def test_direct_construction_is_validated() -> None: + """Validation lives in the constructor, so a hand-built pipeline is checked too.""" + segments = _parse_pipeline("file:/data|zarr3:|zip:") + with pytest.raises(URLPipelineError, match="must be the last segment"): + URLPipeline(segments) + + +def test_empty_pipeline_rejected() -> None: + """A pipeline needs at least a root segment.""" + with pytest.raises(URLPipelineError, match="1 to 32 segments"): + URLPipeline(()) + + +@pytest.mark.parametrize( + "segments", + ["s3://b|zip:", ["s3://b"], ("s3://b",), [PipelineSegment("s3", "//b", None, "s3://b")]], +) +def test_direct_construction_rejects_non_segments(segments: object) -> None: + """Only a tuple of `PipelineSegment` is accepted; `from_url` parses strings.""" + with pytest.raises(TypeError, match="tuple of PipelineSegment"): + URLPipeline(segments) # type: ignore[arg-type] + + +@pytest.mark.parametrize("url", [None, b"s3://b", URLPipeline.from_url("s3://b")]) +def test_from_url_rejects_non_strings(url: object) -> None: + with pytest.raises(TypeError, match="must be a str"): + URLPipeline.from_url(url) # type: ignore[arg-type] + + +def test_hashable() -> None: + """Pipelines are values: equal pipelines hash equal and can key a dict.""" + a = URLPipeline.from_url("s3://b/d.zip|zip:|zarr3:") + b = URLPipeline.from_url("s3://b/d.zip|zip:|zarr3:") + assert a == b + assert hash(a) == hash(b) + assert {a: 1}[b] == 1 diff --git a/tests/test_url_pipeline/test_resolver.py b/tests/test_url_pipeline/test_resolver.py new file mode 100644 index 0000000000..c5c2666737 --- /dev/null +++ b/tests/test_url_pipeline/test_resolver.py @@ -0,0 +1,928 @@ +from __future__ import annotations + +import contextlib +import dataclasses +import sys +import threading +import warnings +from importlib.metadata import EntryPoint, version +from typing import TYPE_CHECKING, ClassVar + +import numpy as np +import pytest +from packaging.version import parse as parse_version + +import zarr +import zarr.registry +from zarr.abc.store import Store +from zarr.abc.url_pipeline import URLPipelineAdapter +from zarr.errors import URLPipelineError, ZarrUserWarning +from zarr.registry import ( + _get_url_adapter, + _list_url_adapter_schemes, + register_url_adapter, +) +from zarr.storage import ( + LocalStore, + ManagedMemoryStore, + MemoryStore, + PipelineContext, + PipelineSegment, + StorePath, + URLPipeline, + WrapperStore, +) +from zarr.storage._common import _has_fsspec, make_store, make_store_path +from zarr.storage._utils import _join_paths + +if TYPE_CHECKING: + from collections.abc import Callable + from pathlib import Path + + from zarr.core.common import AccessModeLiteral + +# fsspec < 2024.12.0 has no AsyncFileSystemWrapper, so a plain memory:// URL +# (a sync filesystem) cannot be opened at all there +_has_async_fsspec_wrapper = _has_fsspec and parse_version(version("fsspec")) >= parse_version( + "2024.12.0" +) + +pytestmark = pytest.mark.usefixtures("clean_url_adapter_registry") + + +class TracingStore(WrapperStore[Store]): + """Wrapper that records the context and segment it was created from.""" + + context: PipelineContext + segment: PipelineSegment + + +class WrapperAdapter(URLPipelineAdapter): + """ + A wrapper-style adapter: resolves the preceding pipeline into a store. + + Follows the wrapper contract: the preceding residual path is carried + forward, joined with this segment's own path. + """ + + @classmethod + async def open_pipeline_segment( + cls, segment: PipelineSegment, context: PipelineContext + ) -> StorePath: + preceding = await context.resolve_preceding() + store = TracingStore(preceding.store) + store.context = context + store.segment = segment + return StorePath(store, _join_paths([preceding.path, segment.body])) + + +class NativeAdapter(URLPipelineAdapter): + """A native-style adapter: consumes the preceding URL as a string.""" + + seen_urls: ClassVar[list[str]] = [] + + @classmethod + async def open_pipeline_segment( + cls, segment: PipelineSegment, context: PipelineContext + ) -> StorePath: + cls.seen_urls.append(context.preceding_url) + store = await MemoryStore.open(read_only=context.read_only) + return StorePath(store, path=segment.body) + + +class RootAdapter(URLPipelineAdapter): + """A read-only root-scheme adapter (no preceding segments).""" + + @classmethod + async def open_pipeline_segment( + cls, segment: PipelineSegment, context: PipelineContext + ) -> StorePath: + assert context.preceding == () + if context.mode in ("w", "w-", "r+"): + raise ValueError("read-only scheme") + store = await MemoryStore.open(read_only=True) + return StorePath(store, path=segment.body.lstrip("/")) + + +class DisobedientAdapter(URLPipelineAdapter): + """An adapter that ignores `context.read_only` (a contract violation).""" + + @classmethod + async def open_pipeline_segment( + cls, segment: PipelineSegment, context: PipelineContext + ) -> StorePath: + return StorePath(await MemoryStore.open(read_only=False)) + + +class TestRegistry: + def test_register_and_get(self) -> None: + register_url_adapter("demo", WrapperAdapter) + assert _get_url_adapter("demo") is WrapperAdapter + assert _get_url_adapter("DEMO") is WrapperAdapter + assert "demo" in _list_url_adapter_schemes() + + def test_unknown_scheme(self) -> None: + with pytest.raises(URLPipelineError, match="no URL pipeline adapter is registered"): + _get_url_adapter("nonexistent-scheme") + + def test_reregistering_scheme_warns(self) -> None: + register_url_adapter("demo", WrapperAdapter) + with pytest.warns(ZarrUserWarning, match="is being replaced"): + register_url_adapter("demo", NativeAdapter) + assert _get_url_adapter("demo") is NativeAdapter + # re-registering the same class is not a collision + register_url_adapter("demo", NativeAdapter) + + @pytest.mark.usefixtures("set_path") + def test_entrypoint_discovery(self) -> None: + assert "example-pkg.entrypoint-scheme" in _list_url_adapter_schemes() + cls = _get_url_adapter("example-pkg.entrypoint-scheme") + assert cls.__name__ == "TestEntrypointURLAdapter" + + @pytest.mark.usefixtures("set_path") + async def test_entrypoint_end_to_end(self) -> None: + result = await URLPipeline.from_url( + "memory://src|example-pkg.entrypoint-scheme:sub/path" + )._resolve_store_path() + assert result.path == "sub/path" + + @pytest.mark.usefixtures("set_path") + def test_entrypoint_scheme_lookup_is_case_insensitive(self) -> None: + # entry-point names are matched case-insensitively, like schemes + cls = _get_url_adapter("Example-PKG.Entrypoint-Scheme") + assert cls.__name__ == "TestEntrypointURLAdapter" + + @pytest.mark.usefixtures("set_path") + def test_loading_one_scheme_leaves_others_pending(self) -> None: + # resolving one scheme must not import other providers' entry points + registry = zarr.registry._url_adapter_registry + assert any(e.name == "example-pkg.entrypoint-scheme" for e in registry.lazy_load_list) + with pytest.raises(URLPipelineError, match="no URL pipeline adapter"): + _get_url_adapter("some-other-scheme") + assert any(e.name == "example-pkg.entrypoint-scheme" for e in registry.lazy_load_list) + assert _get_url_adapter("example-pkg.entrypoint-scheme").__name__ == ( + "TestEntrypointURLAdapter" + ) + + +def _fake_entry_point( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, *, scheme: str, module: str, source: str +) -> EntryPoint: + """ + Write `source` to an importable module and return a `zarr.url_adapters` + entry point for `scheme` pointing at its `Adapter` attribute. + """ + (tmp_path / f"{module}.py").write_text(source) + monkeypatch.syspath_prepend(str(tmp_path)) + monkeypatch.delitem(sys.modules, module, raising=False) + return EntryPoint(name=scheme, value=f"{module}:Adapter", group="zarr.url_adapters") + + +_ADAPTER_SOURCE = """ +from zarr.abc.url_pipeline import URLPipelineAdapter +from zarr.storage import MemoryStore, StorePath +{prelude} + +class Adapter(URLPipelineAdapter): + @classmethod + async def open_pipeline_segment(cls, segment, context): + return StorePath(await MemoryStore.open(read_only=context.read_only)) +""" + + +class TestEntryPointLoading: + """Lazy loading of `zarr.url_adapters` entry points.""" + + def test_adapter_import_may_resolve_other_adapters( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + # an adapter module that itself resolves a URL adapter at import + # time must not deadlock on the registry lock + pending = zarr.registry._url_adapter_registry.lazy_load_list + pending.append( + _fake_entry_point( + tmp_path, + monkeypatch, + scheme="reentrant", + module="reentrant_adapter", + source=_ADAPTER_SOURCE.format( + prelude=( + "import zarr.registry\n" + "try:\n" + " zarr.registry._get_url_adapter('no-such-scheme-at-import')\n" + "except Exception:\n" + " pass\n" + ) + ), + ) + ) + result: list[object] = [] + thread = threading.Thread(target=lambda: result.append(_get_url_adapter("reentrant"))) + thread.start() + thread.join(timeout=10) + assert not thread.is_alive(), "_get_url_adapter deadlocked on a re-entrant lookup" + assert getattr(result[0], "__name__", None) == "Adapter" + + def test_concurrent_loads_keep_other_schemes_pending( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + # two threads resolving different schemes at once, one with a slow + # import, must not drop a third, unrelated pending entry point + pending = zarr.registry._url_adapter_registry.lazy_load_list + for scheme, prelude in [ + ("slow", "import time\ntime.sleep(0.3)\n"), + ("quick", ""), + ("untouched", ""), + ]: + pending.append( + _fake_entry_point( + tmp_path, + monkeypatch, + scheme=scheme, + module=f"{scheme}_adapter", + source=_ADAPTER_SOURCE.format(prelude=prelude), + ) + ) + threads = [ + threading.Thread(target=_get_url_adapter, args=(scheme,)) + for scheme in ("slow", "quick") + ] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=10) + assert not any(thread.is_alive() for thread in threads) + assert "untouched" in _list_url_adapter_schemes() + assert [e.name for e in pending] == ["untouched"] + assert _get_url_adapter("untouched").__name__ == "Adapter" + + def test_import_failure_is_wrapped_and_not_permanent(self) -> None: + pending = zarr.registry._url_adapter_registry.lazy_load_list + pending.append( + EntryPoint(name="broken", value="no_such_module_xyz:Adapter", group="zarr.url_adapters") + ) + with pytest.raises(URLPipelineError, match="could not be loaded"): + _get_url_adapter("broken") + # still advertised, so a transient failure can be retried + assert "broken" in _list_url_adapter_schemes() + with pytest.raises(URLPipelineError, match="could not be loaded"): + _get_url_adapter("broken") + + def test_registered_adapter_shadows_entry_point_with_warning( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + # a builtin registered under a scheme wins over a same-named entry + # point, which is discarded loudly rather than left pending forever + pending = zarr.registry._url_adapter_registry.lazy_load_list + pending.append( + _fake_entry_point( + tmp_path, + monkeypatch, + scheme="shadowed", + module="shadowed_adapter", + source=_ADAPTER_SOURCE.format(prelude="raise RuntimeError('must not import')\n"), + ) + ) + register_url_adapter("shadowed", NativeAdapter) + with pytest.warns(ZarrUserWarning, match="already registered"): + assert _get_url_adapter("shadowed") is NativeAdapter + assert not any(e.name == "shadowed" for e in pending) + # a second lookup is silent + with warnings.catch_warnings(): + warnings.simplefilter("error") + assert _get_url_adapter("shadowed") is NativeAdapter + + def test_duplicate_entry_points_warn_and_first_wins( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + pending = zarr.registry._url_adapter_registry.lazy_load_list + for module in ("dup_first", "dup_second"): + pending.append( + _fake_entry_point( + tmp_path, + monkeypatch, + scheme="dup", + module=module, + source=_ADAPTER_SOURCE.format(prelude=f"ORIGIN = {module!r}\n"), + ) + ) + with pytest.warns(ZarrUserWarning, match="multiple 'zarr.url_adapters' entry points"): + cls = _get_url_adapter("dup") + assert getattr(sys.modules[cls.__module__], "ORIGIN", None) == "dup_first" + assert not any(e.name == "dup" for e in pending) + + +class TestStringsAreNeverPipelines: + """ + Only a `URLPipeline` object is resolved as a pipeline. A plain string + keeps the meaning it had before URL pipelines existed, whatever adapters + are registered. + """ + + @pytest.mark.skipif(sys.platform == "win32", reason="'|' is not allowed in Windows file names") + async def test_local_path_containing_pipe_is_a_local_store(self, tmp_path: Path) -> None: + """A string path with `|` and a registered adapter's name is still a local directory.""" + register_url_adapter("wrap", WrapperAdapter) + store_path = await make_store_path(f"{tmp_path}/a|wrap:x", mode="w") + assert isinstance(store_path.store, LocalStore) + assert store_path.store.root == tmp_path / "a|wrap:x" + + async def test_registered_root_scheme_string_is_not_routed(self) -> None: + """A URL whose scheme names a registered root adapter is not handed to that adapter.""" + calls: list[PipelineSegment] = [] + + class CountingRootAdapter(RootAdapter): + @classmethod + async def open_pipeline_segment( + cls, segment: PipelineSegment, context: PipelineContext + ) -> StorePath: + calls.append(segment) + return await super().open_pipeline_segment(segment, context) + + register_url_adapter("rooty", CountingRootAdapter) + # the string falls through to the fsspec route, which may raise for + # an unknown protocol; whatever happens, the adapter is not consulted + with contextlib.suppress(Exception): + await make_store_path("rooty://org/repo") + assert calls == [] + + async def test_native_schemes_do_not_route_to_adapters(self) -> None: + """Even an adapter forced into the registry never handles `file:`/`memory:`.""" + zarr.registry._url_adapter_registry.register(RootAdapter, "file") + zarr.registry._url_adapter_registry.register(RootAdapter, "memory") + store = await make_store(URLPipeline.from_url("memory://x")) + assert isinstance(store, ManagedMemoryStore) + for url in ("memory:|file:/etc", "memory:|memory:x"): + with pytest.raises(URLPipelineError, match="cannot be an adapter segment"): + await make_store_path(URLPipeline.from_url(url)) + + +class TestResolve: + async def test_wrapper_adapter_chain(self, tmp_path: Path) -> None: + register_url_adapter("wrap", WrapperAdapter) + result = await URLPipeline.from_url( + f"file:{tmp_path.as_posix()}|wrap:inner/path" + )._resolve_store_path() + assert isinstance(result.store, TracingStore) + assert result.path == "inner/path" + assert result.store.context.preceding_url == f"file:{tmp_path.as_posix()}" + assert result.store.segment.scheme == "wrap" + assert result.store.segment.body == "inner/path" + + async def test_nested_wrappers_preserve_residual_paths(self, tmp_path: Path) -> None: + # each wrapper joins the preceding residual path with its own, so + # no segment's path is lost in root|wrap:a|wrap:b + register_url_adapter("wrap", WrapperAdapter) + result = await URLPipeline.from_url( + f"file:{tmp_path.as_posix()}|wrap:a|wrap:b" + )._resolve_store_path() + assert result.path == "a/b" + + async def test_native_adapter_gets_preceding_url(self) -> None: + register_url_adapter("native", NativeAdapter) + NativeAdapter.seen_urls.clear() + await URLPipeline.from_url("s3://bucket/repo|native:")._resolve_store_path() + assert NativeAdapter.seen_urls == ["s3://bucket/repo"] + + async def test_multi_segment_preceding_url(self) -> None: + register_url_adapter("wrap", WrapperAdapter) + register_url_adapter("native", NativeAdapter) + NativeAdapter.seen_urls.clear() + await URLPipeline.from_url("memory://base|wrap:a|native:x")._resolve_store_path() + assert NativeAdapter.seen_urls == ["memory://base|wrap:a"] + + async def test_root_adapter(self) -> None: + register_url_adapter("rooty", RootAdapter) + result = await URLPipeline.from_url("rooty://org/repo")._resolve_store_path() + assert result.path == "org/repo" + assert result.store.read_only + + async def test_root_adapter_composes_with_chain(self) -> None: + register_url_adapter("rooty", RootAdapter) + register_url_adapter("native", NativeAdapter) + NativeAdapter.seen_urls.clear() + await URLPipeline.from_url("rooty://org/repo|native:x")._resolve_store_path() + assert NativeAdapter.seen_urls == ["rooty://org/repo"] + + async def test_read_only_flag(self, tmp_path: Path) -> None: + register_url_adapter("wrap", WrapperAdapter) + result = await URLPipeline.from_url( + f"file:{tmp_path.as_posix()}|wrap:" + )._resolve_store_path(mode="r") + assert isinstance(result.store, TracingStore) + assert result.store.context.read_only + assert result.store.context.mode == "r" + result = await URLPipeline.from_url( + f"file:{tmp_path.as_posix()}|wrap:" + )._resolve_store_path() + assert isinstance(result.store, TracingStore) + assert not result.store.context.read_only + assert result.store.context.mode is None + + async def test_read_only_is_enforced_on_disobedient_adapters(self) -> None: + # the resolver downgrades a writable store returned under mode "r" + # instead of trusting the adapter to have honored context.read_only + register_url_adapter("bad", DisobedientAdapter) + result = await URLPipeline.from_url("memory://base|bad:")._resolve_store_path(mode="r") + assert result.store.read_only + store = await make_store(URLPipeline.from_url("memory://base|bad:"), mode="r") + assert store.read_only + + async def test_read_only_enforcement_without_conversion_raises(self) -> None: + # a disobedient adapter whose store cannot be converted read-only + class StubbornStore(MemoryStore): + def with_read_only(self, read_only: bool = False) -> MemoryStore: + raise NotImplementedError + + class StubbornAdapter(URLPipelineAdapter): + @classmethod + async def open_pipeline_segment( + cls, segment: PipelineSegment, context: PipelineContext + ) -> StorePath: + return StorePath(await StubbornStore.open(read_only=False)) + + register_url_adapter("stubborn", StubbornAdapter) + with pytest.raises(URLPipelineError, match="does not support read-only conversion"): + await URLPipeline.from_url("memory://base|stubborn:")._resolve_store_path(mode="r") + + async def test_read_only_enforcement_failure_closes_store(self) -> None: + # the adapter's store must not leak when it cannot be made read-only + closed: list[bool] = [] + + class StubbornStore(MemoryStore): + def with_read_only(self, read_only: bool = False) -> MemoryStore: + raise NotImplementedError + + def close(self) -> None: + closed.append(True) + super().close() + + class StubbornAdapter(URLPipelineAdapter): + @classmethod + async def open_pipeline_segment( + cls, segment: PipelineSegment, context: PipelineContext + ) -> StorePath: + return StorePath(await StubbornStore.open(read_only=False)) + + register_url_adapter("stubborn", StubbornAdapter) + with pytest.raises(URLPipelineError): + await URLPipeline.from_url("memory://base|stubborn:")._resolve_store_path(mode="r") + assert closed == [True] + + async def test_resolve_preceding_mode_override(self, tmp_path: Path) -> None: + # a wrapper that only reads the preceding resource can open it + # read-only regardless of the caller's mode + class ReadOnlyRootWrapper(WrapperAdapter): + @classmethod + async def open_pipeline_segment( + cls, segment: PipelineSegment, context: PipelineContext + ) -> StorePath: + preceding = await dataclasses.replace(context, mode="r").resolve_preceding() + assert preceding.store.read_only + return StorePath(preceding.store, segment.body) + + register_url_adapter("rowrap", ReadOnlyRootWrapper) + result = await URLPipeline.from_url( + f"file:{tmp_path.as_posix()}|rowrap:x" + )._resolve_store_path(mode="w") + assert result.path == "x" + + async def test_wrapper_on_local_file_root_read_only(self, tmp_path: Path) -> None: + # a wrapper that opens the preceding local *file* read-only gets a + # store rooted at that file (the mechanism a read-only zip: adapter + # can build on) + target = tmp_path / "data.bin" + target.write_bytes(b"payload") + register_url_adapter("wrap", WrapperAdapter) + result = await URLPipeline.from_url(f"file:{target.as_posix()}|wrap:")._resolve_store_path( + mode="r" + ) + assert isinstance(result.store, TracingStore) + assert result.store.read_only + + @pytest.mark.parametrize("mode", [None, "a", "w", "r+", "w-"]) + @pytest.mark.xfail( + strict=True, + raises=FileExistsError, + reason=( + "known limitation: there is no file-resource primitive yet, so resolving a " + "local file root in a writable mode hits LocalStore's create-on-open mkdir" + ), + ) + async def test_wrapper_on_local_file_root_writable_modes( + self, tmp_path: Path, mode: AccessModeLiteral | None + ) -> None: + target = tmp_path / "data.bin" + target.write_bytes(b"payload") + register_url_adapter("wrap", WrapperAdapter) + await URLPipeline.from_url(f"file:{target.as_posix()}|wrap:")._resolve_store_path(mode=mode) + + async def test_resolve_preceding_storage_options_override(self, tmp_path: Path) -> None: + # an adapter that consumed its namespaced keys strips them before + # the root store ever sees them + class ConsumingWrapper(WrapperAdapter): + seen: ClassVar[object | None] = None + + @classmethod + async def open_pipeline_segment( + cls, segment: PipelineSegment, context: PipelineContext + ) -> StorePath: + options = dict(context.storage_options or {}) + cls.seen = options.pop("consuming_secret", None) + preceding = await dataclasses.replace( + context, storage_options=options or None + ).resolve_preceding() + return StorePath(preceding.store, segment.body) + + register_url_adapter("consuming", ConsumingWrapper) + # the local-path root rejects any surviving storage_options, so this + # passing proves the adapter's keys were stripped before resolution + result = await URLPipeline.from_url( + f"file:{tmp_path.as_posix()}|consuming:x" + )._resolve_store_path( + storage_options={"consuming_secret": "s3cr3t"}, + ) + assert result.path == "x" + assert ConsumingWrapper.seen == "s3cr3t" + + async def test_storage_options_visible_to_adapter(self) -> None: + class OptionsProbe(NativeAdapter): + seen_options: dict[str, object] | None = None + + @classmethod + async def open_pipeline_segment( + cls, segment: PipelineSegment, context: PipelineContext + ) -> StorePath: + cls.seen_options = context.storage_options + return await super().open_pipeline_segment(segment, context) + + register_url_adapter("probe", OptionsProbe) + opts = {"anon": True} + await URLPipeline.from_url("s3://bucket/x|probe:")._resolve_store_path(storage_options=opts) + assert OptionsProbe.seen_options == opts + + async def test_wrapper_at_root_position_raises(self) -> None: + # a wrapper adapter used as the pipeline root has nothing to wrap + register_url_adapter("wrap", WrapperAdapter) + with pytest.raises(URLPipelineError, match="no preceding sub-URL"): + await URLPipeline.from_url("wrap:whatever")._resolve_store_path() + + async def test_single_url_is_a_trivial_pipeline(self) -> None: + """A URL without `|` resolves as a pipeline consisting of its root alone.""" + result = await URLPipeline.from_url("memory://plain")._resolve_store_path() + assert isinstance(result.store, ManagedMemoryStore) + assert result.path == "" + + async def test_format_segment_path_joins_residual_path(self, tmp_path: Path) -> None: + """The node path of a trailing format segment extends the adapters' residual path.""" + register_url_adapter("wrap", WrapperAdapter) + result = await URLPipeline.from_url( + f"file:{tmp_path.as_posix()}|wrap:a|zarr3:b" + )._resolve_store_path() + assert result.path == "a/b" + + async def test_unknown_adapter_scheme_raises(self) -> None: + with pytest.raises(URLPipelineError, match="no URL pipeline adapter is registered"): + await URLPipeline.from_url("memory://base|no-such-adapter:")._resolve_store_path() + + async def test_root_value_error_is_wrapped(self) -> None: + # a ValueError from the root store is re-raised as URLPipelineError + register_url_adapter("wrap", WrapperAdapter) + with pytest.raises(URLPipelineError, match="could not resolve the pipeline root"): + await URLPipeline.from_url("vendor.x://[authority]/path|wrap:")._resolve_store_path() + + +class TestSpecRoots: + """The spec's `memory:` / `file:` root semantics inside pipelines.""" + + @pytest.mark.parametrize("root", ["memory:", "memory:/", "memory://"]) + async def test_memory_root_spellings_equivalent(self, root: str) -> None: + register_url_adapter("wrap", WrapperAdapter) + result = await URLPipeline.from_url(f"{root}|wrap:")._resolve_store_path() + assert isinstance(result.store, TracingStore) + inner = result.store._store + assert isinstance(inner, ManagedMemoryStore) + assert inner._name == "" + assert inner.path == "" + + @pytest.mark.parametrize("root", ["memory:a/b", "memory:/a/b", "memory://a/b"]) + async def test_memory_root_path_spellings_equivalent(self, root: str) -> None: + """There is one in-memory store; the whole body is a path within it.""" + register_url_adapter("wrap", WrapperAdapter) + result = await URLPipeline.from_url(f"{root}|wrap:")._resolve_store_path() + assert isinstance(result.store, TracingStore) + inner = result.store._store + assert isinstance(inner, ManagedMemoryStore) + assert inner._name == "" + assert inner.path == "a/b" + + async def test_memory_root_rejects_query(self) -> None: + register_url_adapter("wrap", WrapperAdapter) + with pytest.raises(URLPipelineError, match="do not accept a query"): + await URLPipeline.from_url("memory:a?opt=1|wrap:")._resolve_store_path() + + async def test_file_root_localhost_authority(self, tmp_path: Path) -> None: + # file://localhost/p and file:///p are equivalent to file:/p + register_url_adapter("wrap", WrapperAdapter) + # spell the path URL-style: on Windows C:/... gains a leading slash + posix = tmp_path.as_posix() + url_path = posix if posix.startswith("/") else f"/{posix}" + zarr.open_group(URLPipeline.from_url(f"file://localhost{url_path}|wrap:"), mode="w") + opened = zarr.open_group(URLPipeline.from_url(f"file:{tmp_path}|wrap:"), mode="r") + assert isinstance(opened, zarr.Group) + zarr.open_group(URLPipeline.from_url(f"file://{url_path}|wrap:"), mode="r") + + @pytest.mark.skipif(sys.platform != "win32", reason="drive paths are Windows-only") + async def test_file_root_windows_drive_spellings(self, tmp_path: Path) -> None: + # on Windows, drive paths count as absolute, and the URL spelling + # file:///C:/... has its leading slash stripped before the drive + register_url_adapter("wrap", WrapperAdapter) + drive_path = tmp_path.as_posix() # C:/Users/... + r1 = await URLPipeline.from_url(f"file:{drive_path}|wrap:")._resolve_store_path() + r2 = await URLPipeline.from_url(f"file:///{drive_path}|wrap:")._resolve_store_path() + assert isinstance(r1.store, TracingStore) + assert isinstance(r2.store, TracingStore) + + @pytest.mark.skipif(sys.platform == "win32", reason="POSIX path semantics") + async def test_file_root_drive_letter_is_not_special_on_posix(self) -> None: + """Off Windows, `/C:/x` is an ordinary absolute path and `C:/x` is relative.""" + with pytest.raises(URLPipelineError, match="absolute local path"): + await make_store_path(URLPipeline.from_url("file:C:/drive/data"), mode="r") + with pytest.raises(FileNotFoundError, match="/C:/drive/data"): + await make_store_path(URLPipeline.from_url("file:///C:/drive/data"), mode="r") + + async def test_file_root_rejects_other_authority(self) -> None: + register_url_adapter("wrap", WrapperAdapter) + with pytest.raises(URLPipelineError, match="unsupported authority"): + await URLPipeline.from_url("file://example.com/tmp/x|wrap:")._resolve_store_path() + + @pytest.mark.parametrize("root", ["file:relative/path", "file://localhost"]) + async def test_file_root_rejects_relative_paths(self, root: str) -> None: + register_url_adapter("wrap", WrapperAdapter) + with pytest.raises(URLPipelineError, match="absolute local path"): + await URLPipeline.from_url(f"{root}|wrap:")._resolve_store_path() + + async def test_file_root_rejects_query(self, tmp_path: Path) -> None: + register_url_adapter("wrap", WrapperAdapter) + with pytest.raises(URLPipelineError, match="do not accept a query"): + await URLPipeline.from_url(f"file:{tmp_path}?v=1|wrap:")._resolve_store_path() + + @pytest.mark.skipif( + not _has_async_fsspec_wrapper, + reason="plain memory:// routes to fsspec only with fsspec>=2024.12.0", + ) + async def test_memory_root_is_distinct_from_fsspec_memory_url(self) -> None: + # known divergence: a plain memory:// URL is fsspec's in-memory + # filesystem when fsspec is installed, while a pipeline's memory: root + # is zarr's ManagedMemoryStore; the two do not share data + register_url_adapter("wrap", WrapperAdapter) + plain = await make_store("memory://distinct-store") + piped = await make_store(URLPipeline.from_url("memory://distinct-store|wrap:")) + assert not isinstance(plain, ManagedMemoryStore) + assert isinstance(piped, TracingStore) + assert isinstance(piped._store, ManagedMemoryStore) + + async def test_file_root_does_not_percent_decode(self, tmp_path: Path) -> None: + # documented divergence from RFC 8089: consistent with LocalStore, no + # percent-escape is decoded, so %20 names a literal directory + register_url_adapter("wrap", WrapperAdapter) + await URLPipeline.from_url(f"file:{tmp_path.as_posix()}/a%20b|wrap:")._resolve_store_path() + assert (tmp_path / "a%20b").is_dir() + assert not (tmp_path / "a b").exists() + + +class TestMakeStoreIntegration: + """`make_store` / `make_store_path` accept a `URLPipeline` like any other StoreLike.""" + + async def test_make_store_path_combines_paths(self, tmp_path: Path) -> None: + """Adapter residual path, format-segment node path and the caller's path join in order.""" + register_url_adapter("wrap", WrapperAdapter) + pipeline = URLPipeline.from_url(f"file:{tmp_path.as_posix()}|wrap:residual|zarr3:node") + store_path = await make_store_path(pipeline, path="user/sub") + assert store_path.path == "residual/node/user/sub" + assert isinstance(store_path.store, TracingStore) + + async def test_make_store_rejects_residual_path(self, tmp_path: Path) -> None: + register_url_adapter("wrap", WrapperAdapter) + with pytest.raises(URLPipelineError, match="resolves to a path inside a store"): + await make_store(URLPipeline.from_url(f"file:{tmp_path.as_posix()}|wrap:residual")) + + async def test_make_store_rejects_format_segment_node_path(self, tmp_path: Path) -> None: + """A node path given on the format segment also addresses a path inside the store.""" + register_url_adapter("wrap", WrapperAdapter) + with pytest.raises(URLPipelineError, match="resolves to a path inside a store"): + await make_store(URLPipeline.from_url(f"file:{tmp_path.as_posix()}|wrap:|zarr3:node")) + + async def test_make_store_accepts_slash_root_path(self, tmp_path: Path) -> None: + # an adapter returning path="/" addresses the store root; the raw + # value is normalized before the residual-path rejection + class SlashRootAdapter(WrapperAdapter): + @classmethod + async def open_pipeline_segment( + cls, segment: PipelineSegment, context: PipelineContext + ) -> StorePath: + preceding = await context.resolve_preceding() + return StorePath(preceding.store, "/") + + register_url_adapter("slashy", SlashRootAdapter) + store = await make_store(URLPipeline.from_url(f"file:{tmp_path.as_posix()}|slashy:")) + assert store is not None + + async def test_make_store_no_residual_path(self, tmp_path: Path) -> None: + register_url_adapter("wrap", WrapperAdapter) + store = await make_store(URLPipeline.from_url(f"file:{tmp_path.as_posix()}|wrap:")) + assert isinstance(store, TracingStore) + + async def test_make_store_rejects_invalid_mode_before_resolving(self, tmp_path: Path) -> None: + # an invalid mode is rejected upfront, like every other StoreLike, + # rather than being handed to adapters + register_url_adapter("wrap", WrapperAdapter) + with pytest.raises(ValueError, match="Invalid mode"): + await make_store( + URLPipeline.from_url(f"file:{tmp_path.as_posix()}|wrap:"), + mode="invalid", # type: ignore[arg-type] + ) + + async def test_zarr_open_end_to_end(self, tmp_path: Path) -> None: + register_url_adapter("wrap", WrapperAdapter) + group = zarr.open_group(URLPipeline.from_url(f"file:{tmp_path.as_posix()}|wrap:"), mode="w") + array = group.create_array("x", shape=(4,), dtype="i4") + array[:] = [1, 2, 3, 4] + assert zarr.open_array(URLPipeline.from_url(f"file:{tmp_path.as_posix()}|wrap:x"))[2] == 3 + + async def test_pipeline_store_read_only_mode(self) -> None: + register_url_adapter("native", NativeAdapter) + store_path = await make_store_path(URLPipeline.from_url("memory://base|native:"), mode="r") + assert store_path.read_only + + async def test_mode_a_on_read_only_store_raises(self) -> None: + """Mode "a" on a read-only adapter store raises, as for any read-only store.""" + register_url_adapter("rooty", RootAdapter) + with pytest.raises(ValueError, match="Store is read-only but mode is 'a'"): + await make_store_path(URLPipeline.from_url("rooty://org/repo"), mode="a") + + async def test_explicit_write_mode_reaches_adapter(self) -> None: + register_url_adapter("rooty", RootAdapter) + with pytest.raises(ValueError, match="read-only scheme"): + await make_store_path(URLPipeline.from_url("rooty://org/repo"), mode="w") + + async def test_storage_options_forwarded_to_root(self) -> None: + # storage_options reach the root sub-URL via resolve_preceding -> + # make_store. A local-path root does not accept storage_options, so + # forwarding them must raise the same TypeError as a non-pipeline open. + register_url_adapter("wrap", WrapperAdapter) + with pytest.raises(TypeError, match="'storage_options' was provided but unused"): + await make_store( + URLPipeline.from_url("file:/tmp/some/path|wrap:"), storage_options={"anon": True} + ) + + +class TestOpenFunctionsAcceptPipelines: + """ + `zarr.open`, `open_group`, `open_array` and `open_consolidated` accept a + `URLPipeline` and open the node it addresses; the body of a trailing format + segment is the node path. + """ + + @pytest.fixture + def root(self, tmp_path: Path) -> str: + register_url_adapter("wrap", WrapperAdapter) + url = f"file:{tmp_path.as_posix()}|wrap:" + group = zarr.create_group(URLPipeline.from_url(url + "|zarr2:"), zarr_format=2) + group.create_array("sub/x", shape=(3,), dtype="i4")[:] = [1, 2, 3] + zarr.consolidate_metadata(URLPipeline.from_url(url + "|zarr2:")) + return url + + async def test_open(self, root: str) -> None: + group = zarr.open(URLPipeline.from_url(root + "|zarr2:"), mode="r") + assert isinstance(group, zarr.Group) + assert group.metadata.zarr_format == 2 + assert isinstance(group.store, TracingStore) + array = zarr.open(URLPipeline.from_url(root + "|zarr2:sub/x"), mode="r") + assert isinstance(array, zarr.Array) + assert array[1] == 2 + + async def test_open_group(self, root: str) -> None: + group = zarr.open_group(URLPipeline.from_url(root + "|zarr2:sub"), mode="r") + assert group.path == "sub" + assert group.read_only + + async def test_open_array(self, root: str) -> None: + array = zarr.open_array(URLPipeline.from_url(root + "sub|zarr2:x"), mode="r") + assert array.store_path.path == "sub/x" + assert array[2] == 3 + + async def test_open_consolidated(self, root: str) -> None: + group = zarr.open_consolidated(URLPipeline.from_url(root + "|zarr2:"), mode="r") + assert group.metadata.consolidated_metadata is not None + assert group.metadata.zarr_format == 2 + + async def test_async_open(self, root: str) -> None: + group = await zarr.api.asynchronous.open( + store=URLPipeline.from_url(root + "|zarr2:"), mode="r" + ) + assert isinstance(group, zarr.AsyncGroup) + + @pytest.mark.parametrize( + "open_func", + [zarr.open, zarr.open_group, zarr.open_array, zarr.open_consolidated], + ) + @pytest.mark.skipif(sys.platform == "win32", reason="'|' is not allowed in Windows file names") + async def test_strings_are_not_pipelines( + self, open_func: Callable[..., object], root: str, tmp_path: Path + ) -> None: + """The same text passed as a string addresses a local path, not the pipeline.""" + with pytest.raises(FileNotFoundError): + open_func(f"{tmp_path}|wrap:|zarr2:", mode="r") + + +class TestZarrFormatFromPipeline: + """ + Every function taking a StoreLike honors the format selected by a + `URLPipeline`, not only the top-level zarr.api functions. Functions whose + `zarr_format` defaults to 3 raise when the default conflicts with the + pipeline; `create_array` and `Group.open` accept `zarr_format=None` to defer + to it. + """ + + @pytest.fixture + def v2_pipeline(self, tmp_path: Path) -> URLPipeline: + return URLPipeline.from_url(f"file:{tmp_path.as_posix()}|zarr2:") + + async def test_create_array(self, v2_pipeline: URLPipeline) -> None: + arr = zarr.create_array(v2_pipeline, name="x", shape=(2,), dtype="i4", zarr_format=None) + assert arr.metadata.zarr_format == 2 + with pytest.raises(ValueError, match="conflicts with"): + zarr.create_array(v2_pipeline, name="y", shape=(2,), dtype="i4") + + async def test_create_array_from_data(self, v2_pipeline: URLPipeline) -> None: + # the data path of create_array goes through from_array + arr = zarr.create_array(v2_pipeline, name="x", data=np.arange(3), zarr_format=None) + assert arr.metadata.zarr_format == 2 + with pytest.raises(ValueError, match="conflicts with"): + zarr.create_array(v2_pipeline, name="y", data=np.arange(3)) + + async def test_create_array_default_without_pipeline(self, tmp_path: Path) -> None: + """The default format for plain stores is still 3, independent of the config default.""" + with zarr.config.set({"default_zarr_format": 2}): + arr = zarr.create_array(tmp_path, shape=(2,), dtype="i4") + assert arr.metadata.zarr_format == 3 + + async def test_from_array(self, v2_pipeline: URLPipeline) -> None: + arr = zarr.from_array(v2_pipeline, name="x", data=np.arange(3)) + assert arr.metadata.zarr_format == 2 + + async def test_group_from_store(self, v2_pipeline: URLPipeline, tmp_path: Path) -> None: + group = zarr.Group.from_store(v2_pipeline, zarr_format=2) + assert group.metadata.zarr_format == 2 + with pytest.raises(ValueError, match="conflicts with"): + zarr.Group.from_store(URLPipeline.from_url(f"file:{tmp_path.as_posix()}/other|zarr2:")) + + async def test_group_from_store_default_without_pipeline(self, tmp_path: Path) -> None: + """The default format for plain stores is still 3, independent of the config default.""" + with zarr.config.set({"default_zarr_format": 2}): + group = zarr.Group.from_store(tmp_path) + assert group.metadata.zarr_format == 3 + + async def test_group_open(self, v2_pipeline: URLPipeline) -> None: + zarr.Group.from_store(v2_pipeline, zarr_format=2) + group = zarr.Group.open(v2_pipeline, zarr_format=None) + assert group.metadata.zarr_format == 2 + with pytest.raises(ValueError, match="conflicts with"): + zarr.Group.open(v2_pipeline) + + async def test_array_open(self, v2_pipeline: URLPipeline, tmp_path: Path) -> None: + zarr.create_array(v2_pipeline, name="x", shape=(2,), dtype="i4", zarr_format=None) + # Array.open opens zarr v3 only, so a v2 pipeline conflicts with it + with pytest.raises(ValueError, match="conflicts with"): + zarr.Array.open(URLPipeline.from_url(f"file:{tmp_path.as_posix()}|zarr2:x")) + zarr.create_array( + URLPipeline.from_url(f"file:{tmp_path.as_posix()}/v3|zarr3:"), + name="y", + shape=(2,), + dtype="i4", + ) + arr = zarr.Array.open(URLPipeline.from_url(f"file:{tmp_path.as_posix()}/v3|zarr3:y")) + assert arr.metadata.zarr_format == 3 + + async def test_open_like(self, v2_pipeline: URLPipeline) -> None: + # open_like inherits v2 filters/compressor from the reference; the + # pipeline supplies the format, which open_array then creates with + ref = zarr.create_array({}, shape=(2,), dtype="i4", zarr_format=2) + arr = zarr.open_like(ref, store=v2_pipeline, path="x") + assert arr.metadata.zarr_format == 2 + + async def test_deprecated_async_array_create( + self, v2_pipeline: URLPipeline, tmp_path: Path + ) -> None: + from zarr.core.array import AsyncArray + + arr = await AsyncArray._create(v2_pipeline, shape=(2,), dtype="i4", zarr_format=2) + assert arr.metadata.zarr_format == 2 + with pytest.raises(ValueError, match="conflicts with"): + await AsyncArray._create( + URLPipeline.from_url(f"file:{tmp_path.as_posix()}/o|zarr2:"), + shape=(2,), + dtype="i4", + zarr_format=3, + ) diff --git a/tests/test_url_pipeline/test_spec_examples.py b/tests/test_url_pipeline/test_spec_examples.py new file mode 100644 index 0000000000..758e20d467 --- /dev/null +++ b/tests/test_url_pipeline/test_spec_examples.py @@ -0,0 +1,255 @@ +""" +Conformance tests against the URL pipeline specification. + +Every entry in `SPEC_EXAMPLES` is an example URL from the specification +repository (https://github.com/jbms/url-pipeline @ 1a01ce6), extracted +with the same rule as the spec's own linter (`scripts/lint.py`): backticked +examples on bullet lines, skipping spans without a `:` or `|`. The spec +validates each example (and an uppercased-scheme variant) against its ABNF +grammar, so this corpus is grammar-valid by construction. + +The parser must accept every example, whether or not an adapter exists for its +schemes, split it into the expected schemes, and preserve the text losslessly. To +regenerate, re-extract the examples and recompute the expected schemes with +`_parse_pipeline`, reviewing changes against the spec diff. +""" + +from __future__ import annotations + +import re + +import pytest + +from zarr.storage._url_pipeline import _parse_pipeline + +SPEC_EXAMPLES: list[tuple[str, tuple[str, ...]]] = [ + # README.md + ("s3://bucket/path/to/archive.zip|zip:path/within/zip.zarr/|zarr3:", ("s3", "zip", "zarr3")), + ( + "file:///tmp/dataset.ocdbt/|ocdbt://2025-01-01T01:23:45.678Z/path/within/database", + ("file", "ocdbt"), + ), + ( + "s3+https://example.com/path/to/database.icechunk/|icechunk://tag.v5/path/to/node/|zarr3:", + ("s3+https", "icechunk", "zarr3"), + ), + ( + "vendor1-2.custom-1+proto.ext://[authority]/path?query/part?x", + ("vendor1-2.custom-1+proto.ext",), + ), + ( + "http://example.com/file|vendor1-2.custom+adapter:/path/within/adapter", + ("http", "vendor1-2.custom+adapter"), + ), + ("a.b:?", ("a.b",)), + ("http://somehost/downloads/somefile.zip|zip:", ("http", "zip")), + ("http://example.com/archive.jar|zip:path/to/file.txt", ("http", "zip")), + ("https://host/archive.zip|zip:path/in/outer.zip|zip:path/in/inner", ("https", "zip", "zip")), + ("file:///path/to/archive.zip|zip:path/within/archive", ("file", "zip")), + # avif.md + ("file:/path/to/image.avif|avif:", ("file", "avif")), + ("file:/path/to/image.avif|avif", ("file", "avif")), + # bmp.md + ("file:/path/to/image.bmp|bmp:", ("file", "bmp")), + ("file:/path/to/image.bmp|bmp", ("file", "bmp")), + # byte-range.md + ("file:/path/to/data|byte-range:1000-2000", ("file", "byte-range")), + ("file:/path/to/data|byte-range:0-1", ("file", "byte-range")), + # file.md + ("file:/", ("file",)), + ("file:/a", ("file",)), + ("file:/path/to/file.txt", ("file",)), + ("file://localhost/path/to/file.txt", ("file",)), + ("file://LOCALHOST/path/to/file.txt", ("file",)), + ("file:///path/to/file.txt", ("file",)), + ("file://somehost/sharename/path/to/file.txt", ("file",)), + # gs.md + ("gs://bucket", ("gs",)), + ("gs://bucket/", ("gs",)), + ("gs://bucket/path/within/bucket", ("gs",)), + ("gs://aaa", ("gs",)), + ( + "gs://label1-is-sixty-two-characters-long-xxxxxxxxxxxxxxxxxxxxxxxxxx.label2-is-sixty-two-characters-long-yyyyyyyyyyyyyyyyyyyyyyyyyy.label3-is-sixty-two-characters-long-zzzzzzzzzzzzzzzzzzzzzzzzzz.label4-is-thirty-one-characters", + ("gs",), + ), + # gzip.md + ("file:/path/to/data.gz|gzip:", ("file", "gzip")), + ("file:/path/to/data.gz|gzip", ("file", "gzip")), + # hdf5.md + ("s3://bucket/path/to/file.h5|hdf5:/path/to/dataset", ("s3", "hdf5")), + ("s3://bucket/path/to/file.h5|hdf5:a", ("s3", "hdf5")), + ("file:///path/to/file.h5|hdf5:", ("file", "hdf5")), + ("file:///path/to/file.h5|hdf5", ("file", "hdf5")), + # http.md + ("https://example.com/path/to/resource%20name", ("https",)), + ("https://example.com/path/to/resource?query=value", ("https",)), + ("https://example.com/path/to/?query=value", ("https",)), + ("https://example.com/path/with:colon", ("https",)), + ("https://server.example.com:1234/path/to/array", ("https",)), + ("http://a", ("http",)), + ("http://example.com", ("http",)), + ("http://local%68ost", ("http",)), + ("http://example.com:", ("http",)), + ("http://192.168.10.1:1234", ("http",)), + ("http://[::1]", ("http",)), + ("http://[::ffff:127.0.0.1]", ("http",)), + ("http://example.com/", ("http",)), + # icechunk.md + ("file:///path/to/repo.zarr.icechunk/|icechunk:", ("file", "icechunk")), + ("file:///path/to/repo.zarr.icechunk/|icechunk", ("file", "icechunk")), + ("file:///path/to/repo.zarr.icechunk/|icechunk://branch.main/", ("file", "icechunk")), + ("file:///path/to/repo.zarr.icechunk/|icechunk:path/to/node/", ("file", "icechunk")), + ("file:///path/to/repo.zarr.icechunk/|icechunk:a", ("file", "icechunk")), + ("file:///path/to/repo.zarr.icechunk/|icechunk:/path/to/node/", ("file", "icechunk")), + ("file:///path/to/repo.zarr.icechunk/|icechunk:path/to/node/zarr.json", ("file", "icechunk")), + ("file:///path/to/repo.zarr.icechunk/|icechunk:path/to/node/c/0/0/1", ("file", "icechunk")), + ( + "file:///path/to/repo.zarr.icechunk/|icechunk://branch.mybranch/path/to/node/", + ("file", "icechunk"), + ), + ("file:///path/to/repo.zarr.icechunk/|icechunk://branch.a", ("file", "icechunk")), + ( + "file:///path/to/repo.zarr.icechunk/|icechunk://tag.mytag/path/to/node/", + ("file", "icechunk"), + ), + ("file:///path/to/repo.zarr.icechunk/|icechunk://tag.a", ("file", "icechunk")), + ( + "file:///path/to/repo.zarr.icechunk/|icechunk://FWWFQGAW742XMX0F5MF0/path/to/node/", + ("file", "icechunk"), + ), + ( + "file:///path/to/repo.zarr.icechunk/|icechunk://ABCDEFGHJKMNPQRSTVWX/path/to/node/", + ("file", "icechunk"), + ), + ( + "file:///path/to/repo.zarr.icechunk/|icechunk:|zarr3:path/to/array/", + ("file", "icechunk", "zarr3"), + ), + ( + "file:///path/to/repo.zarr.icechunk/|icechunk://branch.other/|zarr3:path/to/array/", + ("file", "icechunk", "zarr3"), + ), + ( + "file:///path/to/repo.zarr.icechunk/|icechunk://tag.v5/|zarr3:path/to/array/", + ("file", "icechunk", "zarr3"), + ), + ( + "file:///path/to/repo.zarr.icechunk/|icechunk://4N0217AZA4VNPYD0HR0G/|zarr3:path/to/array/", + ("file", "icechunk", "zarr3"), + ), + # jpeg.md + ("file:/path/to/image.jpeg|jpeg:", ("file", "jpeg")), + ("file:/path/to/image.jpeg|jpeg", ("file", "jpeg")), + # json.md + ("file:/path/to/data.json|json:", ("file", "json")), + ("file:/path/to/data.json|json", ("file", "json")), + ("file:/path/to/data.json|json:/path/to/node", ("file", "json")), + ("file:/path/to/data.json|json:/abc~0def", ("file", "json")), + ("file:/path/to/data.json|json:/abc~1def", ("file", "json")), + ("file:/path/to/data.json|json:/", ("file", "json")), + # memory.md + ("memory:", ("memory",)), + ("memory:/", ("memory",)), + ("memory://", ("memory",)), + ("memory:path/to/resource", ("memory",)), + ("memory:a", ("memory",)), + ("memory:another@path+with,lots&of(special;characters)*_!-$'", ("memory",)), + ("memory:/path/to/resource", ("memory",)), + ("memory://path/to/resource", ("memory",)), + # n5.md + ("file:///tmp/data.n5/|n5:path/to/array", ("file", "n5")), + ("file:///tmp/data.n5/|n5:a", ("file", "n5")), + ("file:///tmp/data.n5/|n5:/path/to/array", ("file", "n5")), + ("file:///tmp/data.n5/|n5:", ("file", "n5")), + ("file:///tmp/data.n5/|n5", ("file", "n5")), + # neuroglancer-precomputed.md + ( + "file:///tmp/dataset.precomputed/|neuroglancer-precomputed:", + ("file", "neuroglancer-precomputed"), + ), + ( + "file:///tmp/dataset.precomputed/|neuroglancer-precomputed", + ("file", "neuroglancer-precomputed"), + ), + # ocdbt.md + ("file:///path/to/repo.ocdbt/|ocdbt:", ("file", "ocdbt")), + ("file:///path/to/repo.ocdbt/|ocdbt", ("file", "ocdbt")), + ("file:///path/to/repo.ocdbt/|ocdbt:path/within/repo/", ("file", "ocdbt")), + ("file:///path/to/repo.ocdbt/|ocdbt:path/within/repo", ("file", "ocdbt")), + ("file:///path/to/repo.ocdbt/|ocdbt:a", ("file", "ocdbt")), + ("file:///path/to/repo.ocdbt/|ocdbt:/path/within/repo", ("file", "ocdbt")), + ("file:///path/to/repo.ocdbt/|ocdbt://v123/", ("file", "ocdbt")), + ("file:///path/to/repo.ocdbt/|ocdbt://v1", ("file", "ocdbt")), + ("file:///path/to/repo.ocdbt/|ocdbt://v123/path/within/repo", ("file", "ocdbt")), + ( + "file:///path/to/repo.ocdbt/|ocdbt://2025-01-01T01:23:45.678Z/path/within/database", + ("file", "ocdbt"), + ), + ("file:///path/to/repo.ocdbt/|ocdbt://2025-01-01T01:23:45Z", ("file", "ocdbt")), + ("file:///path/to/repo.ocdbt/|ocdbt://2025-01-01T01:23:45.1Z", ("file", "ocdbt")), + ("file:///path/to/repo.ocdbt/|ocdbt://2025-01-01T01:23:45.123456789Z", ("file", "ocdbt")), + # png.md + ("file:/path/to/image.png|png:", ("file", "png")), + ("file:/path/to/image.png|png", ("file", "png")), + # s3+http.md + ("s3+https://endpoint/path/within/bucket", ("s3+https",)), + ("s3+https://endpoint/bucket/path/within/bucket", ("s3+https",)), + ("s3+https://mybucket.s3.amazonaws.com/path/to/file", ("s3+https",)), + ("s3+https://s3.amazonaws.com/mybucket/path/to/file", ("s3+https",)), + ("s3+http://example.com", ("s3+http",)), + ("s3+http://example.com/", ("s3+http",)), + # s3.md + ("s3://bucket", ("s3",)), + ("s3://bucket/", ("s3",)), + ("s3://bucket/path/within/bucket", ("s3",)), + ("s3://aaa", ("s3",)), + ( + "s3://aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + ("s3",), + ), + # tiff.md + ("file:/path/to/image.tiff|tiff:", ("file", "tiff")), + ("file:/path/to/image.tiff|tiff", ("file", "tiff")), + # webp.md + ("file:/path/to/image.webp|webp:", ("file", "webp")), + ("file:/path/to/image.webp|webp", ("file", "webp")), + # zarr.md + ("file:///path/to/node.zarr/|zarr3:", ("file", "zarr3")), + ("file:///path/to/node.zarr/|zarr3", ("file", "zarr3")), + ("file:///path/to/node.zarr/|zarr2:", ("file", "zarr2")), + ("file:///path/to/node.zarr/|zarr2", ("file", "zarr2")), + ("file:///path/to/node.zarr/|zarr:", ("file", "zarr")), + ("file:///path/to/node.zarr/|zarr", ("file", "zarr")), + # zip.md + ("file:/path/to/archive.zip|zip:path/to/file.txt", ("file", "zip")), + ("file:/path/to/archive.zip|zip", ("file", "zip")), + ("file:/path/to/archive.zip|zip:/path/to/file.txt", ("file", "zip")), + ("file:/path/to/outer.zip|zip:path/to/inner.zip|zip:path/to/file.txt", ("file", "zip", "zip")), + # zstd.md + ("file:/path/to/data.zstd|zstd:", ("file", "zstd")), + ("file:/path/to/data.zstd|zstd", ("file", "zstd")), +] + + +def _uppercase_schemes(example: str) -> str: + """Uppercase every sub-URL scheme, as the spec's linter does.""" + + def repl(match: re.Match[str]) -> str: + return match.group(1) + match.group(2).upper() + + return re.sub(r"(^|\|)([a-zA-Z][a-zA-Z0-9+.-]*)", repl, example) + + +@pytest.mark.parametrize(("url", "expected_schemes"), SPEC_EXAMPLES, ids=str) +def test_spec_example_parses(url: str, expected_schemes: tuple[str, ...]) -> None: + segments = _parse_pipeline(url) + assert tuple(segment.scheme for segment in segments) == expected_schemes + # lossless round-trip of the original text + assert "|".join(segment.raw for segment in segments) == url + + +@pytest.mark.parametrize(("url", "expected_schemes"), SPEC_EXAMPLES, ids=str) +def test_spec_example_schemes_case_insensitive(url: str, expected_schemes: tuple[str, ...]) -> None: + upper = _uppercase_schemes(url) + segments = _parse_pipeline(upper) + assert tuple(segment.scheme for segment in segments) == expected_schemes