zarr.testing.store#

Classes#

Module Contents#

class zarr.testing.store.StoreTests[source]#

Bases: Generic[S, B]

abstract get(store: S, key: str) zarr.core.buffer.Buffer[source]#
Async:

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

abstract set(store: S, key: str, value: zarr.core.buffer.Buffer) None[source]#
Async:

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

async store(store_kwargs: dict[str, Any]) zarr.abc.store.Store[source]#
store_kwargs() dict[str, Any][source]#
async test_clear(store: S) None[source]#
async test_delete(store: S) None[source]#
async test_delete_dir(store: S) None[source]#
async test_exists(store: S) None[source]#
async test_get(
store: S,
key: str,
data: bytes,
byte_range: zarr.abc.store.ByteRequest,
) None[source]#

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

async test_get_many(store: S) None[source]#

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

async test_get_partial_values(
store: S,
key_ranges: list[tuple[str, zarr.abc.store.ByteRequest]],
) None[source]#
async test_is_empty(store: S) None[source]#
async test_list(store: S) None[source]#
async test_list_dir(store: S) None[source]#
async test_list_prefix(store: S) None[source]#

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.

async test_read_only_store_raises(store_kwargs: dict[str, Any]) None[source]#
test_serializable_store(store: S) None[source]#
async test_set(store: S, key: str, data: bytes) None[source]#

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

async test_set_if_not_exists(store: S) None[source]#
async test_set_many(store: S) None[source]#

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

test_store_eq(store: S, store_kwargs: dict[str, Any]) None[source]#
async test_store_open_read_only(store_kwargs: dict[str, Any], read_only: bool) None[source]#
test_store_read_only(store: S) None[source]#
abstract test_store_repr(store: S) None[source]#
abstract test_store_supports_listing(store: S) None[source]#
abstract test_store_supports_partial_writes(store: S) None[source]#
abstract test_store_supports_writes(store: S) None[source]#
test_store_type(store: S) None[source]#
buffer_cls: type[B][source]#
store_cls: type[S][source]#