diff --git a/changes/4450.bugfix.md b/changes/4450.bugfix.md new file mode 100644 index 0000000000..4364d1a9a6 --- /dev/null +++ b/changes/4450.bugfix.md @@ -0,0 +1 @@ +Fixed several ways `ZipStore` could lose data or fail. A store created with `mode="w"` or `mode="x"` now keeps its entries when it is used again after `close()`, `move()`, or unpickling, instead of truncating the archive. Every method now opens the archive on first use, under a lock, so `get`, `get_partial_values`, `set_if_not_exists`, and `clear` no longer raise `AttributeError` on a store that was never opened, and threads sharing a store no longer open the archive twice. An unpickled store now opens its archive on first use rather than when it is unpickled, so writing a dask array to a `ZipStore` no longer produces an unreadable file. Where reopening an archive would silently drop its entries, the store now raises instead: `RuntimeError` after a failed `close()`, `zipfile.BadZipFile` if the file is not a complete zip archive, and `io.UnsupportedOperation` for a file object that is not readable and seekable. The type of the `mode` parameter now includes `"x"`. Builds on [#3593](https://github.com/zarr-developers/zarr-python/pull/3593) by [@oelhammouchi](https://github.com/oelhammouchi). diff --git a/src/zarr/storage/_zip.py b/src/zarr/storage/_zip.py index 6b9ebca561..e29d2ce496 100644 --- a/src/zarr/storage/_zip.py +++ b/src/zarr/storage/_zip.py @@ -21,7 +21,7 @@ if TYPE_CHECKING: from collections.abc import AsyncIterator, Iterable -ZipStoreAccessModeLiteral = Literal["r", "w", "a"] +ZipStoreAccessModeLiteral = Literal["r", "w", "a", "x"] class _RawReaderAdapter(io.RawIOBase): @@ -72,6 +72,17 @@ def readinto(self, b: Any) -> int: return n +def _is_unfinished_archive(path: Path) -> bool: + """Whether the file at `path` exists but is not a zip archive.""" + try: + with path.open("rb") as f: + # is_zipfile swallows OSError, so open the file here to let a + # directory or a permission error surface as itself + return not zipfile.is_zipfile(f) + except FileNotFoundError: + return False + + class ZipStore(Store): """ Store using a ZIP file. @@ -85,10 +96,20 @@ class ZipStore(Store): can only be used for reading (`mode="r"`). The file object must stay open for the lifetime of the store, and operations that require a filesystem location (`clear`, `move`, pickling) are not supported. + Using the store again after `close()` reopens the archive, which + requires a file object that is readable and seekable; otherwise it + raises `io.UnsupportedOperation`. mode : str, optional One of 'r' to read an existing file, 'w' to truncate and write a new file, 'a' to append to an existing file, or 'x' to exclusively create - and write a new file. + and write a new file. 'w' and 'x' apply to the first open only; the + store reopens its archive with 'a' after `close()`, `move()`, or + unpickling, so the entries it already wrote are kept. An unpickled + store opens its archive on first use. If `close()` raises, the + archive may be incomplete, and every later use that would reopen it + raises `RuntimeError` instead; for a writable store backed + by a path, `clear()` replaces the archive and makes the store usable + again. compression : int, optional Compression method to use when writing to the archive. allowZip64 : bool, optional @@ -162,12 +183,48 @@ def __init__( self._zmode = mode self.compression = compression self.allowZip64 = allowZip64 + self._lock = threading.RLock() + self._was_opened = False + self._close_failed = False def _sync_open(self) -> None: if self._is_open: raise ValueError("store is already open") - - self._lock = threading.RLock() + if self._close_failed: + # the central directory may be missing; appending to such a file + # makes zipfile start a new archive and drop the earlier entries + raise RuntimeError( + f"closing the archive of {self!r} failed, so it may be incomplete; " + "the store will not reopen it" + ) + if ( + self.path is not None + and self._was_opened + and self._zmode == "a" + and _is_unfinished_archive(self.path) + ): + # the file lacks a central directory, e.g. because another copy of + # this store still has it open for writing; appending would start + # a new archive and drop the entries already written + raise zipfile.BadZipFile( + f"{self!r} cannot reopen {self.path}: it is not a zip archive. It " + "may be incomplete, for example because another store still has it " + "open for writing." + ) + if ( + self.path is None + and self._fileobj is not None + and self._was_opened + and not (self._fileobj.readable() and self._fileobj.seekable()) + ): + # reopening appends, which needs to read the archive back; on a + # write-only file zipfile would start a new archive and drop the + # earlier entries + raise io.UnsupportedOperation( + "a ZipStore backed by a file object that is not readable and " + "seekable cannot be used again after close(), because the " + "archive it wrote cannot be read back" + ) self._zf = zipfile.ZipFile( self.path if self.path is not None else self._fileobj, # type: ignore[arg-type] @@ -175,11 +232,29 @@ def _sync_open(self) -> None: compression=self.compression, allowZip64=self.allowZip64, ) + # "w" truncates and "x" refuses an existing file. Both apply only to the + # first open: reopening after close(), move(), or unpickling must keep + # the entries already written. + if self._zmode in ("w", "x"): + self._zmode = "a" + self._was_opened = True self._is_open = True + def _zipfile(self) -> zipfile.ZipFile: + """Return the archive, opening it on first use.""" + with self._lock: + if not self._is_open: + self._sync_open() + return self._zf + async def _open(self) -> None: - self._sync_open() + with self._lock: + self._sync_open() + + async def _ensure_open(self) -> None: + # the base class checks _is_open outside the lock + self._zipfile() def __getstate__(self) -> dict[str, Any]: if self.path is None: @@ -189,24 +264,39 @@ def __getstate__(self) -> dict[str, Any]: "cannot pickle a ZipStore backed by a file-like object; " "construct the store from a path instead" ) - # We need a copy to not modify the state of the original store - state = self.__dict__.copy() + # We need a copy to not modify the state of the original store. The + # lock keeps it from catching _sync_open between opening with "w" and + # switching to "a", which would make unpickling truncate the archive. + with self._lock: + state = self.__dict__.copy() for attr in ["_zf", "_lock"]: state.pop(attr, None) return state def __setstate__(self, state: dict[str, Any]) -> None: self.__dict__ = state + self.__dict__.setdefault("_close_failed", False) + self.__dict__.setdefault("_was_opened", False) + self._lock = threading.RLock() + # open on first use: a copy that is never used, like those dask makes + # while building a graph, must not open the file, because zipfile + # writes a central directory when it closes an archive it could not read self._is_open = False - self._sync_open() def close(self) -> None: # docstring inherited - if not self._is_open: - return - super().close() + # hold the lock until the archive is closed: a thread that reopened it + # before its central directory was written would lose the entries with self._lock: - self._zf.close() + if not self._is_open: + return + try: + self._zf.close() + except BaseException: + self._close_failed = True + raise + finally: + super().close() async def clear(self) -> None: # docstring inherited @@ -216,11 +306,20 @@ async def clear(self) -> None: raise NotImplementedError( "clear() is not supported for a ZipStore backed by a file-like object" ) - self._zf.close() - os.remove(self.path) - self._zf = zipfile.ZipFile( - self.path, mode="w", compression=self.compression, allowZip64=self.allowZip64 - ) + if self._zmode == "x": + # "x" lasts only until the first open; opening now keeps it from + # deleting a file this store never claimed + self._zipfile() + self.close() + # the file may be damaged or gone; clear() replaces it either way + self.path.unlink(missing_ok=True) + # replacing the file cannot drop entries, so clear() is the one way + # to recover a store whose close() failed + self._close_failed = False + # if this open fails the store stays closed, and the next use + # creates the archive again + self._zmode = "w" + self._sync_open() def __str__(self) -> str: if self.path is None: @@ -243,11 +342,9 @@ def _get( prototype: BufferPrototype, byte_range: ByteRequest | None = None, ) -> Buffer | None: - if not self._is_open: - self._sync_open() # docstring inherited try: - with self._zf.open(key) as f: # will raise KeyError + with self._zipfile().open(key) as f: # will raise KeyError if byte_range is None: return prototype.buffer.from_bytes(f.read()) elif isinstance(byte_range, RangeByteRequest): @@ -288,8 +385,6 @@ async def get_partial_values( return out def _set(self, key: str, value: Buffer) -> None: - if not self._is_open: - self._sync_open() # generally, this should be called inside a lock keyinfo = zipfile.ZipInfo(filename=key, date_time=time.localtime(time.time())[:6]) keyinfo.compress_type = self.compression @@ -298,13 +393,11 @@ def _set(self, key: str, value: Buffer) -> None: keyinfo.external_attr |= 0x10 # MS-DOS directory flag else: keyinfo.external_attr = 0o644 << 16 # ?rw-r--r-- - self._zf.writestr(keyinfo, value.to_bytes()) + self._zipfile().writestr(keyinfo, value.to_bytes()) async def set(self, key: str, value: Buffer) -> None: # docstring inherited self._check_writable() - if not self._is_open: - self._sync_open() if not isinstance(value, Buffer): raise TypeError( f"ZipStore.set(): `value` must be a Buffer instance. Got an instance of {type(value)} instead." @@ -315,7 +408,7 @@ async def set(self, key: str, value: Buffer) -> None: async def set_if_not_exists(self, key: str, value: Buffer) -> None: self._check_writable() with self._lock: - members = self._zf.namelist() + members = self._zipfile().namelist() if key not in members: self._set(key, value) @@ -337,11 +430,9 @@ async def delete(self, key: str) -> None: async def exists(self, key: str) -> bool: # docstring inherited - if not self._is_open: - self._sync_open() with self._lock: try: - self._zf.getinfo(key) + self._zipfile().getinfo(key) except KeyError: return False else: @@ -349,11 +440,10 @@ async def exists(self, key: str) -> bool: async def list(self) -> AsyncIterator[str]: # docstring inherited - if not self._is_open: - self._sync_open() - with self._lock: - for key in self._zf.namelist(): - yield key + # namelist() is a copy; holding the lock across yield would block + # other threads for as long as the caller iterates + for key in self._zipfile().namelist(): + yield key async def list_prefix(self, prefix: str) -> AsyncIterator[str]: # docstring inherited @@ -363,11 +453,9 @@ async def list_prefix(self, prefix: str) -> AsyncIterator[str]: async def list_dir(self, prefix: str) -> AsyncIterator[str]: # docstring inherited - if not self._is_open: - self._sync_open() prefix = prefix.rstrip("/") - keys = self._zf.namelist() + keys = self._zipfile().namelist() seen = set() if prefix == "": keys_unique = {k.split("/")[0] for k in keys} @@ -393,8 +481,12 @@ async def move(self, path: Path | str) -> None: ) if isinstance(path, str): path = Path(path) - self.close() - os.makedirs(path.parent, exist_ok=True) - shutil.move(self.path, path) - self.path = path - await self._open() + # hold the lock so that no thread reopens the old path mid-move + with self._lock: + # opening first keeps mode "x" from moving a file it may not claim + self._zipfile() + self.close() + os.makedirs(path.parent, exist_ok=True) + shutil.move(self.path, path) + self.path = path + self._sync_open() diff --git a/tests/test_store/test_zip.py b/tests/test_store/test_zip.py index 8931875125..251a28515e 100644 --- a/tests/test_store/test_zip.py +++ b/tests/test_store/test_zip.py @@ -1,12 +1,16 @@ from __future__ import annotations +import gc import io import os +import pathlib import pickle import shutil import tempfile +import threading import zipfile -from typing import TYPE_CHECKING, Literal +from concurrent.futures import Future, ThreadPoolExecutor, wait +from typing import TYPE_CHECKING, Literal, cast import numpy as np import pytest @@ -27,9 +31,12 @@ from zarr.testing.store import StoreTests if TYPE_CHECKING: + from collections.abc import AsyncGenerator, AsyncIterator from pathlib import Path from typing import Any + from zarr.storage._zip import ZipStoreAccessModeLiteral + # TODO: work out where this is coming from and fix pytestmark = [ @@ -39,6 +46,10 @@ ] +async def _drain(keys: AsyncIterator[str]) -> list[str]: + return [k async for k in keys] + + class TestZipStore(StoreTests[ZipStore, cpu.Buffer]): store_cls = ZipStore buffer_cls = cpu.Buffer @@ -188,6 +199,479 @@ async def test_move(self, tmp_path: Path) -> None: assert not origin.exists() assert np.array_equal(array[...], np.arange(10)) + @pytest.mark.parametrize( + ("call", "result", "keys"), + [ + pytest.param( + lambda s: s.get("foo", default_buffer_prototype()), b"bar", ["foo"], id="get" + ), + pytest.param( + lambda s: s.get_partial_values(default_buffer_prototype(), [("foo", None)]), + [b"bar"], + ["foo"], + id="get_partial_values", + ), + pytest.param(lambda s: s.exists("foo"), True, ["foo"], id="exists"), + pytest.param( + lambda s: s.set("bar", cpu.Buffer.from_bytes(b"x")), + None, + ["foo", "bar"], + id="set", + ), + pytest.param( + lambda s: s.set_if_not_exists("bar", cpu.Buffer.from_bytes(b"x")), + None, + ["foo", "bar"], + id="set_if_not_exists", + ), + pytest.param( + lambda s: s.set_if_not_exists("foo", cpu.Buffer.from_bytes(b"x")), + None, + ["foo"], + id="set_if_not_exists_existing", + ), + pytest.param(lambda s: s.clear(), None, [], id="clear"), + pytest.param(lambda s: s.delete("bar"), None, ["foo"], id="delete"), + pytest.param(lambda s: s.delete_dir("bar"), None, ["foo"], id="delete_dir"), + pytest.param(lambda s: s.is_empty(""), False, ["foo"], id="is_empty"), + ], + ) + async def test_methods_open_store_on_first_use( + self, store_kwargs: dict[str, Any], call: Any, result: Any, keys: list[str] + ) -> None: + """ + Every method works on a store that was constructed but never opened, + and sees the entries already in the archive. + """ + seed = await self.store_cls.open(**store_kwargs) + await seed.set("foo", cpu.Buffer.from_bytes(b"bar")) + seed.close() + + store = self.store_cls(**{**store_kwargs, "mode": "a"}) + assert not store._is_open + out = await call(store) + if isinstance(out, list): + out = [b.to_bytes() for b in out] + elif isinstance(out, Buffer): + out = out.to_bytes() + assert out == result + store.close() + with zipfile.ZipFile(store_kwargs["path"]) as zf: + assert zf.namelist() == keys + + @pytest.mark.parametrize("method", ["list", "list_dir"]) + async def test_listing_opens_store_on_first_use( + self, store_kwargs: dict[str, Any], method: str + ) -> None: + """ + Listing a store that was constructed but never opened opens the + archive and yields the entries already in it. + """ + seed = await self.store_cls.open(**store_kwargs) + await seed.set("foo", cpu.Buffer.from_bytes(b"bar")) + seed.close() + + store = self.store_cls(**{**store_kwargs, "mode": "r"}) + args = () if method == "list" else ("",) + assert [k async for k in getattr(store, method)(*args)] == ["foo"] + store.close() + + @pytest.mark.parametrize("mode", ["w", "x"]) + @pytest.mark.parametrize("reopen", ["use_after_close", "move", "pickle"]) + async def test_reopen_keeps_entries( + self, tmp_path: Path, mode: ZipStoreAccessModeLiteral, reopen: str + ) -> None: + """ + "w" truncates and "x" refuses an existing file; neither may apply + when a store that already wrote entries is opened again. + """ + store = ZipStore(tmp_path / "data.zip", mode=mode) + await store.set("foo", cpu.Buffer.from_bytes(b"bar")) + if reopen == "use_after_close": + store.close() + value = await store.get("foo", default_buffer_prototype()) + assert value is not None + assert value.to_bytes() == b"bar" + store.close() + elif reopen == "move": + await store.move(tmp_path / "moved" / "data.zip") + store.close() + else: + store.close() + copy = pickle.loads(pickle.dumps(store)) + value = await copy.get("foo", default_buffer_prototype()) + assert value is not None + assert value.to_bytes() == b"bar" + copy.close() + with zipfile.ZipFile(store.path) as zf: # type: ignore[arg-type] + assert zf.namelist() == ["foo"] + + async def test_close_blocks_concurrent_reopen( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """ + A thread that uses the store while close() is writing the central + directory must wait until the archive is closed. + """ + store = ZipStore(tmp_path / "data.zip", mode="w") + await store.set("foo", cpu.Buffer.from_bytes(b"1")) + write_end_record = store._zf._write_end_record # type: ignore[attr-defined] + writes: list[Future[None]] = [] + + with ThreadPoolExecutor(max_workers=1) as pool: + + def write_during_close() -> None: + writes.append(pool.submit(sync, store.set("bar", cpu.Buffer.from_bytes(b"2")))) + wait(writes, timeout=0.2) + write_end_record() + + monkeypatch.setattr(store._zf, "_write_end_record", write_during_close) + store.close() + writes[0].result(timeout=5) + store.close() + + with zipfile.ZipFile(store.path) as zf: # type: ignore[arg-type] + assert zf.namelist() == ["foo", "bar"] + + async def test_move_blocks_concurrent_reopen( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """ + A thread that uses the store while move() runs must wait for the + move, or it reopens the old path between close and reopen. + """ + origin = tmp_path / "data.zip" + destination = tmp_path / "moved" / "data.zip" + store = ZipStore(origin, mode="w") + await store.set("foo", cpu.Buffer.from_bytes(b"1")) + + writer = threading.Thread( + target=lambda: sync(store.set("bar", cpu.Buffer.from_bytes(b"2"))) + ) + move_file = shutil.move + + def write_then_move(src: Any, dst: Any) -> Any: + # the store is closed here; start a write before move() reopens it + writer.start() + writer.join(timeout=0.2) + return move_file(src, dst) + + monkeypatch.setattr(shutil, "move", write_then_move) + await store.move(destination) + monkeypatch.undo() + writer.join(timeout=5) + assert not writer.is_alive() + store.close() + + assert not origin.exists() + with zipfile.ZipFile(destination) as zf: + assert zf.namelist() == ["foo", "bar"] + + async def test_failed_clear_leaves_store_closed( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """ + If clear() fails after closing the archive, the store is closed + rather than left open on a closed handle, and the next use clears. + """ + store = ZipStore(tmp_path / "data.zip", mode="w") + await store.set("foo", cpu.Buffer.from_bytes(b"1")) + + def unavailable(*args: Any, **kwargs: Any) -> None: + raise OSError("unavailable") + + with monkeypatch.context() as m: + m.setattr(zipfile, "ZipFile", unavailable) + with pytest.raises(OSError, match="unavailable"): + await store.clear() + assert not store._is_open + + await store.set("bar", cpu.Buffer.from_bytes(b"2")) + store.close() + with zipfile.ZipFile(store.path) as zf: # type: ignore[arg-type] + assert zf.namelist() == ["bar"] + + async def test_clear_exclusive_mode_keeps_existing_file(self, tmp_path: Path) -> None: + """ + A never-opened "x" store has not claimed the file, so clear() must + refuse it the way the first open would instead of deleting it. + """ + path = tmp_path / "data.zip" + with zipfile.ZipFile(path, mode="w") as zf: + zf.writestr("foo", b"1") + + store = ZipStore(path, mode="x") + with pytest.raises(FileExistsError): + await store.clear() + with zipfile.ZipFile(path) as zf: + assert zf.namelist() == ["foo"] + + @pytest.mark.parametrize( + "use", + [ + pytest.param(lambda s: s._ensure_open(), id="ensure_open"), + pytest.param(lambda s: _drain(s.list_dir("")), id="list_dir"), + ], + ) + async def test_first_use_waits_for_lock(self, tmp_path: Path, use: Any) -> None: + """ + A thread's first use must wait while another thread holds the lock, + then find the archive that thread opened instead of opening it again. + """ + store = ZipStore(tmp_path / "data.zip", mode="w") + with ThreadPoolExecutor(max_workers=1) as pool: + with store._lock: + first_use = pool.submit(sync, use(store)) + wait([first_use], timeout=0.2) + assert not store._is_open + store._zipfile() + first_use.result(timeout=5) + store.close() + + @pytest.mark.parametrize("error", [OSError, KeyboardInterrupt]) + async def test_failed_close_blocks_reopen( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, error: type[BaseException] + ) -> None: + """ + If the central directory could not be written, reopening in append + mode would start a new archive and drop the entries, so later use + raises instead. + """ + store = ZipStore(tmp_path / "data.zip", mode="w") + await store.set("foo", cpu.Buffer.from_bytes(b"1")) + + def disk_full() -> None: + raise error("disk full") + + monkeypatch.setattr(store._zf, "_write_end_record", disk_full) + with pytest.raises(error, match="disk full"): + store.close() + assert not store._is_open + with pytest.raises(RuntimeError, match="closing the archive"): + await store.set("bar", cpu.Buffer.from_bytes(b"2")) + + # clear() replaces the archive, so it cannot drop entries, and it + # makes the store usable again + await store.clear() + await store.set("baz", cpu.Buffer.from_bytes(b"3")) + store.close() + with zipfile.ZipFile(store.path) as zf: # type: ignore[arg-type] + assert zf.namelist() == ["baz"] + + async def test_unused_copy_of_open_writer_does_not_write(self, tmp_path: Path) -> None: + """ + A copy of a store that is still writing, like those dask makes while + building a graph, leaves the file alone if it is never used, even + when it is collected after the original closes. + """ + store = ZipStore(tmp_path / "data.zip", mode="w") + await store.set("foo", cpu.Buffer.from_bytes(b"1")) + copy = pickle.loads(pickle.dumps(store)) + await store.set("bar", cpu.Buffer.from_bytes(b"2")) + store.close() + del copy + gc.collect() + + with zipfile.ZipFile(store.path) as zf: # type: ignore[arg-type] + assert zf.namelist() == ["foo", "bar"] + assert zf.testzip() is None + + async def test_used_copy_of_open_writer_raises(self, tmp_path: Path) -> None: + """ + The file has no central directory until the original closes, so a + copy that uses it refuses to reopen it instead of starting a new + archive after the entries already written. + """ + store = ZipStore(tmp_path / "data.zip", mode="w") + await store.set("foo", cpu.Buffer.from_bytes(b"1")) + copy = pickle.loads(pickle.dumps(store)) + with pytest.raises(zipfile.BadZipFile, match="not a zip archive"): + await copy.get("foo", default_buffer_prototype()) + store.close() + + with zipfile.ZipFile(store.path) as zf: # type: ignore[arg-type] + assert zf.namelist() == ["foo"] + + async def test_failed_clear_keeps_failed_close_guard( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """ + Clear() lifts the failed-close guard only once the old file is gone. + """ + store = ZipStore(tmp_path / "data.zip", mode="w") + await store.set("foo", cpu.Buffer.from_bytes(b"1")) + + def disk_full() -> None: + raise OSError("disk full") + + def locked(path: Any, missing_ok: bool = False) -> None: + raise PermissionError("locked") + + monkeypatch.setattr(store._zf, "_write_end_record", disk_full) + with pytest.raises(OSError, match="disk full"): + store.close() + with monkeypatch.context() as m: + m.setattr(pathlib.Path, "unlink", locked) + with pytest.raises(PermissionError, match="locked"): + await store.clear() + with pytest.raises(RuntimeError, match="closing the archive"): + await store.set("bar", cpu.Buffer.from_bytes(b"2")) + + @pytest.mark.parametrize("damage", ["garbage", "missing"]) + @pytest.mark.parametrize("close", ["closed", "close_failed"]) + async def test_clear_replaces_damaged_file( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, damage: str, close: str + ) -> None: + """ + Clear() replaces the archive even when the file it wrote is no longer + a zip or is gone, including after a failed close(). + """ + path = tmp_path / "data.zip" + store = ZipStore(path, mode="w") + await store.set("foo", cpu.Buffer.from_bytes(b"1")) + if close == "closed": + store.close() + else: + + def disk_full() -> None: + raise OSError("disk full") + + monkeypatch.setattr(store._zf, "_write_end_record", disk_full) + with pytest.raises(OSError, match="disk full"): + store.close() + if damage == "garbage": + path.write_bytes(b"not a zip") + else: + path.unlink() + + await store.clear() + await store.set("bar", cpu.Buffer.from_bytes(b"2")) + store.close() + with zipfile.ZipFile(path) as zf: + assert zf.namelist() == ["bar"] + + async def test_reopen_recreates_deleted_file(self, tmp_path: Path) -> None: + """ + A file removed between uses is created again, not refused. + """ + path = tmp_path / "data.zip" + store = ZipStore(path, mode="w") + await store.set("foo", cpu.Buffer.from_bytes(b"1")) + store.close() + path.unlink() + + await store.set("bar", cpu.Buffer.from_bytes(b"2")) + store.close() + with zipfile.ZipFile(path) as zf: + assert zf.namelist() == ["bar"] + + async def test_first_open_append_on_empty_file(self, tmp_path: Path) -> None: + """ + Mode "a" on an empty file, e.g. from tempfile.mkstemp, starts an + archive in it; the reopen guard applies only to archives this store wrote. + """ + path = tmp_path / "data.zip" + path.touch() + store = ZipStore(path, mode="a") + await store.set("foo", cpu.Buffer.from_bytes(b"1")) + store.close() + with zipfile.ZipFile(path) as zf: + assert zf.namelist() == ["foo"] + + @pytest.mark.parametrize("file", ["archive", "missing"]) + async def test_clear_read_mode_writable_store(self, tmp_path: Path, file: str) -> None: + """ + A store opened with mode "r" but read_only=False creates the new + archive with "w", since "r" cannot open the file clear() removed, + and it does not need to open the old file first. + """ + path = tmp_path / "data.zip" + if file == "archive": + with zipfile.ZipFile(path, mode="w") as zf: + zf.writestr("foo", b"1") + + store = ZipStore(path, mode="r", read_only=False) + await store.clear() + await store.set("bar", cpu.Buffer.from_bytes(b"2")) + store.close() + with zipfile.ZipFile(path) as zf: + assert zf.namelist() == ["bar"] + + async def test_reopen_reports_directory_at_path(self, tmp_path: Path) -> None: + """ + A directory where the archive was is reported as an OS error, not as + an unfinished archive. + """ + path = tmp_path / "data.zip" + store = ZipStore(path, mode="w") + await store.set("foo", cpu.Buffer.from_bytes(b"1")) + store.close() + path.unlink() + path.mkdir() + + with pytest.raises(OSError): + await store.get("foo", default_buffer_prototype()) + + @pytest.mark.parametrize("mode", ["r", "a"]) + async def test_unpickle_state_from_older_release(self, tmp_path: Path, mode: str) -> None: + """ + A pickle made before _was_opened and _close_failed existed opens the + file as a first open would; for "a" on an empty file that starts an + archive rather than tripping the reopen guard. + """ + path = tmp_path / "data.zip" + if mode == "r": + with zipfile.ZipFile(path, mode="w") as zf: + zf.writestr("foo", b"1") + else: + path.touch() + state = ZipStore(path, mode=mode).__getstate__() # type: ignore[arg-type] + del state["_was_opened"], state["_close_failed"] + + store = ZipStore.__new__(ZipStore) + store.__setstate__(state) + if mode == "a": + await store.set("foo", cpu.Buffer.from_bytes(b"1")) + assert [k async for k in store.list()] == ["foo"] + store.close() + + async def test_list_does_not_hold_lock_while_iterating(self, tmp_path: Path) -> None: + """ + Another thread can use the store while a caller is partway + through iterating list(). + """ + store = ZipStore(tmp_path / "data.zip", mode="w") + await store.set("foo", cpu.Buffer.from_bytes(b"1")) + await store.set("bar", cpu.Buffer.from_bytes(b"2")) + # list() is an async generator; the annotation hides aclose() + keys = cast("AsyncGenerator[str, None]", store.list()) + pool = ThreadPoolExecutor(max_workers=1) + try: + assert await anext(keys) == "foo" + other = pool.submit(sync, _drain(store.list_dir(""))) + assert sorted(other.result(timeout=5)) == ["bar", "foo"] + finally: + await keys.aclose() + pool.shutdown() + store.close() + + async def test_move_exclusive_mode_keeps_existing_file(self, tmp_path: Path) -> None: + """ + A never-opened "x" store has not claimed the file, so move() must + refuse it the way the first open would instead of moving it. + """ + origin = tmp_path / "data.zip" + destination = tmp_path / "moved" / "data.zip" + with zipfile.ZipFile(origin, mode="w") as zf: + zf.writestr("foo", b"1") + + store = ZipStore(origin, mode="x") + with pytest.raises(FileExistsError): + await store.move(destination) + assert not destination.exists() + with zipfile.ZipFile(origin) as zf: + assert zf.namelist() == ["foo"] + class TestZipStoreFileObj: """ZipStore backed by an open binary file-like object instead of a path.""" @@ -219,6 +703,57 @@ def test_write_to_fileobj(self) -> None: array = zarr.open_array(roundtrip, mode="r") assert np.array_equal(array[...], np.arange(4)) + async def test_write_after_close_keeps_entries(self) -> None: + """ + Reopening a file object after close() appends to the archive it + already holds instead of starting a new one after it. + """ + buffer = io.BytesIO() + store = ZipStore(buffer, mode="w", read_only=False) + await store.set("foo", cpu.Buffer.from_bytes(b"1")) + store.close() + await store.set("bar", cpu.Buffer.from_bytes(b"2")) + store.close() + + with zipfile.ZipFile(io.BytesIO(buffer.getvalue())) as zf: + assert zf.namelist() == ["foo", "bar"] + + async def test_write_only_reuse_after_close_raises(self, tmp_path: Path) -> None: + """ + A write-only file object cannot be read back, so reopening it would + start a new archive and drop the entries already written. + """ + path = tmp_path / "data.zip" + with path.open("wb") as f: + store = ZipStore(f, mode="w", read_only=False) + await store.set("foo", cpu.Buffer.from_bytes(b"1")) + store.close() + try: + with pytest.raises(io.UnsupportedOperation, match="readable and seekable"): + await store.set("bar", cpu.Buffer.from_bytes(b"2")) + finally: + # close before the file does, even if the write went through + store.close() + + with zipfile.ZipFile(path) as zf: + assert zf.namelist() == ["foo"] + + async def test_unseekable_reuse_after_close_raises(self) -> None: + """ + A file object that cannot seek cannot be read back either, so reuse + after close() raises the same error as a write-only one. + """ + buffer = io.BytesIO() + pipe = io.BufferedRWPair(buffer, buffer) # readable, not seekable + store = ZipStore(pipe, mode="w", read_only=False) # type: ignore[arg-type] + await store.set("foo", cpu.Buffer.from_bytes(b"1")) + store.close() + try: + with pytest.raises(io.UnsupportedOperation, match="readable and seekable"): + await store.set("bar", cpu.Buffer.from_bytes(b"2")) + finally: + store.close() + async def test_clear_unsupported(self, zip_bytes: bytes) -> None: # clear() requires a filesystem location, so it raises a clear error # for file-object-backed stores @@ -352,8 +887,7 @@ class ZipStoreLifecycleMachine(RuleBasedStateMachine): Invariant under test: a constructed ZipStore can always be closed without raising, regardless of whether it was ever opened or did any I/O. This is a property-based generalization of the former example-based regression tests - for ZipStore.close() being called on a never-opened store (which raised - AttributeError because ``_lock`` is created lazily in ``_sync_open``). + for ZipStore.close() being called on a never-opened store. """ def __init__(self, tmp_path: Path) -> None: