| """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) | |
| 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_compare_equal(self): | |
| # A memoryview is equal to itself: there is no need to compare | |
| # individual values. This is not true for float values since they can | |
| # be NaN, and NaN is not equal to itself. | |
| def check_equal(view, is_equal): | |
| self.assertEqual(view == view, is_equal) | |
| self.assertEqual(view != view, not is_equal) | |
| # Comparison with a different memoryview doesn't use | |
| # the optimization and should give the same result. | |
| view2 = memoryview(view) | |
| self.assertEqual(view2 == view, is_equal) | |
| self.assertEqual(view2 != view2, not is_equal) | |
| # Test integer formats | |
| for int_format in 'bBhHiIlLqQ': | |
| with self.subTest(format=int_format): | |
| a = array.array(int_format, [1, 2, 3]) | |
| m = memoryview(a) | |
| check_equal(m, True) | |
| if int_format in 'bB': | |
| m2 = m.cast('@' + m.format) | |
| check_equal(m2, True) | |
| # Test 'c' format | |
| a = array.array('B', [1, 2, 3]) | |
| m = memoryview(a.tobytes()).cast('c') | |
| check_equal(m, True) | |
| # Test 'n' and 'N' formats | |
| if struct.calcsize('L') == struct.calcsize('N'): | |
| int_format = 'L' | |
| elif struct.calcsize('Q') == struct.calcsize('N'): | |
| int_format = 'Q' | |
| else: | |
| int_format = None | |
| if int_format: | |
| a = array.array(int_format, [1, 2, 3]) | |
| m = memoryview(a.tobytes()).cast('N') | |
| check_equal(m, True) | |
| m = memoryview(a.tobytes()).cast('n') | |
| check_equal(m, True) | |
| # Test '?' format | |
| m = memoryview(b'\0\1\2').cast('?') | |
| check_equal(m, True) | |
| # Test float formats | |
| for float_format in 'fd': | |
| with self.subTest(format=float_format): | |
| a = array.array(float_format, [1.0, 2.0, float('nan')]) | |
| m = memoryview(a) | |
| # nan is not equal to nan | |
| check_equal(m, False) | |
| a = array.array(float_format, [1.0, 2.0, 3.0]) | |
| m = memoryview(a) | |
| check_equal(m, True) | |
| # Test complex formats | |
| for complex_format in ('Zf', 'Zd'): | |
| with self.subTest(format=complex_format): | |
| data = struct.pack(complex_format * 3, 1.0, 2.0, float('nan')) | |
| m = memoryview(data).cast(complex_format) | |
| # nan is not equal to nan | |
| check_equal(m, False) | |
| data = struct.pack(complex_format * 3, 1.0, 2.0, 3.0) | |
| m = memoryview(data).cast(complex_format) | |
| check_equal(m, True) | |
| 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 | |
| class ArrayMemorySliceTest(unittest.TestCase, | |
| BaseMemorySliceTests, BaseArrayMemoryTests): | |
| pass | |
| class BytesMemorySliceSliceTest(unittest.TestCase, | |
| BaseMemorySliceSliceTests, BaseBytesMemoryTests): | |
| pass | |
| class ArrayMemorySliceSliceTest(unittest.TestCase, | |
| BaseMemorySliceSliceTests, BaseArrayMemoryTests): | |
| pass | |
| class OtherTest(unittest.TestCase): | |
| def test_cast_order(self): | |
| # gh-78959: cast(order=...) selects the result memory layout. | |
| b = bytearray(range(6)) | |
| mv = memoryview(b) | |
| c = mv.cast('B', shape=(3, 2)) # default: C-contiguous | |
| self.assertTrue(c.c_contiguous) | |
| self.assertFalse(c.f_contiguous) | |
| self.assertEqual(c.strides, (2, 1)) | |
| self.assertEqual(c.tolist(), [[0, 1], [2, 3], [4, 5]]) | |
| f = mv.cast('B', shape=(3, 2), order='F') # Fortran-contiguous | |
| self.assertTrue(f.f_contiguous) | |
| self.assertFalse(f.c_contiguous) | |
| self.assertEqual(f.strides, (1, 3)) | |
| self.assertEqual(f.tolist(), [[0, 3], [1, 4], [2, 5]]) | |
| # zero-copy view: the flat physical bytes are unchanged | |
| self.assertEqual(f.tobytes('A'), bytes(range(6))) | |
| # explicit 'C' is the default | |
| self.assertTrue(mv.cast('B', shape=(3, 2), order='C').c_contiguous) | |
| # a wider format works too (strides scale by itemsize) | |
| m = memoryview(struct.pack('6i', *range(6))).cast('i', shape=(3, 2), | |
| order='F') | |
| self.assertTrue(m.f_contiguous) | |
| self.assertEqual(m.strides, (4, 12)) | |
| self.assertEqual(m.tolist(), [[0, 3], [1, 4], [2, 5]]) | |
| # more than two dimensions | |
| m3 = memoryview(bytearray(range(24))).cast('B', shape=(2, 3, 4), | |
| order='F') | |
| self.assertTrue(m3.f_contiguous) | |
| self.assertEqual(m3.strides, (1, 2, 6)) | |
| # order is keyword-only and only the strings 'C'/'F' are accepted | |
| self.assertRaises(TypeError, mv.cast, 'B', (3, 2), 'F') | |
| self.assertRaises(TypeError, mv.cast, 'B', shape=(3, 2), order=None) | |
| self.assertRaises(ValueError, mv.cast, 'B', shape=(3, 2), order='X') | |
| self.assertRaises(ValueError, mv.cast, 'B', shape=(3, 2), order='A') | |
| def test_cast_order_writable(self): | |
| # An F-contiguous cast shares memory with the original buffer. | |
| b = bytearray(6) | |
| f = memoryview(b).cast('B', shape=(3, 2), order='F') | |
| f[0, 1] = 9 # column-major: physical offset 3 | |
| self.assertEqual(b[3], 9) | |
| 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_complex_types(self): | |
| float_complex_data = struct.pack('ZfZfZf', 0.0, -1.5j, 1+2j) | |
| double_complex_data = struct.pack('ZdZdZd', 0.0, -1.5j, 1+2j) | |
| float_complex_view = memoryview(float_complex_data).cast('Zf') | |
| double_complex_view = memoryview(double_complex_data).cast('Zd') | |
| self.assertEqual(float_complex_view.nbytes * 2, double_complex_view.nbytes) | |
| self.assertListEqual(float_complex_view.tolist(), double_complex_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, sys.maxsize, -sys.maxsize: | |
| with self.subTest(bytes_per_sep=bytes_per_sep): | |
| self.assertEqual(m2.hex(':', bytes_per_sep), '6564636261') | |
| for bytes_per_sep in sys.maxsize+1, -sys.maxsize-1, 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()) | |
| 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()) | |
| def test_overflows_in_floats(self): | |
| half_data = array.array('e', [0.0]) | |
| float_data = array.array('f', [0.0]) | |
| complex_data = array.array('Zf', [123+321j]) | |
| half_view = memoryview(half_data) | |
| float_view = memoryview(float_data) | |
| complex_view = memoryview(complex_data) | |
| with self.assertRaises(ValueError): | |
| half_view[0] = 123456.0 | |
| with self.assertRaises(ValueError): | |
| float_view[0] = 1e300 | |
| with self.assertRaises(ValueError): | |
| complex_view[0] = 1e300 | |
| self.assertEqual(complex_view[0], 123+321j) | |
| with self.assertRaises(ValueError): | |
| complex_view[0] = 1e300j | |
| self.assertEqual(complex_view[0], 123+321j) | |
| @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() |