From 361cd0be52b8fb9da1394b5737a7485615b4bdf8 Mon Sep 17 00:00:00 2001 From: Fabian Isensee Date: Fri, 6 Nov 2026 09:22:35 +0100 Subject: [PATCH 1/3] Bound decompress2()/decompress() by the sizes of dst and src decompress2(src, dst) passed the chunk's nbytes, not the length of dst, as destsize to blosc2_decompress_ctx(), so a dst smaller than the decompressed data was overrun: heap corruption or a segfault. decompress() used blosc1_decompress(), which passes INT32_MAX as srcsize, so a truncated src was read past its end. Both functions now pass the length of dst in bytes as destsize and raise a RuntimeError naming both sizes when dst is too small, as their docstrings say, and decompress() calls blosc2_decompress() with the length of src. Both lengths are clamped to INT_MAX, the most C-Blosc2 takes, so a dst or src larger than 2 GiB (a chunk at the start of a big buffer) works in both functions; before, decompress() failed with such a dst and decompress2() raised OverflowError with such a src. dst's buffer is released in a finally block, so a failing call (an empty dst included) no longer keeps it exported. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01DiPFzsfE5i74cBSxZN13M8 --- src/blosc2/blosc2_ext.pyx | 37 ++++++--- src/blosc2/core.py | 2 +- tests/test_decompress_buffer_sizes.py | 114 ++++++++++++++++++++++++++ 3 files changed, 140 insertions(+), 13 deletions(-) create mode 100644 tests/test_decompress_buffer_sizes.py diff --git a/src/blosc2/blosc2_ext.pyx b/src/blosc2/blosc2_ext.pyx index de90e254c..9f442fe55 100644 --- a/src/blosc2/blosc2_ext.pyx +++ b/src/blosc2/blosc2_ext.pyx @@ -3364,6 +3364,18 @@ cpdef compress(src, int32_t typesize=8, int clevel=9, filter=blosc2.Filter.SHUFF raise ValueError("Cannot compress") +cdef int32_t _checked_dst_size(Py_ssize_t dst_len, int32_t nbytes) except -1: + """The destsize for decompressing into a dst of `dst_len` bytes: its length, up to the INT_MAX that + C-Blosc2 takes. C-Blosc2 refuses a chunk larger than destsize; this raises the documented errors first, + for an empty dst or one smaller than `nbytes` (the header's size of the decompressed data).""" + if dst_len == 0: + raise ValueError("The dst length must be greater than 0") + if dst_len < nbytes: + raise RuntimeError(f"The dst length ({dst_len} bytes) is smaller than the decompressed size of src " + f"({nbytes} bytes)") + return min(dst_len, INT_MAX) + + def decompress(src, dst=None, as_bytearray=False): cdef int32_t nbytes cdef int32_t cbytes @@ -3374,18 +3386,20 @@ def decompress(src, dst=None, as_bytearray=False): typed_view_src = mem_view_src.cast('B') _check_comp_length('src', len(typed_view_src)) blosc2_cbuffer_sizes(&typed_view_src[0], &nbytes, &cbytes, &blocksize) + cdef int32_t srcsize = min(len(typed_view_src), INT_MAX) cdef Py_buffer buf if dst is not None: PyObject_GetBuffer(dst, &buf, PyBUF_SIMPLE) - if buf.len == 0: - raise ValueError("The dst length must be greater than 0") - size = blosc1_decompress(&typed_view_src[0], buf.buf, buf.len) - PyBuffer_Release(&buf) + try: + size = blosc2_decompress(&typed_view_src[0], srcsize, buf.buf, + _checked_dst_size(buf.len, nbytes)) + finally: + PyBuffer_Release(&buf) else: dst = PyBytes_FromStringAndSize(NULL, nbytes) if dst is None: raise RuntimeError("Could not get a bytes object") - size = blosc1_decompress(&typed_view_src[0], dst, len(dst)) + size = blosc2_decompress(&typed_view_src[0], srcsize, dst, nbytes) if size >= 0: return bytearray(dst) if as_bytearray else dst if size < 0: @@ -3720,20 +3734,19 @@ def decompress2(src, dst=None, **kwargs): cdef int32_t nbytes cdef int32_t cbytes cdef int32_t blocksize - cdef int32_t srcsize = typed_view_src.nbytes + cdef int32_t srcsize = min(len(typed_view_src), INT_MAX) blosc2_cbuffer_sizes(&typed_view_src[0], &nbytes, &cbytes, &blocksize) cdef Py_buffer buf if dst is not None: PyObject_GetBuffer(dst, &buf, PyBUF_SIMPLE) - if buf.len == 0: - blosc2_free_ctx(dctx) - raise ValueError("The dst length must be greater than 0") view = &typed_view_src[0] # For lazy chunks, blosc2_cbuffer_sizes() only reports the header cbytes. # The decode context needs the full source buffer length. - size = _decompress_ctx(dctx, view, srcsize, buf.buf, nbytes) - blosc2_free_ctx(dctx) - PyBuffer_Release(&buf) + try: + size = _decompress_ctx(dctx, view, srcsize, buf.buf, _checked_dst_size(buf.len, nbytes)) + finally: + blosc2_free_ctx(dctx) + PyBuffer_Release(&buf) else: dst = PyBytes_FromStringAndSize(NULL, nbytes) if dst is None: diff --git a/src/blosc2/core.py b/src/blosc2/core.py index dd017a2ef..84592ac7e 100644 --- a/src/blosc2/core.py +++ b/src/blosc2/core.py @@ -2273,7 +2273,7 @@ def decompress2(src: object, dst: object | bytearray = None, **kwargs: dict) -> Raises ------ RuntimeError - If the data cannot be compressed into :paramref:`dst`. + If the data cannot be decompressed into :paramref:`dst`. If an internal error occurs, likely due to an invalid parameter If :paramref:`dst` is `None` and a bytes object could not be created to store the result. TypeError diff --git a/tests/test_decompress_buffer_sizes.py b/tests/test_decompress_buffer_sizes.py new file mode 100644 index 000000000..ce816c2df --- /dev/null +++ b/tests/test_decompress_buffer_sizes.py @@ -0,0 +1,114 @@ +####################################################################### +# Copyright (c) 2019-present, Blosc Development Team +# All rights reserved. +# +# SPDX-License-Identifier: BSD-3-Clause +####################################################################### + +"""decompress() and decompress2() stay within the src and dst buffers they are given.""" + +import array +import sys + +import numpy as np +import pytest + +import blosc2 + +NITEMS = 1 << 14 +DATA = np.arange(NITEMS, dtype=np.int64) +NBYTES = DATA.nbytes +CHUNK = blosc2.compress2(DATA, codec=blosc2.Codec.LZ4, nthreads=1) + +DECOMPRESSORS = [blosc2.decompress, blosc2.decompress2] + + +def _corrupt_src_error(decompress): + return RuntimeError if decompress is blosc2.decompress else ValueError + + +@pytest.mark.parametrize("decompress", DECOMPRESSORS) +@pytest.mark.parametrize( + "make_dst", + [ + pytest.param(bytearray, id="bytearray"), + pytest.param(lambda n: np.zeros(n // 8, dtype=np.int64), id="numpy"), + pytest.param(lambda n: np.zeros((2, n // 16), dtype=np.int64), id="numpy-2d"), + pytest.param(lambda n: np.zeros(n // 2, dtype=np.int16), id="numpy-int16"), + pytest.param(lambda n: memoryview(np.zeros(n // 8, dtype=np.int64)), id="memoryview"), + pytest.param(lambda n: array.array("q", bytes(n)), id="array"), + # A part of a larger array, as when decompressing chunks one after another into their place + pytest.param( + lambda n: np.zeros(n // 4, dtype=np.int64)[n // 16 : n // 16 + n // 8], id="numpy-slice" + ), + ], +) +def test_dst_kinds(decompress, make_dst): + exact = make_dst(NBYTES) + assert decompress(CHUNK, exact) is None + assert bytes(memoryview(exact).cast("B")) == DATA.tobytes() + + larger = make_dst(2 * NBYTES) + assert decompress(CHUNK, larger) is None + larger_bytes = memoryview(larger).cast("B") + assert larger_bytes[:NBYTES] == DATA.tobytes() + assert larger_bytes[NBYTES:] == bytes(NBYTES) + + +@pytest.mark.parametrize("decompress", DECOMPRESSORS) +@pytest.mark.parametrize("missing", [1, 8, NBYTES - 1]) +def test_dst_too_small(decompress, missing): + # dst is the start of a larger buffer: writing past its end shows in the rest + whole = bytearray(NBYTES + 64) + dst = memoryview(whole)[: NBYTES - missing] + with pytest.raises(RuntimeError, match=rf"dst length \({NBYTES - missing} bytes\) is smaller"): + decompress(CHUNK, dst) + assert whole == bytes(len(whole)) + + +@pytest.mark.parametrize("decompress", DECOMPRESSORS) +def test_empty_dst_is_released(decompress): + dst = bytearray() + with pytest.raises(ValueError, match="greater than 0"): + decompress(CHUNK, dst) + dst.extend(b"x") # BufferError while the buffer of dst is still held + + +@pytest.mark.skipif(sys.maxsize < 2**32, reason="needs a 64-bit address space") +@pytest.mark.parametrize("decompress", DECOMPRESSORS) +def test_dst_larger_than_2gib(decompress): + # np.zeros() gets zeroed pages from the OS: only the ones written are touched + dst = np.zeros(2**31 + 8, dtype=np.uint8) + decompress(CHUNK, dst) + assert np.array_equal(dst[:NBYTES].view(np.int64), DATA) + + +@pytest.mark.skipif(sys.maxsize < 2**32, reason="needs a 64-bit address space") +@pytest.mark.parametrize("decompress", DECOMPRESSORS) +def test_src_larger_than_2gib(decompress): + # A chunk at the start of a larger buffer (say, a memory-mapped file): only its own bytes are read + src = np.zeros(2**31 + 8, dtype=np.uint8) + src[: len(CHUNK)] = np.frombuffer(CHUNK, dtype=np.uint8) + assert decompress(src) == DATA.tobytes() + + +@pytest.mark.parametrize("decompress", DECOMPRESSORS) +@pytest.mark.parametrize("missing", [1, 100, len(CHUNK) // 2]) +def test_truncated_src(decompress, missing): + # The bytes past the end of src are the rest of the chunk: reading them would succeed + src = memoryview(bytearray(CHUNK))[: len(CHUNK) - missing] + with pytest.raises(_corrupt_src_error(decompress)): + decompress(src) + with pytest.raises(_corrupt_src_error(decompress)): + decompress(src, np.zeros(NITEMS, dtype=np.int64)) + + +@pytest.mark.parametrize("decompress", DECOMPRESSORS) +def test_truncated_memcpyed_src(decompress): + # Incompressible data: the chunk holds it as is after the header + data = np.random.default_rng(0).integers(0, 256, 1 << 16, dtype=np.uint8) + chunk = blosc2.compress2(data, typesize=1, clevel=0, nthreads=1) + assert len(chunk) == data.nbytes + blosc2.MAX_OVERHEAD + src = memoryview(bytearray(chunk))[: len(chunk) // 2] + with pytest.raises(_corrupt_src_error(decompress)): + decompress(src) From 51f73c7c8e59847b05f2a31984bf9580ff0da4ee Mon Sep 17 00:00:00 2001 From: Fabian Isensee Date: Fri, 6 Nov 2026 09:22:35 +0100 Subject: [PATCH 2/3] Free the dctx when vldecompress_block()/decompress2() get bad args vldecompress_block() created its decompression context before checking src, and decompress2() got the buffer of dst outside the try block that frees its context, so a src that is too short or not a buffer (vldecompress_block()), or a dst that is not a buffer or not contiguous (decompress2()), leaked the context: 1360 bytes per call. Create the context once src is known to be usable, and get the buffer of dst inside the try block. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01DiPFzsfE5i74cBSxZN13M8 --- src/blosc2/blosc2_ext.pyx | 16 +++--- tests/test_decompress_buffer_sizes.py | 81 ++++++++++++++++++++++++++- 2 files changed, 89 insertions(+), 8 deletions(-) diff --git a/src/blosc2/blosc2_ext.pyx b/src/blosc2/blosc2_ext.pyx index 9f442fe55..414a9ad27 100644 --- a/src/blosc2/blosc2_ext.pyx +++ b/src/blosc2/blosc2_ext.pyx @@ -3738,15 +3738,17 @@ def decompress2(src, dst=None, **kwargs): blosc2_cbuffer_sizes(&typed_view_src[0], &nbytes, &cbytes, &blocksize) cdef Py_buffer buf if dst is not None: - PyObject_GetBuffer(dst, &buf, PyBUF_SIMPLE) view = &typed_view_src[0] # For lazy chunks, blosc2_cbuffer_sizes() only reports the header cbytes. # The decode context needs the full source buffer length. try: - size = _decompress_ctx(dctx, view, srcsize, buf.buf, _checked_dst_size(buf.len, nbytes)) + PyObject_GetBuffer(dst, &buf, PyBUF_SIMPLE) + try: + size = _decompress_ctx(dctx, view, srcsize, buf.buf, _checked_dst_size(buf.len, nbytes)) + finally: + PyBuffer_Release(&buf) finally: blosc2_free_ctx(dctx) - PyBuffer_Release(&buf) else: dst = PyBytes_FromStringAndSize(NULL, nbytes) if dst is None: @@ -3888,15 +3890,15 @@ def vldecompress_block(src, int32_t nblock, **kwargs): cdef blosc2_dparams dparams create_dparams_from_kwargs(&dparams, kwargs) - cdef blosc2_context *dctx = blosc2_create_dctx(dparams) - if dctx == NULL: - raise RuntimeError("Could not create decompression context") - cdef const uint8_t[:] typed_view_src mem_view_src = memoryview(src) typed_view_src = mem_view_src.cast('B') _check_comp_length('src', typed_view_src.nbytes) + cdef blosc2_context *dctx = blosc2_create_dctx(dparams) + if dctx == NULL: + raise RuntimeError("Could not create decompression context") + cdef uint8_t *dest = NULL cdef int32_t destsize = 0 cdef int32_t rc diff --git a/tests/test_decompress_buffer_sizes.py b/tests/test_decompress_buffer_sizes.py index ce816c2df..50801ea80 100644 --- a/tests/test_decompress_buffer_sizes.py +++ b/tests/test_decompress_buffer_sizes.py @@ -5,9 +5,12 @@ # SPDX-License-Identifier: BSD-3-Clause ####################################################################### -"""decompress() and decompress2() stay within the src and dst buffers they are given.""" +"""The decompression functions stay within the src and dst buffers they are given, and free what they +took when they fail.""" import array +import ctypes +import gc import sys import numpy as np @@ -112,3 +115,79 @@ def test_truncated_memcpyed_src(decompress): src = memoryview(bytearray(chunk))[: len(chunk) // 2] with pytest.raises(_corrupt_src_error(decompress)): decompress(src) + + +class _Mallinfo2(ctypes.Structure): + _fields_ = [ + (name, ctypes.c_size_t) + for name in ( + "arena", + "ordblks", + "smblks", + "hblks", + "hblkhd", + "usmblks", + "fsmblks", + "uordblks", + "fordblks", + "keepcost", + ) + ] + + +def _get_mallinfo2(): + if not sys.platform.startswith("linux"): + return None + try: + mallinfo2 = ctypes.CDLL(None).mallinfo2 # glibc >= 2.33 + except (OSError, AttributeError): + return None + mallinfo2.restype = _Mallinfo2 + return mallinfo2 + + +_mallinfo2 = _get_mallinfo2() + + +def _c_bytes_left_per_call(fn, ncalls=300, nrounds=3): + """Smallest growth of malloc's in-use bytes per call of `fn` over `nrounds` rounds.""" + for _ in range(50): # warm-up: caches and other one-time allocations + fn() + growth = [] + for _ in range(nrounds): + gc.collect() + info = _mallinfo2() + before = info.uordblks + info.hblkhd + for _ in range(ncalls): + fn() + gc.collect() + info = _mallinfo2() + growth.append((info.uordblks + info.hblkhd - before) / ncalls) + return min(growth) + + +@pytest.mark.skipif(_mallinfo2 is None, reason="needs glibc's mallinfo2()") +@pytest.mark.parametrize( + ("call", "exc"), + [ + pytest.param( + lambda: blosc2.blosc2_ext.vldecompress_block(b"abc", 0), ValueError, id="vlblock-short-src" + ), + pytest.param( + lambda: blosc2.blosc2_ext.vldecompress_block(3, 0), TypeError, id="vlblock-no-buffer-src" + ), + pytest.param(lambda: blosc2.decompress2(CHUNK, 3), TypeError, id="decompress2-no-buffer-dst"), + pytest.param( + lambda: blosc2.decompress2(CHUNK, np.zeros(64, dtype=np.int64)[::2]), + ValueError, + id="decompress2-strided-dst", + ), + ], +) +def test_failed_call_frees_dctx(call, exc): + def fail(): + with pytest.raises(exc): + call() + + # The decompression context (about 1.3 kB) was leaked + assert _c_bytes_left_per_call(fail) < 256 From 7d3bcf0f06bd450a24a3d623fbc1a0f13aff848a Mon Sep 17 00:00:00 2001 From: Fabian Isensee Date: Fri, 6 Nov 2026 13:11:25 +0100 Subject: [PATCH 3/3] Refuse read-only dst buffers when decompressing decompress(), decompress2() and SChunk.decompress_chunk() got dst's buffer with PyBUF_SIMPLE, which read-only exporters grant too, so they decompressed into bytes objects, read-only memoryviews and NumPy arrays with writeable=False. Writing into an immutable object corrupts every other user of it (equal bytes objects can be shared). Ask for a writable buffer instead: the exporter then raises (BufferError, or ValueError for a read-only NumPy array), as the docstrings now say. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01DiPFzsfE5i74cBSxZN13M8 --- src/blosc2/blosc2_ext.pyx | 7 +++--- src/blosc2/core.py | 17 +++++++++------ src/blosc2/schunk.py | 8 +++++-- tests/test_decompress_buffer_sizes.py | 31 +++++++++++++++++++++++++-- 4 files changed, 49 insertions(+), 14 deletions(-) diff --git a/src/blosc2/blosc2_ext.pyx b/src/blosc2/blosc2_ext.pyx index 414a9ad27..2d557a477 100644 --- a/src/blosc2/blosc2_ext.pyx +++ b/src/blosc2/blosc2_ext.pyx @@ -26,6 +26,7 @@ import cython from cpython cimport ( Py_buffer, PyBUF_SIMPLE, + PyBUF_WRITABLE, PyBuffer_Release, PyBytes_AsString, PyBytes_FromStringAndSize, @@ -3389,7 +3390,7 @@ def decompress(src, dst=None, as_bytearray=False): cdef int32_t srcsize = min(len(typed_view_src), INT_MAX) cdef Py_buffer buf if dst is not None: - PyObject_GetBuffer(dst, &buf, PyBUF_SIMPLE) + PyObject_GetBuffer(dst, &buf, PyBUF_SIMPLE | PyBUF_WRITABLE) try: size = blosc2_decompress(&typed_view_src[0], srcsize, buf.buf, _checked_dst_size(buf.len, nbytes)) @@ -3742,7 +3743,7 @@ def decompress2(src, dst=None, **kwargs): # For lazy chunks, blosc2_cbuffer_sizes() only reports the header cbytes. # The decode context needs the full source buffer length. try: - PyObject_GetBuffer(dst, &buf, PyBUF_SIMPLE) + PyObject_GetBuffer(dst, &buf, PyBUF_SIMPLE | PyBUF_WRITABLE) try: size = _decompress_ctx(dctx, view, srcsize, buf.buf, _checked_dst_size(buf.len, nbytes)) finally: @@ -4439,7 +4440,7 @@ cdef class SChunk: cdef int32_t blocksize cdef Py_buffer buf if dst is not None: - PyObject_GetBuffer(dst, &buf, PyBUF_SIMPLE) + PyObject_GetBuffer(dst, &buf, PyBUF_SIMPLE | PyBUF_WRITABLE) if buf.len == 0: raise ValueError("The dst length must be greater than 0") size = blosc2_schunk_decompress_chunk(self.schunk, nchunk, buf.buf, buf.len) diff --git a/src/blosc2/core.py b/src/blosc2/core.py index 84592ac7e..feead6e4e 100644 --- a/src/blosc2/core.py +++ b/src/blosc2/core.py @@ -169,7 +169,7 @@ def decompress( dst: NumPy object or bytearray The destination NumPy object or bytearray to fill, the length of which must be greater than 0. - The user must ensure it has enough capacity to hold + It must be writable and have enough capacity to hold the decompressed data. Default is None, meaning that a new `bytes` or `bytearray` object is created, filled and returned. @@ -188,6 +188,8 @@ def decompress( Raises ------ + BufferError + Raised if :paramref:`dst` is read-only, like a `bytes` object or a read-only memoryview. RuntimeError Raised if the compressed data is corrupted or the output buffer is not large enough. Also raised if a `bytes` object could not be obtained. @@ -195,7 +197,7 @@ def decompress( Raised if :paramref:`src` does not support the Buffer Protocol. ValueError Raised if the length of :paramref:`src` is smaller than the minimum required length. - Also raised if `dst` is not `None` and its length is 0. + Also raised if `dst` is not `None` and its length is 0, or if it is a read-only NumPy array. Examples -------- @@ -2247,10 +2249,9 @@ def decompress2(src: object, dst: object | bytearray = None, **kwargs: dict) -> bytearray, memoryview, or numpy.ndarray. dst: NumPy object or bytearray, optional The destination NumPy object or bytearray to fill. The length - must be greater than 0. The user must ensure - it has enough capacity for the decompressed - data. Default is `None`, meaning a new bytes object - is created, filled and returned. + must be greater than 0. It must be writable and have enough + capacity for the decompressed data. Default is `None`, + meaning a new bytes object is created, filled and returned. kwargs: dict, optional Decompression parameters. The default values are in :class:`blosc2.DParams`. @@ -2272,6 +2273,8 @@ def decompress2(src: object, dst: object | bytearray = None, **kwargs: dict) -> Raises ------ + BufferError + If :paramref:`dst` is read-only, like a `bytes` object or a read-only memoryview. RuntimeError If the data cannot be decompressed into :paramref:`dst`. If an internal error occurs, likely due to an invalid parameter @@ -2280,7 +2283,7 @@ def decompress2(src: object, dst: object | bytearray = None, **kwargs: dict) -> If :paramref:`src` does not support the Buffer Protocol. ValueError If the length of :paramref:`src` is smaller than the minimum. - If :paramref:`dst` is not None and its length is 0. + If :paramref:`dst` is not None and its length is 0, or if it is a read-only NumPy array. """ if kwargs is not None and "dparams" in kwargs: if len(kwargs) > 1: diff --git a/src/blosc2/schunk.py b/src/blosc2/schunk.py index 33452acae..a003ab70b 100644 --- a/src/blosc2/schunk.py +++ b/src/blosc2/schunk.py @@ -860,8 +860,8 @@ def decompress_chunk(self, nchunk: int, dst: object = None) -> str | bytes: The index of the chunk that will be decompressed. dst: NumPy object or bytearray The destination NumPy object or bytearray to fill, the length - of which must be greater than 0. The user must ensure - that it has enough capacity to host the decompressed + of which must be greater than 0. It must be writable and + have enough capacity to host the decompressed chunk. Default is None, meaning that a new bytes object is created, filled and returned. @@ -874,8 +874,12 @@ def decompress_chunk(self, nchunk: int, dst: object = None) -> str | bytes: Raises ------ + BufferError + If :paramref:`dst` is read-only, like a `bytes` object or a read-only memoryview. RunTimeError If a problem is detected. + ValueError + If the length of :paramref:`dst` is 0, or if it is a read-only NumPy array. Examples -------- diff --git a/tests/test_decompress_buffer_sizes.py b/tests/test_decompress_buffer_sizes.py index 50801ea80..7b8406809 100644 --- a/tests/test_decompress_buffer_sizes.py +++ b/tests/test_decompress_buffer_sizes.py @@ -5,8 +5,8 @@ # SPDX-License-Identifier: BSD-3-Clause ####################################################################### -"""The decompression functions stay within the src and dst buffers they are given, and free what they -took when they fail.""" +"""The decompression functions stay within the src and dst buffers they are given, write only into a +writable dst, and free what they took when they fail.""" import array import ctypes @@ -117,6 +117,33 @@ def test_truncated_memcpyed_src(decompress): decompress(src) +def _decompress_chunk(src, dst): + # src is CHUNK, which compresses DATA: decompress the same data from a super-chunk + return blosc2.SChunk(chunksize=NBYTES, data=DATA).decompress_chunk(0, dst) + + +def _readonly_numpy(nbytes): + arr = np.zeros(nbytes, dtype=np.uint8) + arr.flags.writeable = False + return arr + + +@pytest.mark.parametrize("decompress", [*DECOMPRESSORS, _decompress_chunk]) +@pytest.mark.parametrize( + ("make_dst", "exc"), + [ + pytest.param(bytes, BufferError, id="bytes"), + pytest.param(lambda n: memoryview(bytearray(n)).toreadonly(), BufferError, id="readonly-memoryview"), + pytest.param(_readonly_numpy, ValueError, id="readonly-numpy"), + ], +) +def test_readonly_dst_is_refused(decompress, make_dst, exc): + dst = make_dst(NBYTES) + with pytest.raises(exc, match=r"not writable|read-only"): + decompress(CHUNK, dst) + assert bytes(dst) == bytes(NBYTES) + + class _Mallinfo2(ctypes.Structure): _fields_ = [ (name, ctypes.c_size_t)