Store

Store

zarr.testing.store

B module-attribute

B = TypeVar('B', bound=Buffer)

S module-attribute

S = TypeVar('S', bound=Store)

__all__ module-attribute

__all__ = ['StoreTests']

LatencyStore

Bases: WrapperStore[Store]

A wrapper class that takes any store class in its constructor and adds latency to the set and get methods. This can be used for performance testing.

Source code in zarr/testing/store.py

class LatencyStore(WrapperStore[Store]):
    """
    A wrapper class that takes any store class in its constructor and
    adds latency to the `set` and `get` methods. This can be used for
    performance testing.
    """

    get_latency: float
    set_latency: float

    def __init__(self, store: Store, *, get_latency: float = 0, set_latency: float = 0) -> None:
        self.get_latency = float(get_latency)
        self.set_latency = float(set_latency)
        self._store = store

    def _with_store(self, store: Store) -> Self:
        return type(self)(store, get_latency=self.get_latency, set_latency=self.set_latency)

    async def set(self, key: str, value: Buffer) -> None:
        """
        Add latency to the ``set`` method.

        Calls ``asyncio.sleep(self.set_latency)`` before invoking the wrapped ``set`` method.

        Parameters
        ----------
        key : str
            The key to set
        value : Buffer
            The value to set

        Returns
        -------
        None
        """
        await asyncio.sleep(self.set_latency)
        await self._store.set(key, value)

    async def get(
        self, key: str, prototype: BufferPrototype, byte_range: ByteRequest | None = None
    ) -> Buffer | None:
        """
        Add latency to the ``get`` method.

        Calls ``asyncio.sleep(self.get_latency)`` before invoking the wrapped ``get`` method.

        Parameters
        ----------
        key : str
            The key to get
        prototype : BufferPrototype
            The BufferPrototype to use.
        byte_range : ByteRequest, optional
            An optional byte range.

        Returns
        -------
        buffer : Buffer or None
        """
        await asyncio.sleep(self.get_latency)
        return await self._store.get(key, prototype=prototype, byte_range=byte_range)

get_latency instance-attribute

get_latency: float = float(get_latency)

read_only property

read_only: bool

Is the store read-only?

set_latency instance-attribute

set_latency: float = float(set_latency)

supports_consolidated_metadata property

supports_consolidated_metadata: bool

Does the store support consolidated metadata?.

If it doesn't an error will be raised on requests to consolidate the metadata. Returning False can be useful for stores which implement their own consolidation mechanism outside of the zarr-python implementation.

supports_deletes property

supports_deletes: bool

Does the store support deletes?

supports_listing property

supports_listing: bool

Does the store support listing?

supports_partial_writes property

supports_partial_writes: Literal[False]

Does the store support partial writes?

Partial writes are no longer used by Zarr, so this is always false.

supports_writes property

supports_writes: bool

Does the store support writes?

__enter__

__enter__() -> Self

Enter a context manager that will close the store upon exiting.

Source code in zarr/storage/_wrapper.py

def __enter__(self) -> Self:
    return self._with_store(self._store.__enter__())

__eq__

__eq__(value: object) -> bool

Equality comparison.

Source code in zarr/storage/_wrapper.py

def __eq__(self, value: object) -> bool:
    return type(self) is type(value) and self._store.__eq__(value._store)  # type: ignore[attr-defined]

__exit__

__exit__(
    exc_type: type[BaseException] | None,
    exc_value: BaseException | None,
    traceback: TracebackType | None,
) -> None

Close the store.

Source code in zarr/storage/_wrapper.py

def __exit__(
    self,
    exc_type: type[BaseException] | None,
    exc_value: BaseException | None,
    traceback: TracebackType | None,
) -> None:
    return self._store.__exit__(exc_type, exc_value, traceback)

__init__

__init__(
    store: Store,
    *,
    get_latency: float = 0,
    set_latency: float = 0,
) -> None

Source code in zarr/testing/store.py

def __init__(self, store: Store, *, get_latency: float = 0, set_latency: float = 0) -> None:
    self.get_latency = float(get_latency)
    self.set_latency = float(set_latency)
    self._store = store

__repr__

__repr__() -> str

Source code in zarr/storage/_wrapper.py

def __repr__(self) -> str:
    return f"WrapperStore({self._store.__class__.__name__}, '{self._store}')"

__str__

__str__() -> str

Source code in zarr/storage/_wrapper.py

def __str__(self) -> str:
    return f"wrapping-{self._store}"

clear async

clear() -> None

Clear the store.

Remove all keys and values from the store.

Source code in zarr/storage/_wrapper.py

async def clear(self) -> None:
    return await self._store.clear()

close

close() -> None

Close the store.

Source code in zarr/storage/_wrapper.py

def close(self) -> None:
    self._store.close()

delete async

delete(key: str) -> None

Remove a key from the store

Parameters:

• key (str) –

Source code in zarr/storage/_wrapper.py

async def delete(self, key: str) -> None:
    await self._store.delete(key)

delete_dir async

delete_dir(prefix: str) -> None

Remove all keys and prefixes in the store that begin with a given prefix.

Source code in zarr/storage/_wrapper.py

async def delete_dir(self, prefix: str) -> None:
    return await self._store.delete_dir(prefix)

exists async

exists(key: str) -> bool

Check if a key exists in the store.

Parameters:

• key (str) –

Returns:

• bool –

Source code in zarr/storage/_wrapper.py

async def exists(self, key: str) -> bool:
    return await self._store.exists(key)

get async

get(
    key: str,
    prototype: BufferPrototype,
    byte_range: ByteRequest | None = None,
) -> Buffer | None

Add latency to the get method.

Calls asyncio.sleep(self.get_latency) before invoking the wrapped get method.

Parameters:

• key (str) –

  The key to get

• prototype (BufferPrototype) –

  The BufferPrototype to use.

• byte_range (ByteRequest, default: None ) –

  An optional byte range.

Returns:

• buffer ( Buffer or None ) –

Source code in zarr/testing/store.py

async def get(
    self, key: str, prototype: BufferPrototype, byte_range: ByteRequest | None = None
) -> Buffer | None:
    """
    Add latency to the ``get`` method.

    Calls ``asyncio.sleep(self.get_latency)`` before invoking the wrapped ``get`` method.

    Parameters
    ----------
    key : str
        The key to get
    prototype : BufferPrototype
        The BufferPrototype to use.
    byte_range : ByteRequest, optional
        An optional byte range.

    Returns
    -------
    buffer : Buffer or None
    """
    await asyncio.sleep(self.get_latency)
    return await self._store.get(key, prototype=prototype, byte_range=byte_range)

get_partial_values async

get_partial_values(
    prototype: BufferPrototype,
    key_ranges: Iterable[tuple[str, ByteRequest | None]],
) -> list[Buffer | None]

Retrieve possibly partial values from given key_ranges.

Parameters:

• prototype (BufferPrototype) –

  The prototype of the output buffer. Stores may support a default buffer prototype.

• key_ranges (Iterable[tuple[str, tuple[int | None, int | None]]]) –

  Ordered set of key, range pairs, a key may occur multiple times with different ranges

Returns:

• list of values, in the order of the key_ranges, may contain null/none for missing keys –

Source code in zarr/storage/_wrapper.py

async def get_partial_values(
    self,
    prototype: BufferPrototype,
    key_ranges: Iterable[tuple[str, ByteRequest | None]],
) -> list[Buffer | None]:
    return await self._store.get_partial_values(prototype, key_ranges)

getsize async

getsize(key: str) -> int

Return the size, in bytes, of a value in a Store.

Parameters:

• key (str) –

Returns:

• nbytes ( int ) –

  The size of the value (in bytes).

Raises:

• FileNotFoundError –

  When the given key does not exist in the store.

Source code in zarr/abc/store.py

async def getsize(self, key: str) -> int:
    """
    Return the size, in bytes, of a value in a Store.

    Parameters
    ----------
    key : str

    Returns
    -------
    nbytes : int
        The size of the value (in bytes).

    Raises
    ------
    FileNotFoundError
        When the given key does not exist in the store.
    """
    # Note to implementers: this default implementation is very inefficient since
    # it requires reading the entire object. Many systems will have ways to get the
    # size of an object without reading it.
    # avoid circular import
    from zarr.core.buffer.core import default_buffer_prototype

    value = await self.get(key, prototype=default_buffer_prototype())
    if value is None:
        raise FileNotFoundError(key)
    return len(value)

getsize_prefix async

getsize_prefix(prefix: str) -> int

Return the size, in bytes, of all values under a prefix.

Parameters:

• prefix (str) –

  The prefix of the directory to measure.

Returns:

• nbytes ( int ) –

  The sum of the sizes of the values in the directory (in bytes).

See Also

zarr.Array.nbytes_stored Store.getsize

Notes

getsize_prefix is just provided as a potentially faster alternative to listing all the keys under a prefix calling Store.getsize on each.

In general, prefix should be the path of an Array or Group in the Store. Implementations may differ on the behavior when some other prefix is provided.

Source code in zarr/abc/store.py

async def getsize_prefix(self, prefix: str) -> int:
    """
    Return the size, in bytes, of all values under a prefix.

    Parameters
    ----------
    prefix : str
        The prefix of the directory to measure.

    Returns
    -------
    nbytes : int
        The sum of the sizes of the values in the directory (in bytes).

    See Also
    --------
    zarr.Array.nbytes_stored
    Store.getsize

    Notes
    -----
    ``getsize_prefix`` is just provided as a potentially faster alternative to
    listing all the keys under a prefix calling [`Store.getsize`][zarr.abc.store.Store.getsize] on each.

    In general, ``prefix`` should be the path of an Array or Group in the Store.
    Implementations may differ on the behavior when some other ``prefix``
    is provided.
    """
    # TODO: Overlap listing keys with getsize calls.
    # Currently, we load the list of keys into memory and only then move
    # on to getting sizes. Ideally we would overlap those two, which should
    # improve tail latency and might reduce memory pressure (since not all keys
    # would be in memory at once).

    # avoid circular import
    from zarr.core.common import concurrent_map
    from zarr.core.config import config

    keys = [(x,) async for x in self.list_prefix(prefix)]
    limit = config.get("async.concurrency")
    sizes = await concurrent_map(keys, self.getsize, limit=limit)
    return sum(sizes)

is_empty async

is_empty(prefix: str) -> bool

Check if the directory is empty.

Parameters:

• prefix (str) –

  Prefix of keys to check.

Returns:

• bool –

  True if the store is empty, False otherwise.

Source code in zarr/storage/_wrapper.py

async def is_empty(self, prefix: str) -> bool:
    return await self._store.is_empty(prefix)

list

list() -> AsyncIterator[str]

Retrieve all keys in the store.

Returns:

• AsyncIterator[str] –

Source code in zarr/storage/_wrapper.py

def list(self) -> AsyncIterator[str]:
    return self._store.list()

list_dir

list_dir(prefix: str) -> AsyncIterator[str]

Retrieve all keys and prefixes with a given prefix and which do not contain the character “/” after the given prefix.

Parameters:

• prefix (str) –

Returns:

• AsyncIterator[str] –

Source code in zarr/storage/_wrapper.py

def list_dir(self, prefix: str) -> AsyncIterator[str]:
    return self._store.list_dir(prefix)

list_prefix

list_prefix(prefix: str) -> AsyncIterator[str]

Retrieve all keys in the store that begin with a given prefix. Keys are returned relative to the root of the store.

Parameters:

• prefix (str) –

Returns:

• AsyncIterator[str] –

Source code in zarr/storage/_wrapper.py

def list_prefix(self, prefix: str) -> AsyncIterator[str]:
    return self._store.list_prefix(prefix)

open async classmethod

open(
    store_cls: type[T_Store], *args: Any, **kwargs: Any
) -> Self

Create and open the store.

Parameters:

• *args (Any, default: () ) –

  Positional arguments to pass to the store constructor.

• **kwargs (Any, default: {} ) –

  Keyword arguments to pass to the store constructor.

Returns:

• Store –

  The opened store instance.

Source code in zarr/storage/_wrapper.py

@classmethod
async def open(cls: type[Self], store_cls: type[T_Store], *args: Any, **kwargs: Any) -> Self:
    store = store_cls(*args, **kwargs)
    await store._open()
    return cls(store=store)

set async

set(key: str, value: Buffer) -> None

Add latency to the set method.

Calls asyncio.sleep(self.set_latency) before invoking the wrapped set method.

Parameters:

• key (str) –

  The key to set

• value (Buffer) –

  The value to set

Returns:

• None –

Source code in zarr/testing/store.py

async def set(self, key: str, value: Buffer) -> None:
    """
    Add latency to the ``set`` method.

    Calls ``asyncio.sleep(self.set_latency)`` before invoking the wrapped ``set`` method.

    Parameters
    ----------
    key : str
        The key to set
    value : Buffer
        The value to set

    Returns
    -------
    None
    """
    await asyncio.sleep(self.set_latency)
    await self._store.set(key, value)

set_if_not_exists async

set_if_not_exists(key: str, value: Buffer) -> None

Store a key to value if the key is not already present.

Parameters:

• key (str) –
• value (Buffer) –

Source code in zarr/storage/_wrapper.py

async def set_if_not_exists(self, key: str, value: Buffer) -> None:
    return await self._store.set_if_not_exists(key, value)

with_read_only

with_read_only(read_only: bool = False) -> Self

Return a new store with a new read_only setting.

The new store points to the same location with the specified new read_only state. The returned Store is not automatically opened, and this store is not automatically closed.

Parameters:

• read_only (bool, default: False ) –

  If True, the store will be created in read-only mode. Defaults to False.

Returns:

• A new store of the same type with the new read only attribute. –

Source code in zarr/storage/_wrapper.py

def with_read_only(self, read_only: bool = False) -> Self:
    return self._with_store(cast(T_Store, self._store.with_read_only(read_only)))

StoreTests

Bases: Generic[S, B]

Source code in zarr/testing/store.py

class StoreTests(Generic[S, B]):
    store_cls: type[S]
    buffer_cls: type[B]

    @abstractmethod
    async def set(self, store: S, key: str, value: Buffer) -> None:
        """
        Insert a value into a storage backend, with a specific key.
        This should not use any store methods. Bypassing the store methods allows them to be
        tested.
        """
        ...

    @abstractmethod
    async def get(self, store: S, key: str) -> Buffer:
        """
        Retrieve a value from a storage backend, by key.
        This should not use any store methods. Bypassing the store methods allows them to be
        tested.
        """
        ...

    @abstractmethod
    @pytest.fixture
    def store_kwargs(self, *args: Any, **kwargs: Any) -> dict[str, Any]:
        """Kwargs for instantiating a store"""
        ...

    @abstractmethod
    def test_store_repr(self, store: S) -> None: ...

    @abstractmethod
    def test_store_supports_writes(self, store: S) -> None: ...

    def test_store_supports_partial_writes(self, store: S) -> None:
        assert not store.supports_partial_writes

    @abstractmethod
    def test_store_supports_listing(self, store: S) -> None: ...

    @pytest.fixture
    def open_kwargs(self, store_kwargs: dict[str, Any]) -> dict[str, Any]:
        return store_kwargs

    @pytest.fixture
    async def store(self, open_kwargs: dict[str, Any]) -> Store:
        return await self.store_cls.open(**open_kwargs)

    @pytest.fixture
    async def store_not_open(self, store_kwargs: dict[str, Any]) -> Store:
        return self.store_cls(**store_kwargs)

    def test_store_type(self, store: S) -> None:
        assert isinstance(store, Store)
        assert isinstance(store, self.store_cls)

    def test_store_eq(self, store: S, store_kwargs: dict[str, Any]) -> None:
        # check self equality
        assert store == store

        # check store equality with same inputs
        # asserting this is important for being able to compare (de)serialized stores
        store2 = self.store_cls(**store_kwargs)
        assert store == store2

    async def test_serializable_store(self, store: S) -> None:
        new_store: S = pickle.loads(pickle.dumps(store))
        assert new_store == store
        assert new_store.read_only == store.read_only
        # quickly roundtrip data to a key to test that new store works
        data_buf = self.buffer_cls.from_bytes(b"\x01\x02\x03\x04")
        key = "foo"
        await store.set(key, data_buf)
        observed = await store.get(key, prototype=default_buffer_prototype())
        assert_bytes_equal(observed, data_buf)

    def test_store_read_only(self, store: S) -> None:
        assert not store.read_only

        with pytest.raises(AttributeError):
            store.read_only = False  # type: ignore[misc]

    @pytest.mark.parametrize("read_only", [True, False])
    async def test_store_open_read_only(self, open_kwargs: dict[str, Any], read_only: bool) -> None:
        open_kwargs["read_only"] = read_only
        store = await self.store_cls.open(**open_kwargs)
        assert store._is_open
        assert store.read_only == read_only

    async def test_store_context_manager(self, open_kwargs: dict[str, Any]) -> None:
        # Test that the context manager closes the store
        with await self.store_cls.open(**open_kwargs) as store:
            assert store._is_open
            # Test trying to open an already open store
            with pytest.raises(ValueError, match="store is already open"):
                await store._open()
        assert not store._is_open

    async def test_read_only_store_raises(self, open_kwargs: dict[str, Any]) -> None:
        kwargs = {**open_kwargs, "read_only": True}
        store = await self.store_cls.open(**kwargs)
        assert store.read_only

        # set
        with pytest.raises(
            ValueError, match="store was opened in read-only mode and does not support writing"
        ):
            await store.set("foo", self.buffer_cls.from_bytes(b"bar"))

        # delete
        with pytest.raises(
            ValueError, match="store was opened in read-only mode and does not support writing"
        ):
            await store.delete("foo")

    async def test_with_read_only_store(self, open_kwargs: dict[str, Any]) -> None:
        kwargs = {**open_kwargs, "read_only": True}
        store = await self.store_cls.open(**kwargs)
        assert store.read_only

        # Test that you cannot write to a read-only store
        with pytest.raises(
            ValueError, match="store was opened in read-only mode and does not support writing"
        ):
            await store.set("foo", self.buffer_cls.from_bytes(b"bar"))

        # Check if the store implements with_read_only
        try:
            writer = store.with_read_only(read_only=False)
        except NotImplementedError:
            # Test that stores that do not implement with_read_only raise NotImplementedError with the correct message
            with pytest.raises(
                NotImplementedError,
                match=f"with_read_only is not implemented for the {type(store)} store type.",
            ):
                store.with_read_only(read_only=False)
            return

        # Test that you can write to a new store copy
        assert not writer._is_open
        assert not writer.read_only
        await writer.set("foo", self.buffer_cls.from_bytes(b"bar"))
        await writer.delete("foo")

        # Test that you cannot write to the original store
        assert store.read_only
        with pytest.raises(
            ValueError, match="store was opened in read-only mode and does not support writing"
        ):
            await store.set("foo", self.buffer_cls.from_bytes(b"bar"))
        with pytest.raises(
            ValueError, match="store was opened in read-only mode and does not support writing"
        ):
            await store.delete("foo")

        # Test that you cannot write to a read-only store copy
        reader = store.with_read_only(read_only=True)
        assert reader.read_only
        with pytest.raises(
            ValueError, match="store was opened in read-only mode and does not support writing"
        ):
            await reader.set("foo", self.buffer_cls.from_bytes(b"bar"))
        with pytest.raises(
            ValueError, match="store was opened in read-only mode and does not support writing"
        ):
            await reader.delete("foo")

    @pytest.mark.parametrize("key", ["c/0", "foo/c/0.0", "foo/0/0"])
    @pytest.mark.parametrize(
        ("data", "byte_range"),
        [
            (b"\x01\x02\x03\x04", None),
            (b"\x01\x02\x03\x04", RangeByteRequest(1, 4)),
            (b"\x01\x02\x03\x04", OffsetByteRequest(1)),
            (b"\x01\x02\x03\x04", SuffixByteRequest(1)),
            (b"", None),
        ],
    )
    async def test_get(self, store: S, key: str, data: bytes, byte_range: ByteRequest) -> None:
        """
        Ensure that data can be read from the store using the store.get method.
        """
        data_buf = self.buffer_cls.from_bytes(data)
        await self.set(store, key, data_buf)
        observed = await store.get(key, prototype=default_buffer_prototype(), byte_range=byte_range)
        start, stop = _normalize_byte_range_index(data_buf, byte_range=byte_range)
        expected = data_buf[start:stop]
        assert_bytes_equal(observed, expected)

    async def test_get_not_open(self, store_not_open: S) -> None:
        """
        Ensure that data can be read from the store that isn't yet open using the store.get method.
        """
        assert not store_not_open._is_open
        data_buf = self.buffer_cls.from_bytes(b"\x01\x02\x03\x04")
        key = "c/0"
        await self.set(store_not_open, key, data_buf)
        observed = await store_not_open.get(key, prototype=default_buffer_prototype())
        assert_bytes_equal(observed, data_buf)

    async def test_get_raises(self, store: S) -> None:
        """
        Ensure that a ValueError is raise for invalid byte range syntax
        """
        data_buf = self.buffer_cls.from_bytes(b"\x01\x02\x03\x04")
        await self.set(store, "c/0", data_buf)
        with pytest.raises((ValueError, TypeError), match=r"Unexpected byte_range, got.*"):
            await store.get("c/0", prototype=default_buffer_prototype(), byte_range=(0, 2))  # type: ignore[arg-type]

    async def test_get_many(self, store: S) -> None:
        """
        Ensure that multiple keys can be retrieved at once with the _get_many method.
        """
        keys = tuple(map(str, range(10)))
        values = tuple(f"{k}".encode() for k in keys)
        for k, v in zip(keys, values, strict=False):
            await self.set(store, k, self.buffer_cls.from_bytes(v))
        observed_buffers = await _collect_aiterator(
            store._get_many(
                zip(
                    keys,
                    (default_buffer_prototype(),) * len(keys),
                    (None,) * len(keys),
                    strict=False,
                )
            )
        )
        observed_kvs = sorted(((k, b.to_bytes()) for k, b in observed_buffers))  # type: ignore[union-attr]
        expected_kvs = sorted(((k, b) for k, b in zip(keys, values, strict=False)))
        assert observed_kvs == expected_kvs

    @pytest.mark.parametrize("key", ["c/0", "foo/c/0.0", "foo/0/0"])
    @pytest.mark.parametrize("data", [b"\x01\x02\x03\x04", b""])
    async def test_getsize(self, store: S, key: str, data: bytes) -> None:
        """
        Test the result of store.getsize().
        """
        data_buf = self.buffer_cls.from_bytes(data)
        expected = len(data_buf)
        await self.set(store, key, data_buf)
        observed = await store.getsize(key)
        assert observed == expected

    async def test_getsize_prefix(self, store: S) -> None:
        """
        Test the result of store.getsize_prefix().
        """
        data_buf = self.buffer_cls.from_bytes(b"\x01\x02\x03\x04")
        keys = ["c/0/0", "c/0/1", "c/1/0", "c/1/1"]
        keys_values = [(k, data_buf) for k in keys]
        await store._set_many(keys_values)
        expected = len(data_buf) * len(keys)
        observed = await store.getsize_prefix("c")
        assert observed == expected

    async def test_getsize_raises(self, store: S) -> None:
        """
        Test that getsize() raise a FileNotFoundError if the key doesn't exist.
        """
        with pytest.raises(FileNotFoundError):
            await store.getsize("c/1000")

    @pytest.mark.parametrize("key", ["zarr.json", "c/0", "foo/c/0.0", "foo/0/0"])
    @pytest.mark.parametrize("data", [b"\x01\x02\x03\x04", b""])
    async def test_set(self, store: S, key: str, data: bytes) -> None:
        """
        Ensure that data can be written to the store using the store.set method.
        """
        assert not store.read_only
        data_buf = self.buffer_cls.from_bytes(data)
        await store.set(key, data_buf)
        observed = await self.get(store, key)
        assert_bytes_equal(observed, data_buf)

    async def test_set_not_open(self, store_not_open: S) -> None:
        """
        Ensure that data can be written to the store that's not yet open using the store.set method.
        """
        assert not store_not_open._is_open
        data_buf = self.buffer_cls.from_bytes(b"\x01\x02\x03\x04")
        key = "c/0"
        await store_not_open.set(key, data_buf)
        observed = await self.get(store_not_open, key)
        assert_bytes_equal(observed, data_buf)

    async def test_set_many(self, store: S) -> None:
        """
        Test that a dict of key : value pairs can be inserted into the store via the
        `_set_many` method.
        """
        keys = ["zarr.json", "c/0", "foo/c/0.0", "foo/0/0"]
        data_buf = [self.buffer_cls.from_bytes(k.encode()) for k in keys]
        store_dict = dict(zip(keys, data_buf, strict=True))
        await store._set_many(store_dict.items())
        for k, v in store_dict.items():
            assert (await self.get(store, k)).to_bytes() == v.to_bytes()

    @pytest.mark.parametrize(
        "key_ranges",
        [
            [],
            [("zarr.json", RangeByteRequest(0, 2))],
            [("c/0", RangeByteRequest(0, 2)), ("zarr.json", None)],
            [
                ("c/0/0", RangeByteRequest(0, 2)),
                ("c/0/1", SuffixByteRequest(2)),
                ("c/0/2", OffsetByteRequest(2)),
            ],
        ],
    )
    async def test_get_partial_values(
        self, store: S, key_ranges: list[tuple[str, ByteRequest]]
    ) -> None:
        # put all of the data
        for key, _ in key_ranges:
            await self.set(store, key, self.buffer_cls.from_bytes(bytes(key, encoding="utf-8")))

        # read back just part of it
        observed_maybe = await store.get_partial_values(
            prototype=default_buffer_prototype(), key_ranges=key_ranges
        )

        observed: list[Buffer] = []
        expected: list[Buffer] = []

        for obs in observed_maybe:
            assert obs is not None
            observed.append(obs)

        for idx in range(len(observed)):
            key, byte_range = key_ranges[idx]
            result = await store.get(
                key, prototype=default_buffer_prototype(), byte_range=byte_range
            )
            assert result is not None
            expected.append(result)

        assert all(
            obs.to_bytes() == exp.to_bytes() for obs, exp in zip(observed, expected, strict=True)
        )

    async def test_exists(self, store: S) -> None:
        assert not await store.exists("foo")
        await store.set("foo/zarr.json", self.buffer_cls.from_bytes(b"bar"))
        assert await store.exists("foo/zarr.json")

    async def test_delete(self, store: S) -> None:
        if not store.supports_deletes:
            pytest.skip("store does not support deletes")
        await store.set("foo/zarr.json", self.buffer_cls.from_bytes(b"bar"))
        assert await store.exists("foo/zarr.json")
        await store.delete("foo/zarr.json")
        assert not await store.exists("foo/zarr.json")

    async def test_delete_dir(self, store: S) -> None:
        if not store.supports_deletes:
            pytest.skip("store does not support deletes")
        await store.set("zarr.json", self.buffer_cls.from_bytes(b"root"))
        await store.set("foo-bar/zarr.json", self.buffer_cls.from_bytes(b"root"))
        await store.set("foo/zarr.json", self.buffer_cls.from_bytes(b"bar"))
        await store.set("foo/c/0", self.buffer_cls.from_bytes(b"chunk"))
        await store.delete_dir("foo")
        assert await store.exists("zarr.json")
        assert await store.exists("foo-bar/zarr.json")
        assert not await store.exists("foo/zarr.json")
        assert not await store.exists("foo/c/0")

    async def test_delete_nonexistent_key_does_not_raise(self, store: S) -> None:
        if not store.supports_deletes:
            pytest.skip("store does not support deletes")
        await store.delete("nonexistent_key")

    async def test_is_empty(self, store: S) -> None:
        assert await store.is_empty("")
        await self.set(
            store, "foo/bar", self.buffer_cls.from_bytes(bytes("something", encoding="utf-8"))
        )
        assert not await store.is_empty("")
        assert await store.is_empty("fo")
        assert not await store.is_empty("foo/")
        assert not await store.is_empty("foo")
        assert await store.is_empty("spam/")

    async def test_clear(self, store: S) -> None:
        await self.set(
            store, "key", self.buffer_cls.from_bytes(bytes("something", encoding="utf-8"))
        )
        await store.clear()
        assert await store.is_empty("")

    async def test_list(self, store: S) -> None:
        assert await _collect_aiterator(store.list()) == ()
        prefix = "foo"
        data = self.buffer_cls.from_bytes(b"")
        store_dict = {
            prefix + "/zarr.json": data,
            **{prefix + f"/c/{idx}": data for idx in range(10)},
        }
        await store._set_many(store_dict.items())
        expected_sorted = sorted(store_dict.keys())
        observed = await _collect_aiterator(store.list())
        observed_sorted = sorted(observed)
        assert observed_sorted == expected_sorted

    async def test_list_prefix(self, store: S) -> None:
        """
        Test that the `list_prefix` method works as intended. Given a prefix, it should return
        all the keys in storage that start with this prefix.
        """
        prefixes = ("", "a/", "a/b/", "a/b/c/")
        data = self.buffer_cls.from_bytes(b"")
        fname = "zarr.json"
        store_dict = {p + fname: data for p in prefixes}

        await store._set_many(store_dict.items())

        for prefix in prefixes:
            observed = tuple(sorted(await _collect_aiterator(store.list_prefix(prefix))))
            expected: tuple[str, ...] = ()
            for key in store_dict:
                if key.startswith(prefix):
                    expected += (key,)
            expected = tuple(sorted(expected))
            assert observed == expected

    async def test_list_empty_path(self, store: S) -> None:
        """
        Verify that list and list_prefix work correctly when path is an empty string,
        i.e. no unwanted replacement occurs.
        """
        data = self.buffer_cls.from_bytes(b"")
        store_dict = {
            "foo/bar/zarr.json": data,
            "foo/bar/c/1": data,
            "foo/baz/c/0": data,
        }
        await store._set_many(store_dict.items())

        # Test list()
        observed_list = await _collect_aiterator(store.list())
        observed_list_sorted = sorted(observed_list)
        expected_list_sorted = sorted(store_dict.keys())
        assert observed_list_sorted == expected_list_sorted

        # Test list_prefix() with an empty prefix
        observed_prefix_empty = await _collect_aiterator(store.list_prefix(""))
        observed_prefix_empty_sorted = sorted(observed_prefix_empty)
        expected_prefix_empty_sorted = sorted(store_dict.keys())
        assert observed_prefix_empty_sorted == expected_prefix_empty_sorted

        # Test list_prefix() with a non-empty prefix
        observed_prefix = await _collect_aiterator(store.list_prefix("foo/bar/"))
        observed_prefix_sorted = sorted(observed_prefix)
        expected_prefix_sorted = sorted(k for k in store_dict if k.startswith("foo/bar/"))
        assert observed_prefix_sorted == expected_prefix_sorted

    async def test_list_dir(self, store: S) -> None:
        roots_and_keys: list[tuple[str, dict[str, Buffer]]] = [
            (
                "foo",
                {
                    "foo/zarr.json": self.buffer_cls.from_bytes(b"bar"),
                    "foo/c/1": self.buffer_cls.from_bytes(b"\x01"),
                },
            ),
            (
                "foo/bar",
                {
                    "foo/bar/foobar_first_child": self.buffer_cls.from_bytes(b"1"),
                    "foo/bar/foobar_second_child/zarr.json": self.buffer_cls.from_bytes(b"2"),
                },
            ),
        ]

        assert await _collect_aiterator(store.list_dir("")) == ()

        for root, store_dict in roots_and_keys:
            assert await _collect_aiterator(store.list_dir(root)) == ()

            await store._set_many(store_dict.items())

            keys_observed = await _collect_aiterator(store.list_dir(root))
            keys_expected = {k.removeprefix(root + "/").split("/")[0] for k in store_dict}
            assert sorted(keys_observed) == sorted(keys_expected)

            keys_observed = await _collect_aiterator(store.list_dir(root + "/"))
            assert sorted(keys_expected) == sorted(keys_observed)

    async def test_set_if_not_exists(self, store: S) -> None:
        key = "k"
        data_buf = self.buffer_cls.from_bytes(b"0000")
        await self.set(store, key, data_buf)

        new = self.buffer_cls.from_bytes(b"1111")
        await store.set_if_not_exists("k", new)  # no error

        result = await store.get(key, default_buffer_prototype())
        assert result == data_buf

        await store.set_if_not_exists("k2", new)  # no error

        result = await store.get("k2", default_buffer_prototype())
        assert result == new

    async def test_get_bytes(self, store: S) -> None:
        """
        Test that the get_bytes method reads bytes.
        """
        data = b"hello world"
        key = "zarr.json"
        await self.set(store, key, self.buffer_cls.from_bytes(data))
        assert await store._get_bytes(key, prototype=default_buffer_prototype()) == data
        with pytest.raises(FileNotFoundError):
            await store._get_bytes("nonexistent_key", prototype=default_buffer_prototype())

    def test_get_bytes_sync(self, store: S) -> None:
        """
        Test that the get_bytes_sync method reads bytes.
        """
        data = b"hello world"
        key = "zarr.json"
        sync(self.set(store, key, self.buffer_cls.from_bytes(data)))
        assert store._get_bytes_sync(key, prototype=default_buffer_prototype()) == data

    async def test_get_json(self, store: S) -> None:
        """
        Test that the get_json method reads json.
        """
        data = {"foo": "bar"}
        data_bytes = json.dumps(data).encode("utf-8")
        key = "zarr.json"
        await self.set(store, key, self.buffer_cls.from_bytes(data_bytes))
        assert await store._get_json(key, prototype=default_buffer_prototype()) == data

    def test_get_json_sync(self, store: S) -> None:
        """
        Test that the get_json method reads json.
        """
        data = {"foo": "bar"}
        data_bytes = json.dumps(data).encode("utf-8")
        key = "zarr.json"
        sync(self.set(store, key, self.buffer_cls.from_bytes(data_bytes)))
        assert store._get_json_sync(key, prototype=default_buffer_prototype()) == data

buffer_cls instance-attribute

buffer_cls: type[B]

store_cls instance-attribute

store_cls: type[S]

get abstractmethod async

get(store: S, key: str) -> Buffer

Retrieve a value from a storage backend, by key. This should not use any store methods. Bypassing the store methods allows them to be tested.

Source code in zarr/testing/store.py

@abstractmethod
async def get(self, store: S, key: str) -> Buffer:
    """
    Retrieve a value from a storage backend, by key.
    This should not use any store methods. Bypassing the store methods allows them to be
    tested.
    """
    ...

open_kwargs

open_kwargs(store_kwargs: dict[str, Any]) -> dict[str, Any]

Source code in zarr/testing/store.py

@pytest.fixture
def open_kwargs(self, store_kwargs: dict[str, Any]) -> dict[str, Any]:
    return store_kwargs

set abstractmethod async

set(store: S, key: str, value: Buffer) -> None

Insert a value into a storage backend, with a specific key. This should not use any store methods. Bypassing the store methods allows them to be tested.

Source code in zarr/testing/store.py

@abstractmethod
async def set(self, store: S, key: str, value: Buffer) -> None:
    """
    Insert a value into a storage backend, with a specific key.
    This should not use any store methods. Bypassing the store methods allows them to be
    tested.
    """
    ...

store async

store(open_kwargs: dict[str, Any]) -> Store

Source code in zarr/testing/store.py

@pytest.fixture
async def store(self, open_kwargs: dict[str, Any]) -> Store:
    return await self.store_cls.open(**open_kwargs)

store_kwargs abstractmethod

store_kwargs(*args: Any, **kwargs: Any) -> dict[str, Any]

Kwargs for instantiating a store

Source code in zarr/testing/store.py

@abstractmethod
@pytest.fixture
def store_kwargs(self, *args: Any, **kwargs: Any) -> dict[str, Any]:
    """Kwargs for instantiating a store"""
    ...

store_not_open async

store_not_open(store_kwargs: dict[str, Any]) -> Store

Source code in zarr/testing/store.py

@pytest.fixture
async def store_not_open(self, store_kwargs: dict[str, Any]) -> Store:
    return self.store_cls(**store_kwargs)

test_clear async

test_clear(store: S) -> None

Source code in zarr/testing/store.py

async def test_clear(self, store: S) -> None:
    await self.set(
        store, "key", self.buffer_cls.from_bytes(bytes("something", encoding="utf-8"))
    )
    await store.clear()
    assert await store.is_empty("")

test_delete async

test_delete(store: S) -> None

Source code in zarr/testing/store.py

async def test_delete(self, store: S) -> None:
    if not store.supports_deletes:
        pytest.skip("store does not support deletes")
    await store.set("foo/zarr.json", self.buffer_cls.from_bytes(b"bar"))
    assert await store.exists("foo/zarr.json")
    await store.delete("foo/zarr.json")
    assert not await store.exists("foo/zarr.json")

test_delete_dir async

test_delete_dir(store: S) -> None

Source code in zarr/testing/store.py

async def test_delete_dir(self, store: S) -> None:
    if not store.supports_deletes:
        pytest.skip("store does not support deletes")
    await store.set("zarr.json", self.buffer_cls.from_bytes(b"root"))
    await store.set("foo-bar/zarr.json", self.buffer_cls.from_bytes(b"root"))
    await store.set("foo/zarr.json", self.buffer_cls.from_bytes(b"bar"))
    await store.set("foo/c/0", self.buffer_cls.from_bytes(b"chunk"))
    await store.delete_dir("foo")
    assert await store.exists("zarr.json")
    assert await store.exists("foo-bar/zarr.json")
    assert not await store.exists("foo/zarr.json")
    assert not await store.exists("foo/c/0")

test_delete_nonexistent_key_does_not_raise async

test_delete_nonexistent_key_does_not_raise(
    store: S,
) -> None

Source code in zarr/testing/store.py

async def test_delete_nonexistent_key_does_not_raise(self, store: S) -> None:
    if not store.supports_deletes:
        pytest.skip("store does not support deletes")
    await store.delete("nonexistent_key")

test_exists async

test_exists(store: S) -> None

Source code in zarr/testing/store.py

async def test_exists(self, store: S) -> None:
    assert not await store.exists("foo")
    await store.set("foo/zarr.json", self.buffer_cls.from_bytes(b"bar"))
    assert await store.exists("foo/zarr.json")

test_get async

test_get(
    store: S,
    key: str,
    data: bytes,
    byte_range: ByteRequest,
) -> None

Ensure that data can be read from the store using the store.get method.

Source code in zarr/testing/store.py

@pytest.mark.parametrize("key", ["c/0", "foo/c/0.0", "foo/0/0"])
@pytest.mark.parametrize(
    ("data", "byte_range"),
    [
        (b"\x01\x02\x03\x04", None),
        (b"\x01\x02\x03\x04", RangeByteRequest(1, 4)),
        (b"\x01\x02\x03\x04", OffsetByteRequest(1)),
        (b"\x01\x02\x03\x04", SuffixByteRequest(1)),
        (b"", None),
    ],
)
async def test_get(self, store: S, key: str, data: bytes, byte_range: ByteRequest) -> None:
    """
    Ensure that data can be read from the store using the store.get method.
    """
    data_buf = self.buffer_cls.from_bytes(data)
    await self.set(store, key, data_buf)
    observed = await store.get(key, prototype=default_buffer_prototype(), byte_range=byte_range)
    start, stop = _normalize_byte_range_index(data_buf, byte_range=byte_range)
    expected = data_buf[start:stop]
    assert_bytes_equal(observed, expected)

test_get_bytes async

test_get_bytes(store: S) -> None

Test that the get_bytes method reads bytes.

Source code in zarr/testing/store.py

async def test_get_bytes(self, store: S) -> None:
    """
    Test that the get_bytes method reads bytes.
    """
    data = b"hello world"
    key = "zarr.json"
    await self.set(store, key, self.buffer_cls.from_bytes(data))
    assert await store._get_bytes(key, prototype=default_buffer_prototype()) == data
    with pytest.raises(FileNotFoundError):
        await store._get_bytes("nonexistent_key", prototype=default_buffer_prototype())

test_get_bytes_sync

test_get_bytes_sync(store: S) -> None

Test that the get_bytes_sync method reads bytes.

Source code in zarr/testing/store.py

def test_get_bytes_sync(self, store: S) -> None:
    """
    Test that the get_bytes_sync method reads bytes.
    """
    data = b"hello world"
    key = "zarr.json"
    sync(self.set(store, key, self.buffer_cls.from_bytes(data)))
    assert store._get_bytes_sync(key, prototype=default_buffer_prototype()) == data

test_get_json async

test_get_json(store: S) -> None

Test that the get_json method reads json.

Source code in zarr/testing/store.py

async def test_get_json(self, store: S) -> None:
    """
    Test that the get_json method reads json.
    """
    data = {"foo": "bar"}
    data_bytes = json.dumps(data).encode("utf-8")
    key = "zarr.json"
    await self.set(store, key, self.buffer_cls.from_bytes(data_bytes))
    assert await store._get_json(key, prototype=default_buffer_prototype()) == data

test_get_json_sync

test_get_json_sync(store: S) -> None

Test that the get_json method reads json.

Source code in zarr/testing/store.py

def test_get_json_sync(self, store: S) -> None:
    """
    Test that the get_json method reads json.
    """
    data = {"foo": "bar"}
    data_bytes = json.dumps(data).encode("utf-8")
    key = "zarr.json"
    sync(self.set(store, key, self.buffer_cls.from_bytes(data_bytes)))
    assert store._get_json_sync(key, prototype=default_buffer_prototype()) == data

test_get_many async

test_get_many(store: S) -> None

Ensure that multiple keys can be retrieved at once with the _get_many method.

Source code in zarr/testing/store.py

async def test_get_many(self, store: S) -> None:
    """
    Ensure that multiple keys can be retrieved at once with the _get_many method.
    """
    keys = tuple(map(str, range(10)))
    values = tuple(f"{k}".encode() for k in keys)
    for k, v in zip(keys, values, strict=False):
        await self.set(store, k, self.buffer_cls.from_bytes(v))
    observed_buffers = await _collect_aiterator(
        store._get_many(
            zip(
                keys,
                (default_buffer_prototype(),) * len(keys),
                (None,) * len(keys),
                strict=False,
            )
        )
    )
    observed_kvs = sorted(((k, b.to_bytes()) for k, b in observed_buffers))  # type: ignore[union-attr]
    expected_kvs = sorted(((k, b) for k, b in zip(keys, values, strict=False)))
    assert observed_kvs == expected_kvs

test_get_not_open async

test_get_not_open(store_not_open: S) -> None

Ensure that data can be read from the store that isn't yet open using the store.get method.

Source code in zarr/testing/store.py

async def test_get_not_open(self, store_not_open: S) -> None:
    """
    Ensure that data can be read from the store that isn't yet open using the store.get method.
    """
    assert not store_not_open._is_open
    data_buf = self.buffer_cls.from_bytes(b"\x01\x02\x03\x04")
    key = "c/0"
    await self.set(store_not_open, key, data_buf)
    observed = await store_not_open.get(key, prototype=default_buffer_prototype())
    assert_bytes_equal(observed, data_buf)

test_get_partial_values async

test_get_partial_values(
    store: S, key_ranges: list[tuple[str, ByteRequest]]
) -> None

Source code in zarr/testing/store.py

@pytest.mark.parametrize(
    "key_ranges",
    [
        [],
        [("zarr.json", RangeByteRequest(0, 2))],
        [("c/0", RangeByteRequest(0, 2)), ("zarr.json", None)],
        [
            ("c/0/0", RangeByteRequest(0, 2)),
            ("c/0/1", SuffixByteRequest(2)),
            ("c/0/2", OffsetByteRequest(2)),
        ],
    ],
)
async def test_get_partial_values(
    self, store: S, key_ranges: list[tuple[str, ByteRequest]]
) -> None:
    # put all of the data
    for key, _ in key_ranges:
        await self.set(store, key, self.buffer_cls.from_bytes(bytes(key, encoding="utf-8")))

    # read back just part of it
    observed_maybe = await store.get_partial_values(
        prototype=default_buffer_prototype(), key_ranges=key_ranges
    )

    observed: list[Buffer] = []
    expected: list[Buffer] = []

    for obs in observed_maybe:
        assert obs is not None
        observed.append(obs)

    for idx in range(len(observed)):
        key, byte_range = key_ranges[idx]
        result = await store.get(
            key, prototype=default_buffer_prototype(), byte_range=byte_range
        )
        assert result is not None
        expected.append(result)

    assert all(
        obs.to_bytes() == exp.to_bytes() for obs, exp in zip(observed, expected, strict=True)
    )

test_get_raises async

test_get_raises(store: S) -> None

Ensure that a ValueError is raise for invalid byte range syntax

Source code in zarr/testing/store.py

async def test_get_raises(self, store: S) -> None:
    """
    Ensure that a ValueError is raise for invalid byte range syntax
    """
    data_buf = self.buffer_cls.from_bytes(b"\x01\x02\x03\x04")
    await self.set(store, "c/0", data_buf)
    with pytest.raises((ValueError, TypeError), match=r"Unexpected byte_range, got.*"):
        await store.get("c/0", prototype=default_buffer_prototype(), byte_range=(0, 2))  # type: ignore[arg-type]

test_getsize async

test_getsize(store: S, key: str, data: bytes) -> None

Test the result of store.getsize().

Source code in zarr/testing/store.py

@pytest.mark.parametrize("key", ["c/0", "foo/c/0.0", "foo/0/0"])
@pytest.mark.parametrize("data", [b"\x01\x02\x03\x04", b""])
async def test_getsize(self, store: S, key: str, data: bytes) -> None:
    """
    Test the result of store.getsize().
    """
    data_buf = self.buffer_cls.from_bytes(data)
    expected = len(data_buf)
    await self.set(store, key, data_buf)
    observed = await store.getsize(key)
    assert observed == expected

test_getsize_prefix async

test_getsize_prefix(store: S) -> None

Test the result of store.getsize_prefix().

Source code in zarr/testing/store.py

async def test_getsize_prefix(self, store: S) -> None:
    """
    Test the result of store.getsize_prefix().
    """
    data_buf = self.buffer_cls.from_bytes(b"\x01\x02\x03\x04")
    keys = ["c/0/0", "c/0/1", "c/1/0", "c/1/1"]
    keys_values = [(k, data_buf) for k in keys]
    await store._set_many(keys_values)
    expected = len(data_buf) * len(keys)
    observed = await store.getsize_prefix("c")
    assert observed == expected

test_getsize_raises async

test_getsize_raises(store: S) -> None

Test that getsize() raise a FileNotFoundError if the key doesn't exist.

Source code in zarr/testing/store.py

async def test_getsize_raises(self, store: S) -> None:
    """
    Test that getsize() raise a FileNotFoundError if the key doesn't exist.
    """
    with pytest.raises(FileNotFoundError):
        await store.getsize("c/1000")

test_is_empty async

test_is_empty(store: S) -> None

Source code in zarr/testing/store.py

async def test_is_empty(self, store: S) -> None:
    assert await store.is_empty("")
    await self.set(
        store, "foo/bar", self.buffer_cls.from_bytes(bytes("something", encoding="utf-8"))
    )
    assert not await store.is_empty("")
    assert await store.is_empty("fo")
    assert not await store.is_empty("foo/")
    assert not await store.is_empty("foo")
    assert await store.is_empty("spam/")

test_list async

test_list(store: S) -> None

Source code in zarr/testing/store.py

async def test_list(self, store: S) -> None:
    assert await _collect_aiterator(store.list()) == ()
    prefix = "foo"
    data = self.buffer_cls.from_bytes(b"")
    store_dict = {
        prefix + "/zarr.json": data,
        **{prefix + f"/c/{idx}": data for idx in range(10)},
    }
    await store._set_many(store_dict.items())
    expected_sorted = sorted(store_dict.keys())
    observed = await _collect_aiterator(store.list())
    observed_sorted = sorted(observed)
    assert observed_sorted == expected_sorted

test_list_dir async

test_list_dir(store: S) -> None

Source code in zarr/testing/store.py

async def test_list_dir(self, store: S) -> None:
    roots_and_keys: list[tuple[str, dict[str, Buffer]]] = [
        (
            "foo",
            {
                "foo/zarr.json": self.buffer_cls.from_bytes(b"bar"),
                "foo/c/1": self.buffer_cls.from_bytes(b"\x01"),
            },
        ),
        (
            "foo/bar",
            {
                "foo/bar/foobar_first_child": self.buffer_cls.from_bytes(b"1"),
                "foo/bar/foobar_second_child/zarr.json": self.buffer_cls.from_bytes(b"2"),
            },
        ),
    ]

    assert await _collect_aiterator(store.list_dir("")) == ()

    for root, store_dict in roots_and_keys:
        assert await _collect_aiterator(store.list_dir(root)) == ()

        await store._set_many(store_dict.items())

        keys_observed = await _collect_aiterator(store.list_dir(root))
        keys_expected = {k.removeprefix(root + "/").split("/")[0] for k in store_dict}
        assert sorted(keys_observed) == sorted(keys_expected)

        keys_observed = await _collect_aiterator(store.list_dir(root + "/"))
        assert sorted(keys_expected) == sorted(keys_observed)

test_list_empty_path async

test_list_empty_path(store: S) -> None

Verify that list and list_prefix work correctly when path is an empty string, i.e. no unwanted replacement occurs.

Source code in zarr/testing/store.py

async def test_list_empty_path(self, store: S) -> None:
    """
    Verify that list and list_prefix work correctly when path is an empty string,
    i.e. no unwanted replacement occurs.
    """
    data = self.buffer_cls.from_bytes(b"")
    store_dict = {
        "foo/bar/zarr.json": data,
        "foo/bar/c/1": data,
        "foo/baz/c/0": data,
    }
    await store._set_many(store_dict.items())

    # Test list()
    observed_list = await _collect_aiterator(store.list())
    observed_list_sorted = sorted(observed_list)
    expected_list_sorted = sorted(store_dict.keys())
    assert observed_list_sorted == expected_list_sorted

    # Test list_prefix() with an empty prefix
    observed_prefix_empty = await _collect_aiterator(store.list_prefix(""))
    observed_prefix_empty_sorted = sorted(observed_prefix_empty)
    expected_prefix_empty_sorted = sorted(store_dict.keys())
    assert observed_prefix_empty_sorted == expected_prefix_empty_sorted

    # Test list_prefix() with a non-empty prefix
    observed_prefix = await _collect_aiterator(store.list_prefix("foo/bar/"))
    observed_prefix_sorted = sorted(observed_prefix)
    expected_prefix_sorted = sorted(k for k in store_dict if k.startswith("foo/bar/"))
    assert observed_prefix_sorted == expected_prefix_sorted

test_list_prefix async

test_list_prefix(store: S) -> None

Test that the list_prefix method works as intended. Given a prefix, it should return all the keys in storage that start with this prefix.

Source code in zarr/testing/store.py

async def test_list_prefix(self, store: S) -> None:
    """
    Test that the `list_prefix` method works as intended. Given a prefix, it should return
    all the keys in storage that start with this prefix.
    """
    prefixes = ("", "a/", "a/b/", "a/b/c/")
    data = self.buffer_cls.from_bytes(b"")
    fname = "zarr.json"
    store_dict = {p + fname: data for p in prefixes}

    await store._set_many(store_dict.items())

    for prefix in prefixes:
        observed = tuple(sorted(await _collect_aiterator(store.list_prefix(prefix))))
        expected: tuple[str, ...] = ()
        for key in store_dict:
            if key.startswith(prefix):
                expected += (key,)
        expected = tuple(sorted(expected))
        assert observed == expected

test_read_only_store_raises async

test_read_only_store_raises(
    open_kwargs: dict[str, Any],
) -> None

Source code in zarr/testing/store.py

async def test_read_only_store_raises(self, open_kwargs: dict[str, Any]) -> None:
    kwargs = {**open_kwargs, "read_only": True}
    store = await self.store_cls.open(**kwargs)
    assert store.read_only

    # set
    with pytest.raises(
        ValueError, match="store was opened in read-only mode and does not support writing"
    ):
        await store.set("foo", self.buffer_cls.from_bytes(b"bar"))

    # delete
    with pytest.raises(
        ValueError, match="store was opened in read-only mode and does not support writing"
    ):
        await store.delete("foo")

test_serializable_store async

test_serializable_store(store: S) -> None

Source code in zarr/testing/store.py

async def test_serializable_store(self, store: S) -> None:
    new_store: S = pickle.loads(pickle.dumps(store))
    assert new_store == store
    assert new_store.read_only == store.read_only
    # quickly roundtrip data to a key to test that new store works
    data_buf = self.buffer_cls.from_bytes(b"\x01\x02\x03\x04")
    key = "foo"
    await store.set(key, data_buf)
    observed = await store.get(key, prototype=default_buffer_prototype())
    assert_bytes_equal(observed, data_buf)

test_set async

test_set(store: S, key: str, data: bytes) -> None

Ensure that data can be written to the store using the store.set method.

Source code in zarr/testing/store.py

@pytest.mark.parametrize("key", ["zarr.json", "c/0", "foo/c/0.0", "foo/0/0"])
@pytest.mark.parametrize("data", [b"\x01\x02\x03\x04", b""])
async def test_set(self, store: S, key: str, data: bytes) -> None:
    """
    Ensure that data can be written to the store using the store.set method.
    """
    assert not store.read_only
    data_buf = self.buffer_cls.from_bytes(data)
    await store.set(key, data_buf)
    observed = await self.get(store, key)
    assert_bytes_equal(observed, data_buf)

test_set_if_not_exists async

test_set_if_not_exists(store: S) -> None

Source code in zarr/testing/store.py

async def test_set_if_not_exists(self, store: S) -> None:
    key = "k"
    data_buf = self.buffer_cls.from_bytes(b"0000")
    await self.set(store, key, data_buf)

    new = self.buffer_cls.from_bytes(b"1111")
    await store.set_if_not_exists("k", new)  # no error

    result = await store.get(key, default_buffer_prototype())
    assert result == data_buf

    await store.set_if_not_exists("k2", new)  # no error

    result = await store.get("k2", default_buffer_prototype())
    assert result == new

test_set_many async

test_set_many(store: S) -> None

Test that a dict of key : value pairs can be inserted into the store via the _set_many method.

Source code in zarr/testing/store.py

async def test_set_many(self, store: S) -> None:
    """
    Test that a dict of key : value pairs can be inserted into the store via the
    `_set_many` method.
    """
    keys = ["zarr.json", "c/0", "foo/c/0.0", "foo/0/0"]
    data_buf = [self.buffer_cls.from_bytes(k.encode()) for k in keys]
    store_dict = dict(zip(keys, data_buf, strict=True))
    await store._set_many(store_dict.items())
    for k, v in store_dict.items():
        assert (await self.get(store, k)).to_bytes() == v.to_bytes()

test_set_not_open async

test_set_not_open(store_not_open: S) -> None

Ensure that data can be written to the store that's not yet open using the store.set method.

Source code in zarr/testing/store.py

async def test_set_not_open(self, store_not_open: S) -> None:
    """
    Ensure that data can be written to the store that's not yet open using the store.set method.
    """
    assert not store_not_open._is_open
    data_buf = self.buffer_cls.from_bytes(b"\x01\x02\x03\x04")
    key = "c/0"
    await store_not_open.set(key, data_buf)
    observed = await self.get(store_not_open, key)
    assert_bytes_equal(observed, data_buf)

test_store_context_manager async

test_store_context_manager(
    open_kwargs: dict[str, Any],
) -> None

Source code in zarr/testing/store.py

async def test_store_context_manager(self, open_kwargs: dict[str, Any]) -> None:
    # Test that the context manager closes the store
    with await self.store_cls.open(**open_kwargs) as store:
        assert store._is_open
        # Test trying to open an already open store
        with pytest.raises(ValueError, match="store is already open"):
            await store._open()
    assert not store._is_open

test_store_eq

test_store_eq(
    store: S, store_kwargs: dict[str, Any]
) -> None

Source code in zarr/testing/store.py

def test_store_eq(self, store: S, store_kwargs: dict[str, Any]) -> None:
    # check self equality
    assert store == store

    # check store equality with same inputs
    # asserting this is important for being able to compare (de)serialized stores
    store2 = self.store_cls(**store_kwargs)
    assert store == store2

test_store_open_read_only async

test_store_open_read_only(
    open_kwargs: dict[str, Any], read_only: bool
) -> None

Source code in zarr/testing/store.py

@pytest.mark.parametrize("read_only", [True, False])
async def test_store_open_read_only(self, open_kwargs: dict[str, Any], read_only: bool) -> None:
    open_kwargs["read_only"] = read_only
    store = await self.store_cls.open(**open_kwargs)
    assert store._is_open
    assert store.read_only == read_only

test_store_read_only

test_store_read_only(store: S) -> None

Source code in zarr/testing/store.py

def test_store_read_only(self, store: S) -> None:
    assert not store.read_only

    with pytest.raises(AttributeError):
        store.read_only = False  # type: ignore[misc]

test_store_repr abstractmethod

test_store_repr(store: S) -> None

Source code in zarr/testing/store.py

@abstractmethod
def test_store_repr(self, store: S) -> None: ...

test_store_supports_listing abstractmethod

test_store_supports_listing(store: S) -> None

Source code in zarr/testing/store.py

@abstractmethod
def test_store_supports_listing(self, store: S) -> None: ...

test_store_supports_partial_writes

test_store_supports_partial_writes(store: S) -> None

Source code in zarr/testing/store.py

def test_store_supports_partial_writes(self, store: S) -> None:
    assert not store.supports_partial_writes

test_store_supports_writes abstractmethod

test_store_supports_writes(store: S) -> None

Source code in zarr/testing/store.py

@abstractmethod
def test_store_supports_writes(self, store: S) -> None: ...

test_store_type

test_store_type(store: S) -> None

Source code in zarr/testing/store.py

def test_store_type(self, store: S) -> None:
    assert isinstance(store, Store)
    assert isinstance(store, self.store_cls)

test_with_read_only_store async

test_with_read_only_store(
    open_kwargs: dict[str, Any],
) -> None

Source code in zarr/testing/store.py

async def test_with_read_only_store(self, open_kwargs: dict[str, Any]) -> None:
    kwargs = {**open_kwargs, "read_only": True}
    store = await self.store_cls.open(**kwargs)
    assert store.read_only

    # Test that you cannot write to a read-only store
    with pytest.raises(
        ValueError, match="store was opened in read-only mode and does not support writing"
    ):
        await store.set("foo", self.buffer_cls.from_bytes(b"bar"))

    # Check if the store implements with_read_only
    try:
        writer = store.with_read_only(read_only=False)
    except NotImplementedError:
        # Test that stores that do not implement with_read_only raise NotImplementedError with the correct message
        with pytest.raises(
            NotImplementedError,
            match=f"with_read_only is not implemented for the {type(store)} store type.",
        ):
            store.with_read_only(read_only=False)
        return

    # Test that you can write to a new store copy
    assert not writer._is_open
    assert not writer.read_only
    await writer.set("foo", self.buffer_cls.from_bytes(b"bar"))
    await writer.delete("foo")

    # Test that you cannot write to the original store
    assert store.read_only
    with pytest.raises(
        ValueError, match="store was opened in read-only mode and does not support writing"
    ):
        await store.set("foo", self.buffer_cls.from_bytes(b"bar"))
    with pytest.raises(
        ValueError, match="store was opened in read-only mode and does not support writing"
    ):
        await store.delete("foo")

    # Test that you cannot write to a read-only store copy
    reader = store.with_read_only(read_only=True)
    assert reader.read_only
    with pytest.raises(
        ValueError, match="store was opened in read-only mode and does not support writing"
    ):
        await reader.set("foo", self.buffer_cls.from_bytes(b"bar"))
    with pytest.raises(
        ValueError, match="store was opened in read-only mode and does not support writing"
    ):
        await reader.delete("foo")