mirrors/RustPython
0
0 Fork 0
mirror of https://github.com/RustPython/RustPython.git synced 2026-10-29 03:04:38 +08:00
Code Activity
Files
fd41418bd2ff772bfecfcfbce28194c013a828dc
RustPython/Lib/test/test_memoryview.py
T
Jeong, YunWon adf7da508b memoryview: count exports, hash the exporter, and index through __index__ (#8553) 
* memoryview: index through `__index__` and name the dimension

A tuple key was taken apart only when every item was already an `int`, so
`m[I(), 2]` with an `__index__` on `I` was reported as an invalid slice key
rather than indexed. What kind of key it is now follows from the types in
it, the way `is_multiindex` decides, and each item is converted where it
sits; an item that answers `__index__` but raises on the way reports that
rather than making the whole key invalid.

That conversion runs Python, which can release the view, so the
multi-dimensional read goes through `unpack_single` and its re-check
rather than reading the bytes itself.

`IndexError` named the index that was out of bounds where
`lookup_dimension` names the dimension it was out of bounds on, counted
from one, and the one-dimensional path had a message of its own.

Assisted-by: Claude

* memoryview: count what a view has exported, and hash the exporter

The view kept no count of the buffers taken from it, so `release()` always
succeeded, even while something was still reading through an export. It now
answers `BufferError` the way `_memory_release` does, and `__exit__` reports
the same refusal.

`memory_hash` asks the exporter for its hash and throws the answer away, so
a view is no more hashable than what it looks at; a read-only view over a
`bytearray` hashed here where CPython refuses it. Asking runs Python, so the
view counts as exported for the duration and a release attempted from inside
that hash is refused rather than obeyed.

Assisted-by: Claude

* bytes, bytearray, memoryview: measure the hex separator as `len()` does

`hex()` took the separator's length from its bytes rather than from the
object, so a `bytes` subclass that answers `__len__` was measured by what it
holds instead of what it says: `b'abcd'.hex(S(b'::'))` with `__len__`
returning 1 was refused where `_Py_strhex_impl` accepts it, and a two-byte
answer was accepted where it refuses. The check also came after the
`bytes_per_sep == 0` and empty-data shortcuts, so `b'abcd'.hex('::', 0)`
never reached it at all.

Measuring runs Python, and the bytes to be written out must not be borrowed
while it does, so the separator is now resolved from the arguments before
the buffer is reached. For a memoryview that also means the view counts as
exported for the duration, so a release attempted from inside `__len__` is
refused.

Assisted-by: Claude

* array: refuse to hash an array

`array.array` is mutable but inherited `object.__hash__`, so an array could
be put in a set or used as a dict key and then changed underneath it. It is
unhashable, as `tp_hash = PyObject_HashNotImplemented` makes it.

A read-only memoryview over an array now reports the exporter as unhashable
too, which is what `memory_hash` asks it for.

Assisted-by: Claude

* Remove expectedFailure from three memoryview tests

test_hash_use_after_free, test_hex_use_after_free and
test_use_released_memory pass.

Assisted-by: Claude

* memoryview: bound the dimensions a cast can name

`cast()` accepted a shape of any length, so `mv.cast('B', (1,)*64 + (8,))`
built a 65-dimensional view where `memoryview_cast_impl` refuses anything
past `PyBUF_MAX_NDIM`. The limit is answered before the shape is looked at
any further, as it is there.

Assisted-by: Claude
2026-08-18 22:19:08 +09:00

849 lines
29 KiB
Python
Vendored
Raw Blame History

"""Unit tests for the memoryview
Some tests are in test_bytes. Many tests that require _testbuffer.ndarray
are in test_buffer.
"""
import unittest
import test.support
import sys
import gc
import weakref
import array
import io
import copy
import pickle
import struct
from itertools import product
from test import support
from test.support import import_helper, threading_helper
class MyObject:
pass
class AbstractMemoryTests:
source_bytes = b"abcdef"
@property
def _source(self):
return self.source_bytes
@property
def _types(self):
return filter(None, [self.ro_type, self.rw_type])
def check_getitem_with_type(self, tp):
b = tp(self._source)
oldrefcount = sys.getrefcount(b)
m = self._view(b)
self.assertEqual(m[0], ord(b"a"))
self.assertIsInstance(m[0], int)
self.assertEqual(m[5], ord(b"f"))
self.assertEqual(m[-1], ord(b"f"))
self.assertEqual(m[-6], ord(b"a"))
# Bounds checking
self.assertRaises(IndexError, lambda: m[6])
self.assertRaises(IndexError, lambda: m[-7])
self.assertRaises(IndexError, lambda: m[sys.maxsize])
self.assertRaises(IndexError, lambda: m[-sys.maxsize])
# Type checking
self.assertRaises(TypeError, lambda: m[None])
self.assertRaises(TypeError, lambda: m[0.0])
self.assertRaises(TypeError, lambda: m["a"])
m = None
self.assertEqual(sys.getrefcount(b), oldrefcount)
def test_getitem(self):
for tp in self._types:
self.check_getitem_with_type(tp)
def test_index(self):
for tp in self._types:
b = tp(self._source)
m = self._view(b) # may be a sub-view
l = m.tolist()
k = 2 * len(self._source)
for chi in self._source:
if chi in l:
self.assertEqual(m.index(chi), l.index(chi))
else:
self.assertRaises(ValueError, m.index, chi)
for start, stop in product(range(-k, k), range(-k, k)):
index = -1
try:
index = l.index(chi, start, stop)
except ValueError:
pass
if index == -1:
self.assertRaises(ValueError, m.index, chi, start, stop)
else:
self.assertEqual(m.index(chi, start, stop), index)
def test_iter(self):
for tp in self._types:
b = tp(self._source)
m = self._view(b)
self.assertEqual(list(m), [m[i] for i in range(len(m))])
def test_count(self):
for tp in self._types:
b = tp(self._source)
m = self._view(b)
l = m.tolist()
for ch in list(m):
self.assertEqual(m.count(ch), l.count(ch))
b = tp((b'a' * 5) + (b'c' * 3))
m = self._view(b) # may be sliced
l = m.tolist()
with self.subTest('count', buffer=b):
self.assertEqual(m.count(ord('a')), l.count(ord('a')))
self.assertEqual(m.count(ord('b')), l.count(ord('b')))
self.assertEqual(m.count(ord('c')), l.count(ord('c')))
def test_setitem_readonly(self):
if not self.ro_type:
self.skipTest("no read-only type to test")
b = self.ro_type(self._source)
oldrefcount = sys.getrefcount(b)
m = self._view(b)
def setitem(value):
m[0] = value
self.assertRaises(TypeError, setitem, b"a")
self.assertRaises(TypeError, setitem, 65)
self.assertRaises(TypeError, setitem, memoryview(b"a"))
m = None
self.assertEqual(sys.getrefcount(b), oldrefcount)
def test_setitem_writable(self):
if not self.rw_type:
self.skipTest("no writable type to test")
tp = self.rw_type
b = self.rw_type(self._source)
oldrefcount = sys.getrefcount(b)
m = self._view(b)
m[0] = ord(b'1')
self._check_contents(tp, b, b"1bcdef")
m[0:1] = tp(b"0")
self._check_contents(tp, b, b"0bcdef")
m[1:3] = tp(b"12")
self._check_contents(tp, b, b"012def")
m[1:1] = tp(b"")
self._check_contents(tp, b, b"012def")
m[:] = tp(b"abcdef")
self._check_contents(tp, b, b"abcdef")
# Overlapping copies of a view into itself
m[0:3] = m[2:5]
self._check_contents(tp, b, b"cdedef")
m[:] = tp(b"abcdef")
m[2:5] = m[0:3]
self._check_contents(tp, b, b"ababcf")
def setitem(key, value):
m[key] = tp(value)
# Bounds checking
self.assertRaises(IndexError, setitem, 6, b"a")
self.assertRaises(IndexError, setitem, -7, b"a")
self.assertRaises(IndexError, setitem, sys.maxsize, b"a")
self.assertRaises(IndexError, setitem, -sys.maxsize, b"a")
# Wrong index/slice types
self.assertRaises(TypeError, setitem, 0.0, b"a")
self.assertRaises(TypeError, setitem, (0,), b"a")
self.assertRaises(TypeError, setitem, (slice(0,1,1), 0), b"a")
self.assertRaises(TypeError, setitem, (0, slice(0,1,1)), b"a")
self.assertRaises(TypeError, setitem, (0,), b"a")
self.assertRaises(TypeError, setitem, "a", b"a")
# Not implemented: multidimensional slices
slices = (slice(0,1,1), slice(0,1,2))
self.assertRaises(NotImplementedError, setitem, slices, b"a")
# Trying to resize the memory object
exc = ValueError if m.format == 'c' else TypeError
self.assertRaises(exc, setitem, 0, b"")
self.assertRaises(exc, setitem, 0, b"ab")
self.assertRaises(ValueError, setitem, slice(1,1), b"a")
self.assertRaises(ValueError, setitem, slice(0,2), b"a")
m = None
self.assertEqual(sys.getrefcount(b), oldrefcount)
def test_delitem(self):
for tp in self._types:
b = tp(self._source)
m = self._view(b)
with self.assertRaises(TypeError):
del m[1]
with self.assertRaises(TypeError):
del m[1:4]
def test_tobytes(self):
for tp in self._types:
m = self._view(tp(self._source))
b = m.tobytes()
# This calls self.getitem_type() on each separate byte of b"abcdef"
expected = b"".join(
self.getitem_type(bytes([c])) for c in b"abcdef")
self.assertEqual(b, expected)
self.assertIsInstance(b, bytes)
def test_tolist(self):
for tp in self._types:
m = self._view(tp(self._source))
l = m.tolist()
self.assertEqual(l, list(b"abcdef"))
def test_compare(self):
# memoryviews can compare for equality with other objects
# having the buffer interface.
for tp in self._types:
m = self._view(tp(self._source))
for tp_comp in self._types:
self.assertTrue(m == tp_comp(b"abcdef"))
self.assertFalse(m != tp_comp(b"abcdef"))
self.assertFalse(m == tp_comp(b"abcde"))
self.assertTrue(m != tp_comp(b"abcde"))
self.assertFalse(m == tp_comp(b"abcde1"))
self.assertTrue(m != tp_comp(b"abcde1"))
self.assertTrue(m == m)
self.assertTrue(m == m[:])
self.assertTrue(m[0:6] == m[:])
self.assertFalse(m[0:5] == m)
# Comparison with objects which don't support the buffer API
self.assertFalse(m == "abcdef")
self.assertTrue(m != "abcdef")
self.assertFalse("abcdef" == m)
self.assertTrue("abcdef" != m)
# Unordered comparisons
for c in (m, b"abcdef"):
self.assertRaises(TypeError, lambda: m < c)
self.assertRaises(TypeError, lambda: c <= m)
self.assertRaises(TypeError, lambda: m >= c)
self.assertRaises(TypeError, lambda: c > m)
def check_attributes_with_type(self, tp):
m = self._view(tp(self._source))
self.assertEqual(m.format, self.format)
self.assertEqual(m.itemsize, self.itemsize)
self.assertEqual(m.ndim, 1)
self.assertEqual(m.shape, (6,))
self.assertEqual(len(m), 6)
self.assertEqual(m.strides, (self.itemsize,))
self.assertEqual(m.suboffsets, ())
return m
def test_attributes_readonly(self):
if not self.ro_type:
self.skipTest("no read-only type to test")
m = self.check_attributes_with_type(self.ro_type)
self.assertEqual(m.readonly, True)
def test_attributes_writable(self):
if not self.rw_type:
self.skipTest("no writable type to test")
m = self.check_attributes_with_type(self.rw_type)
self.assertEqual(m.readonly, False)
def test_getbuffer(self):
# Test PyObject_GetBuffer() on a memoryview object.
for tp in self._types:
b = tp(self._source)
oldrefcount = sys.getrefcount(b)
m = self._view(b)
oldviewrefcount = sys.getrefcount(m)
s = str(m, "utf-8")
self._check_contents(tp, b, s.encode("utf-8"))
self.assertEqual(sys.getrefcount(m), oldviewrefcount)
m = None
self.assertEqual(sys.getrefcount(b), oldrefcount)
def test_gc(self):
for tp in self._types:
if not isinstance(tp, type):
# If tp is a factory rather than a plain type, skip
continue
class MyView():
def __init__(self, base):
self.m = memoryview(base)
class MySource(tp):
pass
# Create a reference cycle through a memoryview object.
# This exercises mbuf_clear().
b = MySource(tp(b'abc'))
m = self._view(b)
o = MyObject()
b.m = m
b.o = o
wr = weakref.ref(o)
b = m = o = None
# The cycle must be broken
gc.collect()
self.assertTrue(wr() is None, wr())
# This exercises memory_clear().
m = MyView(tp(b'abc'))
o = MyObject()
m.x = m
m.o = o
wr = weakref.ref(o)
m = o = None
# The cycle must be broken
gc.collect()
self.assertTrue(wr() is None, wr())
def _check_released(self, m, tp):
check = self.assertRaisesRegex(ValueError, "released")
with check: bytes(m)
with check: m.tobytes()
with check: m.tolist()
with check: m[0]
with check: m[0] = b'x'
with check: len(m)
with check: m.format
with check: m.itemsize
with check: m.ndim
with check: m.readonly
with check: m.shape
with check: m.strides
with check:
with m:
pass
# str() and repr() still function
self.assertIn("released memory", str(m))
self.assertIn("released memory", repr(m))
self.assertEqual(m, m)
self.assertNotEqual(m, memoryview(tp(self._source)))
self.assertNotEqual(m, tp(self._source))
def test_contextmanager(self):
for tp in self._types:
b = tp(self._source)
m = self._view(b)
with m as cm:
self.assertIs(cm, m)
self._check_released(m, tp)
m = self._view(b)
# Can release explicitly inside the context manager
with m:
m.release()
def test_release(self):
for tp in self._types:
b = tp(self._source)
m = self._view(b)
m.release()
self._check_released(m, tp)
# Can be called a second time (it's a no-op)
m.release()
self._check_released(m, tp)
def test_writable_readonly(self):
# Issue #10451: memoryview incorrectly exposes a readonly
# buffer as writable causing a segfault if using mmap
tp = self.ro_type
if tp is None:
self.skipTest("no read-only type to test")
b = tp(self._source)
m = self._view(b)
i = io.BytesIO(b'ZZZZ')
self.assertRaises(TypeError, i.readinto, m)
def test_getbuf_fail(self):
self.assertRaises(TypeError, self._view, {})
def test_hash(self):
# Memoryviews of readonly (hashable) types are hashable, and they
# hash as hash(obj.tobytes()).
tp = self.ro_type
if tp is None:
self.skipTest("no read-only type to test")
b = tp(self._source)
m = self._view(b)
self.assertEqual(hash(m), hash(b"abcdef"))
# Releasing the memoryview keeps the stored hash value (as with weakrefs)
m.release()
self.assertEqual(hash(m), hash(b"abcdef"))
# Hashing a memoryview for the first time after it is released
# results in an error (as with weakrefs).
m = self._view(b)
m.release()
self.assertRaises(ValueError, hash, m)
def test_hash_writable(self):
# Memoryviews of writable types are unhashable
tp = self.rw_type
if tp is None:
self.skipTest("no writable type to test")
b = tp(self._source)
m = self._view(b)
self.assertRaises(ValueError, hash, m)
def test_hash_use_after_free(self):
# Prevent crash in memoryview(v).__hash__ with re-entrant v.__hash__.
# Regression test for https://github.com/python/cpython/issues/142664.
class E(array.array):
def __hash__(self):
mv.release()
self.clear()
return 123
v = E('B', b'A' * 4096)
mv = memoryview(v).toreadonly() # must be read-only for hash()
self.assertRaises(BufferError, hash, mv)
self.assertRaises(BufferError, mv.__hash__)
def test_weakref(self):
# Check memoryviews are weakrefable
for tp in self._types:
b = tp(self._source)
m = self._view(b)
L = []
def callback(wr, b=b):
L.append(b)
wr = weakref.ref(m, callback)
self.assertIs(wr(), m)
del m
test.support.gc_collect()
self.assertIs(wr(), None)
self.assertIs(L[0], b)
def test_reversed(self):
for tp in self._types:
b = tp(self._source)
m = self._view(b)
aslist = list(reversed(m.tolist()))
self.assertEqual(list(reversed(m)), aslist)
self.assertEqual(list(reversed(m)), list(m[::-1]))
def test_toreadonly(self):
for tp in self._types:
b = tp(self._source)
m = self._view(b)
mm = m.toreadonly()
self.assertTrue(mm.readonly)
self.assertTrue(memoryview(mm).readonly)
self.assertEqual(mm.tolist(), m.tolist())
mm.release()
m.tolist()
def test_issue22668(self):
a = array.array('H', [256, 256, 256, 256])
x = memoryview(a)
m = x.cast('B')
b = m.cast('H')
c = b[0:2]
d = memoryview(b)
del b
self.assertEqual(c[0], 256)
self.assertEqual(d[0], 256)
self.assertEqual(c.format, "H")
self.assertEqual(d.format, "H")
_ = m.cast('I')
self.assertEqual(c[0], 256)
self.assertEqual(d[0], 256)
self.assertEqual(c.format, "H")
self.assertEqual(d.format, "H")
def test_hex_use_after_free(self):
# Prevent UAF in memoryview.hex(sep) with re-entrant sep.__len__.
# Regression test for https://github.com/python/cpython/issues/143195.
ba = bytearray(b'A' * 1024)
mv = memoryview(ba)
class S(bytes):
def __len__(self):
mv.release()
ba.clear()
return 1
self.assertRaises(BufferError, mv.hex, S(b':'))
# Variations on source objects for the buffer: bytes-like objects, then arrays
# with itemsize > 1.
# NOTE: support for multi-dimensional objects is unimplemented.
class BaseBytesMemoryTests(AbstractMemoryTests):
ro_type = bytes
rw_type = bytearray
getitem_type = bytes
itemsize = 1
format = 'B'
class BaseArrayMemoryTests(AbstractMemoryTests):
ro_type = None
rw_type = lambda self, b: array.array('i', list(b))
getitem_type = lambda self, b: array.array('i', list(b)).tobytes()
itemsize = array.array('i').itemsize
format = 'i'
@unittest.skip('XXX test should be adapted for non-byte buffers')
def test_getbuffer(self):
pass
@unittest.skip('XXX NotImplementedError: tolist() only supports byte views')
def test_tolist(self):
pass
# Variations on indirection levels: memoryview, slice of memoryview,
# slice of slice of memoryview.
# This is important to test allocation subtleties.
class BaseMemoryviewTests:
def _view(self, obj):
return memoryview(obj)
def _check_contents(self, tp, obj, contents):
self.assertEqual(obj, tp(contents))
def test_count(self):
super().test_count()
for tp in self._types:
b = tp((b'a' * 5) + (b'c' * 3))
m = self._view(b) # should not be sliced
self.assertEqual(len(b), len(m))
with self.subTest('count', buffer=b):
self.assertEqual(m.count(ord('a')), 5)
self.assertEqual(m.count(ord('b')), 0)
self.assertEqual(m.count(ord('c')), 3)
class BaseMemorySliceTests:
source_bytes = b"XabcdefY"
def _view(self, obj):
m = memoryview(obj)
return m[1:7]
def _check_contents(self, tp, obj, contents):
self.assertEqual(obj[1:7], tp(contents))
def test_refs(self):
for tp in self._types:
m = memoryview(tp(self._source))
oldrefcount = sys.getrefcount(m)
m[1:2]
self.assertEqual(sys.getrefcount(m), oldrefcount)
class BaseMemorySliceSliceTests:
source_bytes = b"XabcdefY"
def _view(self, obj):
m = memoryview(obj)
return m[:7][1:]
def _check_contents(self, tp, obj, contents):
self.assertEqual(obj[1:7], tp(contents))
# Concrete test classes
class BytesMemoryviewTest(unittest.TestCase,
BaseMemoryviewTests, BaseBytesMemoryTests):
def test_constructor(self):
for tp in self._types:
ob = tp(self._source)
self.assertTrue(memoryview(ob))
self.assertTrue(memoryview(object=ob))
self.assertRaises(TypeError, memoryview)
self.assertRaises(TypeError, memoryview, ob, ob)
self.assertRaises(TypeError, memoryview, argument=ob)
self.assertRaises(TypeError, memoryview, ob, argument=True)
@unittest.expectedFailure # TODO: RUSTPYTHON; AssertionError: False is not true : <test.test_memoryview.MyObject object at 0x84ecb1920>
def test_gc(self):
return super().test_gc()
class ArrayMemoryviewTest(unittest.TestCase,
BaseMemoryviewTests, BaseArrayMemoryTests):
def test_array_assign(self):
# Issue #4569: segfault when mutating a memoryview with itemsize != 1
a = array.array('i', range(10))
m = memoryview(a)
new_a = array.array('i', range(9, -1, -1))
m[:] = new_a
self.assertEqual(a, new_a)
def test_boolean_format(self):
# Test '?' format (keep all the checks below for UBSan)
# See github.com/python/cpython/issues/148390.
# m1a and m1b are equivalent to [False, True, False]
m1a = memoryview(b'\0\2\0').cast('?')
self.assertEqual(m1a.tolist(), [False, True, False])
m1b = memoryview(b'\0\4\0').cast('?')
self.assertEqual(m1b.tolist(), [False, True, False])
self.assertEqual(m1a, m1b)
# m2a and m2b are equivalent to [True, True, True]
m2a = memoryview(b'\1\3\5').cast('?')
self.assertEqual(m2a.tolist(), [True, True, True])
m2b = memoryview(b'\2\4\6').cast('?')
self.assertEqual(m2b.tolist(), [True, True, True])
self.assertEqual(m2a, m2b)
allbytes = bytes(range(256))
allbytes = memoryview(allbytes).cast('?')
self.assertEqual(allbytes.tolist(), [False] + [True] * 255)
class BytesMemorySliceTest(unittest.TestCase,
BaseMemorySliceTests, BaseBytesMemoryTests):
pass
@unittest.expectedFailure # TODO: RUSTPYTHON; AssertionError: False is not true : <test.test_memoryview.MyObject object at 0x84ecb13e0>
def test_gc(self):
return super().test_gc()
class ArrayMemorySliceTest(unittest.TestCase,
BaseMemorySliceTests, BaseArrayMemoryTests):
pass
class BytesMemorySliceSliceTest(unittest.TestCase,
BaseMemorySliceSliceTests, BaseBytesMemoryTests):
pass
@unittest.expectedFailure # TODO: RUSTPYTHON; AssertionError: False is not true : <test.test_memoryview.MyObject object at 0x84ddca1c0>
def test_gc(self):
return super().test_gc()
class ArrayMemorySliceSliceTest(unittest.TestCase,
BaseMemorySliceSliceTests, BaseArrayMemoryTests):
pass
class OtherTest(unittest.TestCase):
def test_ctypes_cast(self):
# Issue 15944: Allow all source formats when casting to bytes.
ctypes = import_helper.import_module("ctypes")
p6 = bytes(ctypes.c_double(0.6))
d = ctypes.c_double()
m = memoryview(d).cast("B")
m[:2] = p6[:2]
m[2:] = p6[2:]
self.assertEqual(d.value, 0.6)
for format in "Bbc":
with self.subTest(format):
d = ctypes.c_double()
m = memoryview(d).cast(format)
m[:2] = memoryview(p6).cast(format)[:2]
m[2:] = memoryview(p6).cast(format)[2:]
self.assertEqual(d.value, 0.6)
def test_half_float(self):
half_data = struct.pack('eee', 0.0, -1.5, 1.5)
float_data = struct.pack('fff', 0.0, -1.5, 1.5)
half_view = memoryview(half_data).cast('e')
float_view = memoryview(float_data).cast('f')
self.assertEqual(half_view.nbytes * 2, float_view.nbytes)
self.assertListEqual(half_view.tolist(), float_view.tolist())
def test_memoryview_hex(self):
# Issue #9951: memoryview.hex() segfaults with non-contiguous buffers.
x = b'0' * 200000
m1 = memoryview(x)
m2 = m1[::-1]
self.assertEqual(m2.hex(), '30' * 200000)
def test_memoryview_hex_separator(self):
x = bytes(range(97, 102))
m1 = memoryview(x)
m2 = m1[::-1]
self.assertEqual(m2.hex(':'), '65:64:63:62:61')
self.assertEqual(m2.hex(':', 2), '65:6463:6261')
self.assertEqual(m2.hex(':', -2), '6564:6362:61')
self.assertEqual(m2.hex(sep=':', bytes_per_sep=2), '65:6463:6261')
self.assertEqual(m2.hex(sep=':', bytes_per_sep=-2), '6564:6362:61')
for bytes_per_sep in 5, -5, 2**31-1, -(2**31-1):
with self.subTest(bytes_per_sep=bytes_per_sep):
self.assertEqual(m2.hex(':', bytes_per_sep), '6564636261')
for bytes_per_sep in 2**31, -2**31, 2**1000, -2**1000:
with self.subTest(bytes_per_sep=bytes_per_sep):
try:
self.assertEqual(m2.hex(':', bytes_per_sep), '6564636261')
except OverflowError:
pass
def test_copy(self):
m = memoryview(b'abc')
with self.assertRaises(TypeError):
copy.copy(m)
def test_pickle(self):
m = memoryview(b'abc')
for proto in range(pickle.HIGHEST_PROTOCOL + 1):
with self.assertRaises(TypeError):
pickle.dumps(m, proto)
def test_use_released_memory(self):
# gh-92888: Previously it was possible to use a memoryview even after
# backing buffer is freed in certain cases. This tests that those
# cases raise an exception.
size = 128
def release():
m.release()
nonlocal ba
ba = bytearray(size)
class MyIndex:
def __index__(self):
release()
return 4
class MyFloat:
def __float__(self):
release()
return 4.25
class MyBool:
def __bool__(self):
release()
return True
ba = None
m = memoryview(bytearray(b'\xff'*size))
with self.assertRaises(ValueError):
m[MyIndex()]
ba = None
m = memoryview(bytearray(b'\xff'*size))
self.assertEqual(list(m[:MyIndex()]), [255] * 4)
ba = None
m = memoryview(bytearray(b'\xff'*size))
self.assertEqual(list(m[MyIndex():8]), [255] * 4)
ba = None
m = memoryview(bytearray(b'\xff'*size)).cast('B', (64, 2))
with self.assertRaisesRegex(ValueError, "operation forbidden"):
m[MyIndex(), 0]
ba = None
m = memoryview(bytearray(b'\xff'*size)).cast('B', (2, 64))
with self.assertRaisesRegex(ValueError, "operation forbidden"):
m[0, MyIndex()]
ba = None
m = memoryview(bytearray(b'\xff'*size))
with self.assertRaisesRegex(ValueError, "operation forbidden"):
m[MyIndex()] = 42
self.assertEqual(ba[:8], b'\0'*8)
ba = None
m = memoryview(bytearray(b'\xff'*size))
with self.assertRaisesRegex(ValueError, "operation forbidden"):
m[:MyIndex()] = b'spam'
self.assertEqual(ba[:8], b'\0'*8)
ba = None
m = memoryview(bytearray(b'\xff'*size))
with self.assertRaisesRegex(ValueError, "operation forbidden"):
m[MyIndex():8] = b'spam'
self.assertEqual(ba[:8], b'\0'*8)
ba = None
m = memoryview(bytearray(b'\xff'*size)).cast('B', (64, 2))
with self.assertRaisesRegex(ValueError, "operation forbidden"):
m[MyIndex(), 0] = 42
self.assertEqual(ba[8:16], b'\0'*8)
ba = None
m = memoryview(bytearray(b'\xff'*size)).cast('B', (2, 64))
with self.assertRaisesRegex(ValueError, "operation forbidden"):
m[0, MyIndex()] = 42
self.assertEqual(ba[:8], b'\0'*8)
ba = None
m = memoryview(bytearray(b'\xff'*size))
with self.assertRaisesRegex(ValueError, "operation forbidden"):
m[0] = MyIndex()
self.assertEqual(ba[:8], b'\0'*8)
for fmt in 'bhilqnBHILQN':
with self.subTest(fmt=fmt):
ba = None
m = memoryview(bytearray(b'\xff'*size)).cast(fmt)
with self.assertRaisesRegex(ValueError, "operation forbidden"):
m[0] = MyIndex()
self.assertEqual(ba[:8], b'\0'*8)
for fmt in 'fd':
with self.subTest(fmt=fmt):
ba = None
m = memoryview(bytearray(b'\xff'*size)).cast(fmt)
with self.assertRaisesRegex(ValueError, "operation forbidden"):
m[0] = MyFloat()
self.assertEqual(ba[:8], b'\0'*8)
ba = None
m = memoryview(bytearray(b'\xff'*size)).cast('?')
with self.assertRaisesRegex(ValueError, "operation forbidden"):
m[0] = MyBool()
self.assertEqual(ba[:8], b'\0'*8)
def test_buffer_reference_loop(self):
m = memoryview(b'abc').__buffer__(0)
o = MyObject()
o.m = m
o.o = o
wr = weakref.ref(o)
del m, o
gc.collect()
self.assertIsNone(wr())
@unittest.expectedFailure # TODO: RUSTPYTHON; AttributeError: module 'pickle' has no attribute 'PickleBuffer'
def test_picklebuffer_reference_loop(self):
pb = pickle.PickleBuffer(memoryview(b'abc'))
o = MyObject()
o.pb = pb
o.o = o
wr = weakref.ref(o)
del pb, o
gc.collect()
self.assertIsNone(wr())
@threading_helper.requires_working_threading()
@support.requires_resource("cpu")
class RacingTest(unittest.TestCase):
def test_racing_getbuf_and_releasebuf(self):
"""Repeatly access the memoryview for racing."""
try:
from multiprocessing.managers import SharedMemoryManager
except ImportError:
self.skipTest("Test requires multiprocessing")
from threading import Thread, Event
start = Event()
with SharedMemoryManager() as smm:
obj = smm.ShareableList(range(100))
def test():
# Issue gh-127085, the `ShareableList.count` is just a
# convenient way to mess the `exports` counter of `memoryview`,
# this issue has no direct relation with `ShareableList`.
start.wait(support.SHORT_TIMEOUT)
for i in range(10):
obj.count(1)
threads = [Thread(target=test) for _ in range(10)]
with threading_helper.start_threads(threads):
start.set()
del obj
if __name__ == "__main__":
unittest.main()
View Git Blame Copy Permalink