Merge pull request #78 from dClimate/ci/update-zarr-3.0.9

Ci/update zarr 3.0.9

Author: Faolain

py_hamt/encryption_hamt_store.py

@@ -110,6 +110,19 @@ def __init__(
         self.header = header
         self.metadata_read_cache: dict[str, bytes] = {}
 
+    def with_read_only(self, read_only: bool = False) -> "SimpleEncryptedZarrHAMTStore":
+        if read_only == self.read_only:
+            return self
+
+        clone = type(self).__new__(type(self))
+        clone.hamt = self.hamt
+        clone.encryption_key = self.encryption_key
+        clone.header = self.header
+        clone.metadata_read_cache = self.metadata_read_cache
+        clone._forced_read_only = read_only  # safe; attribute is declared
+        zarr.abc.store.Store.__init__(clone, read_only=read_only)
+        return clone
+
     def _encrypt(self, val: bytes) -> bytes:
         """Encrypts data using ChaCha20-Poly1305."""
         nonce = get_random_bytes(24)  # XChaCha20 uses a 24-byte nonce

py_hamt/store_httpx.py

@@ -232,9 +232,21 @@ def __init__(
     # helper: get or create the client bound to the current running loop    #
     # --------------------------------------------------------------------- #
     def _loop_client(self) -> httpx.AsyncClient:
-        """Get or create a client for the current event loop."""
+        """Get or create a client for the current event loop.
+
+        If the instance was previously closed but owns its clients, a fresh
+        client mapping is lazily created on demand.  Users that supplied their
+        own ``httpx.AsyncClient`` still receive an error when the instance has
+        been closed, as we cannot safely recreate their client.
+        """
         if self._closed:
-            raise RuntimeError("KuboCAS is closed; create a new instance")
+            if not self._owns_client:
+                raise RuntimeError("KuboCAS is closed; create a new instance")
+            # We previously closed all internally-owned clients. Reset the
+            # state so that new clients can be created lazily.
+            self._closed = False
+            self._client_per_loop = {}
+
         loop: asyncio.AbstractEventLoop = asyncio.get_running_loop()
         try:
             return self._client_per_loop[loop]
@@ -245,7 +257,7 @@ def _loop_client(self) -> httpx.AsyncClient:
                 headers=self._default_headers,
                 auth=self._default_auth,
                 limits=httpx.Limits(max_connections=64, max_keepalive_connections=32),
-                # Uncomment when they finally support Robost HTTP/2 GOAWAY responses
+                # Uncomment when they finally support Robust HTTP/2 GOAWAY responses
                 # http2=True,
             )
             self._client_per_loop[loop] = client

py_hamt/zarr_hamt_store.py

@@ -51,6 +51,8 @@ class ZarrHAMTStore(zarr.abc.store.Store):
     ```
     """
 
+    _forced_read_only: bool | None = None  # sentinel for wrapper clones
+
     def __init__(self, hamt: HAMT, read_only: bool = False) -> None:
         """
         ### `hamt` and `read_only`
@@ -79,10 +81,36 @@ def __init__(self, hamt: HAMT, read_only: bool = False) -> None:
         """@private"""
 
     @property
-    def read_only(self) -> bool:
-        """@private"""
+    def read_only(self) -> bool:  # type: ignore[override]
+        if self._forced_read_only is not None:  # instance attr overrides
+            return self._forced_read_only
         return self.hamt.read_only
 
+    def with_read_only(self, read_only: bool = False) -> "ZarrHAMTStore":
+        """
+        Return this store (if the flag already matches) or a *shallow*
+        clone that presents the requested read‑only status.
+
+        The clone **shares** the same :class:`~py_hamt.hamt.HAMT`
+        instance; no flushing, network traffic or async work is done.
+        """
+        # Fast path
+        if read_only == self.read_only:
+            return self  # Same mode, return same instance
+
+        # Create new instance with different read_only flag
+        # Creates a *bare* instance without running its __init__
+        clone = type(self).__new__(type(self))
+
+        # Copy attributes that matter
+        clone.hamt = self.hamt  # Share the HAMT
+        clone._forced_read_only = read_only
+        clone.metadata_read_cache = self.metadata_read_cache.copy()
+
+        # Re‑initialise the zarr base class so that Zarr sees the flag
+        zarr.abc.store.Store.__init__(clone, read_only=read_only)
+        return clone
+
     def __eq__(self, other: object) -> bool:
         """@private"""
         if not isinstance(other, ZarrHAMTStore):
@@ -145,6 +173,9 @@ def supports_partial_writes(self) -> bool:
 
     async def set(self, key: str, value: zarr.core.buffer.Buffer) -> None:
         """@private"""
+        if self.read_only:
+            raise Exception("Cannot write to a read-only store.")
+
         if key in self.metadata_read_cache:
             self.metadata_read_cache[key] = value.to_bytes()
         await self.hamt.set(key, value.to_bytes())
@@ -167,6 +198,8 @@ def supports_deletes(self) -> bool:
 
     async def delete(self, key: str) -> None:
         """@private"""
+        if self.read_only:
+            raise Exception("Cannot write to a read-only store.")
         try:
             await self.hamt.delete(key)
             # In practice these lines never seem to be needed, creating and appending data are the only operations most zarrs actually undergo

pyproject.toml

@@ -9,7 +9,7 @@ dependencies = [
     "dag-cbor>=0.3.3",
     "msgspec>=0.18.6",
     "multiformats[full]>=0.3.1.post4",
-    "zarr>=3.0.8",
+    "zarr==3.0.9",
     "pycryptodome>=3.21.0",
 ]
 

tests/test_kubocas_session.py

@@ -3,6 +3,7 @@
 import unittest
 from threading import Event, Thread
 
+import httpx
 import pytest
 
 from py_hamt import KuboCAS
@@ -189,18 +190,34 @@ def fake_run(coro):
 
 
 @pytest.mark.asyncio
-async def test_loop_client_raises_after_close():
-    """
-    Verify that calling _loop_client() on a closed KuboCAS instance
-    raises a RuntimeError.
-    """
-    # Arrange: Create a KuboCAS instance
+async def test_loop_client_reopens_after_close():
+    """Calling _loop_client() after aclose() recreates a fresh client."""
     cas = KuboCAS()
 
-    # Act: Close the instance. This should set the internal _closed flag.
+    first = cas._loop_client()
     await cas.aclose()
 
-    # Assert: Check that calling _loop_client again raises the expected error.
+    # Should no longer raise; instead a new client is created.
+    reopened = cas._loop_client()
+    assert isinstance(reopened, httpx.AsyncClient)
+    assert reopened is not first
+    assert cas._closed is False
+
+    await cas.aclose()
+
+
+@pytest.mark.asyncio
+async def test_loop_client_rejects_reuse_of_external_client(global_client_session):
+    """Calling _loop_client() after aclose() raises when client is user-supplied."""
+    cas = KuboCAS(
+        client=global_client_session,
+        rpc_base_url="http://127.0.0.1:5001",
+        gateway_base_url="http://127.0.0.1:8080",
+    )
+    assert cas._loop_client() is global_client_session
+
+    await cas.aclose()
+    cas._closed = True  # simulate closed instance with external client
     with pytest.raises(RuntimeError, match="KuboCAS is closed; create a new instance"):
         cas._loop_client()
 

tests/test_read_only_guards.py

@@ -0,0 +1,71 @@
+# tests/test_read_only_guards.py
+import numpy as np
+import pytest
+from Crypto.Random import get_random_bytes
+
+from py_hamt import HAMT, InMemoryCAS, SimpleEncryptedZarrHAMTStore, ZarrHAMTStore
+
+
+# ---------- helpers ----------------------------------------------------
+async def _rw_plain():
+    cas = InMemoryCAS()
+    hamt = await HAMT.build(cas=cas, values_are_bytes=True)
+    return ZarrHAMTStore(hamt, read_only=False)
+
+
+async def _rw_enc():
+    cas = InMemoryCAS()
+    hamt = await HAMT.build(cas=cas, values_are_bytes=True)
+    key, hdr = get_random_bytes(32), b"hdr"
+    return SimpleEncryptedZarrHAMTStore(hamt, False, key, hdr)
+
+
+# ---------- plain store ------------------------------------------------
+@pytest.mark.asyncio
+async def test_plain_read_only_guards():
+    rw = await _rw_plain()
+    ro = rw.with_read_only(True)
+
+    assert ro.read_only is True
+    with pytest.raises(Exception):
+        await ro.set("k", np.array([1], dtype="u1"))
+    with pytest.raises(Exception):
+        await ro.delete("k")
+
+
+@pytest.mark.asyncio
+async def test_plain_with_same_flag_returns_self():
+    rw = await _rw_plain()
+    assert rw.with_read_only(False) is rw  # early‑return path
+
+
+@pytest.mark.asyncio
+async def test_roundtrip_plain_store():
+    rw = await _rw_plain()  # writable store
+    ro = rw.with_read_only(True)  # clone → RO
+    assert ro.read_only is True
+    assert ro.hamt is rw.hamt
+
+    # idempotent: RO→RO returns same object
+    assert ro.with_read_only(True) is ro
+
+    # back to RW (new wrapper)
+    rw2 = ro.with_read_only(False)
+    assert rw2.read_only is False and rw2 is not ro
+    assert rw2.hamt is rw.hamt
+
+    # guard: cannot write through RO wrapper
+    with pytest.raises(Exception):
+        await ro.set("k", np.array([0], dtype="u1"))
+
+
+# ---------- encrypted store -------------------------------------------
+@pytest.mark.asyncio
+async def test_encrypted_read_only_guards_and_self():
+    rw = await _rw_enc()
+    assert rw.with_read_only(False) is rw  # same‑flag path
+    ro = rw.with_read_only(True)
+    with pytest.raises(Exception):
+        await ro.set("k", np.array([2], dtype="u1"))
+    with pytest.raises(Exception):
+        await ro.delete("k")

tests/test_zarr_ipfs.py

@@ -189,3 +189,24 @@ async def test_list_dir_dedup():
     await hamt.set("foo/bar/1", b"")
     results = [n async for n in zhs.list_dir("foo/")]
     assert results == ["bar"]  # no duplicates
+
+
+@pytest.mark.asyncio
+async def test_open_rw_store_triggers_helper():
+    """
+    A write‑enabled ZarrHAMTStore must open cleanly in read‑mode.
+    Behaviour before 3.2.0:  ValueError  → helper missing.
+    Behaviour after  3.2.0:  success     → helper present.
+    """
+    # --- 1. create a small dataset and a RW HAMT backed by in‑memory CAS
+    cas = InMemoryCAS()
+    hamt = await HAMT.build(cas=cas, values_are_bytes=True)  # read/write
+    store_rw = ZarrHAMTStore(hamt, read_only=False)
+
+    ds = xr.Dataset({"x": ("t", np.arange(3))})
+    ds.to_zarr(store=store_rw, mode="w", zarr_format=3)
+
+    # --- 2. try to re‑open **the same write‑enabled store** in *read* mode
+    #       – this calls Store.with_read_only(True) internally
+    reopened = xr.open_zarr(store=store_rw)  # <-- MUST NOT raise
+    assert reopened.x.shape == (3,)  # sanity check

tests/testing_utils.py

@@ -169,7 +169,7 @@ def create_ipfs():
     if client is None:
         pytest.skip("Neither IPFS daemon nor Docker available – skipping IPFS tests")
 
-    image = "ipfs/kubo:v0.35.0"
+    image = "ipfs/kubo:v0.36.0"
     rpc_p = _free_port()
     gw_p = _free_port()