Rust-related/RustPython
1
0
Fork 1
mirror of https://github.com/RustPython/RustPython.git synced 2026-06-02 19:39:49 +09:00
Code Issues Packages Projects Releases Wiki Activity

Move pickletester.py to the correct place

Browse Source
This commit is contained in:
ShaharNaveh
2026-02-05 13:32:40 +02:00
parent d7c259c8c9
commit 0258e8d10a
2 changed files with 404 additions and 5211 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5139
Lib/pickletester.py vendored
View File

File diff suppressed because it is too large Load Diff

476
Lib/test/pickletester.py vendored
View File

@@ -1012,6 +1012,26 @@ class AbstractUnpickleTests:
self.assertIs(self.loads(b'I01\n.'), True)
self.assertIs(self.loads(b'I00\n.'), False)
def test_zero_padded_integers(self):
self.assertEqual(self.loads(b'I010\n.'), 10)
self.assertEqual(self.loads(b'I-010\n.'), -10)
self.assertEqual(self.loads(b'I0010\n.'), 10)
self.assertEqual(self.loads(b'I-0010\n.'), -10)
self.assertEqual(self.loads(b'L010\n.'), 10)
self.assertEqual(self.loads(b'L-010\n.'), -10)
self.assertEqual(self.loads(b'L0010\n.'), 10)
self.assertEqual(self.loads(b'L-0010\n.'), -10)
self.assertEqual(self.loads(b'L010L\n.'), 10)
self.assertEqual(self.loads(b'L-010L\n.'), -10)
def test_nondecimal_integers(self):
self.assertRaises(ValueError, self.loads, b'I0b10\n.')
self.assertRaises(ValueError, self.loads, b'I0o10\n.')
self.assertRaises(ValueError, self.loads, b'I0x10\n.')
self.assertRaises(ValueError, self.loads, b'L0b10L\n.')
self.assertRaises(ValueError, self.loads, b'L0o10L\n.')
self.assertRaises(ValueError, self.loads, b'L0x10L\n.')
def test_empty_bytestring(self):
# issue 11286
empty = self.loads(b'\x80\x03U\x00q\x00.', encoding='koi8-r')
@@ -1234,24 +1254,37 @@ class AbstractUnpickleTests:
self.assertIs(unpickler.find_class('os.path', 'join'), os.path.join)
self.assertIs(unpickler4.find_class('builtins', 'str.upper'), str.upper)
with self.assertRaises(AttributeError):
with self.assertRaisesRegex(AttributeError,
r"module 'builtins' has no attribute 'str\.upper'"):
unpickler.find_class('builtins', 'str.upper')
with self.assertRaises(AttributeError):
with self.assertRaisesRegex(AttributeError,
"module 'math' has no attribute 'spam'"):
unpickler.find_class('math', 'spam')
with self.assertRaises(AttributeError):
with self.assertRaisesRegex(AttributeError,
"module 'math' has no attribute 'spam'"):
unpickler4.find_class('math', 'spam')
with self.assertRaises(AttributeError):
with self.assertRaisesRegex(AttributeError,
r"module 'math' has no attribute 'log\.spam'"):
unpickler.find_class('math', 'log.spam')
with self.assertRaises(AttributeError):
with self.assertRaisesRegex(AttributeError,
r"Can't resolve path 'log\.spam' on module 'math'") as cm:
unpickler4.find_class('math', 'log.spam')
with self.assertRaises(AttributeError):
self.assertEqual(str(cm.exception.__context__),
"'builtin_function_or_method' object has no attribute 'spam'")
with self.assertRaisesRegex(AttributeError,
r"module 'math' has no attribute 'log\.<locals>\.spam'"):
unpickler.find_class('math', 'log.<locals>.spam')
with self.assertRaises(AttributeError):
with self.assertRaisesRegex(AttributeError,
r"Can't resolve path 'log\.<locals>\.spam' on module 'math'") as cm:
unpickler4.find_class('math', 'log.<locals>.spam')
with self.assertRaises(AttributeError):
self.assertEqual(str(cm.exception.__context__),
"'builtin_function_or_method' object has no attribute '<locals>'")
with self.assertRaisesRegex(AttributeError,
"module 'math' has no attribute ''"):
unpickler.find_class('math', '')
with self.assertRaises(AttributeError):
with self.assertRaisesRegex(AttributeError,
"module 'math' has no attribute ''"):
unpickler4.find_class('math', '')
self.assertRaises(ModuleNotFoundError, unpickler.find_class, 'spam', 'log')
self.assertRaises(ValueError, unpickler.find_class, '', 'log')
@@ -1637,48 +1670,77 @@ class AbstractPicklingErrorTests:
obj = REX([print, ()])
for proto in protocols:
with self.subTest(proto=proto):
with self.assertRaises(pickle.PicklingError):
with self.assertRaises(pickle.PicklingError) as cm:
self.dumps(obj, proto)
self.assertEqual(str(cm.exception),
'__reduce__ must return a string or tuple, not list')
self.assertEqual(cm.exception.__notes__, [
'when serializing test.pickletester.REX object'])
obj = REX((print,))
for proto in protocols:
with self.subTest(proto=proto):
with self.assertRaises(pickle.PicklingError):
with self.assertRaises(pickle.PicklingError) as cm:
self.dumps(obj, proto)
self.assertEqual(str(cm.exception),
'tuple returned by __reduce__ must contain 2 through 6 elements')
self.assertEqual(cm.exception.__notes__, [
'when serializing test.pickletester.REX object'])
obj = REX((print, (), None, None, None, None, None))
for proto in protocols:
with self.subTest(proto=proto):
with self.assertRaises(pickle.PicklingError):
with self.assertRaises(pickle.PicklingError) as cm:
self.dumps(obj, proto)
self.assertEqual(str(cm.exception),
'tuple returned by __reduce__ must contain 2 through 6 elements')
self.assertEqual(cm.exception.__notes__, [
'when serializing test.pickletester.REX object'])
def test_bad_reconstructor(self):
obj = REX((42, ()))
for proto in protocols:
with self.subTest(proto=proto):
with self.assertRaises(pickle.PicklingError):
with self.assertRaises(pickle.PicklingError) as cm:
self.dumps(obj, proto)
self.assertEqual(str(cm.exception),
'first item of the tuple returned by __reduce__ '
'must be callable, not int')
self.assertEqual(cm.exception.__notes__, [
'when serializing test.pickletester.REX object'])
def test_unpickleable_reconstructor(self):
obj = REX((UnpickleableCallable(), ()))
for proto in protocols:
with self.subTest(proto=proto):
with self.assertRaises(CustomError):
with self.assertRaises(CustomError) as cm:
self.dumps(obj, proto)
self.assertEqual(cm.exception.__notes__, [
'when serializing test.pickletester.REX reconstructor',
'when serializing test.pickletester.REX object'])
def test_bad_reconstructor_args(self):
obj = REX((print, []))
for proto in protocols:
with self.subTest(proto=proto):
with self.assertRaises(pickle.PicklingError):
with self.assertRaises(pickle.PicklingError) as cm:
self.dumps(obj, proto)
self.assertEqual(str(cm.exception),
'second item of the tuple returned by __reduce__ '
'must be a tuple, not list')
self.assertEqual(cm.exception.__notes__, [
'when serializing test.pickletester.REX object'])
def test_unpickleable_reconstructor_args(self):
obj = REX((print, (1, 2, UNPICKLEABLE)))
for proto in protocols:
with self.subTest(proto=proto):
with self.assertRaises(CustomError):
with self.assertRaises(CustomError) as cm:
self.dumps(obj, proto)
self.assertEqual(cm.exception.__notes__, [
'when serializing tuple item 2',
'when serializing test.pickletester.REX reconstructor arguments',
'when serializing test.pickletester.REX object'])
def test_bad_newobj_args(self):
obj = REX((copyreg.__newobj__, ()))
@@ -1686,74 +1748,154 @@ class AbstractPicklingErrorTests:
with self.subTest(proto=proto):
with self.assertRaises((IndexError, pickle.PicklingError)) as cm:
self.dumps(obj, proto)
self.assertIn(str(cm.exception), {
'tuple index out of range',
'__newobj__ expected at least 1 argument, got 0'})
self.assertEqual(cm.exception.__notes__, [
'when serializing test.pickletester.REX object'])
obj = REX((copyreg.__newobj__, [REX]))
for proto in protocols[2:]:
with self.subTest(proto=proto):
with self.assertRaises((IndexError, pickle.PicklingError)):
with self.assertRaises(pickle.PicklingError) as cm:
self.dumps(obj, proto)
self.assertEqual(str(cm.exception),
'second item of the tuple returned by __reduce__ '
'must be a tuple, not list')
self.assertEqual(cm.exception.__notes__, [
'when serializing test.pickletester.REX object'])
def test_bad_newobj_class(self):
obj = REX((copyreg.__newobj__, (NoNew(),)))
for proto in protocols[2:]:
with self.subTest(proto=proto):
with self.assertRaises(pickle.PicklingError):
with self.assertRaises(pickle.PicklingError) as cm:
self.dumps(obj, proto)
self.assertIn(str(cm.exception), {
'first argument to __newobj__() has no __new__',
f'first argument to __newobj__() must be a class, not {__name__}.NoNew'})
self.assertEqual(cm.exception.__notes__, [
'when serializing test.pickletester.REX object'])
def test_wrong_newobj_class(self):
obj = REX((copyreg.__newobj__, (str,)))
for proto in protocols[2:]:
with self.subTest(proto=proto):
with self.assertRaises(pickle.PicklingError):
with self.assertRaises(pickle.PicklingError) as cm:
self.dumps(obj, proto)
self.assertEqual(str(cm.exception),
f'first argument to __newobj__() must be {REX!r}, not {str!r}')
self.assertEqual(cm.exception.__notes__, [
'when serializing test.pickletester.REX object'])
def test_unpickleable_newobj_class(self):
class LocalREX(REX): pass
obj = LocalREX((copyreg.__newobj__, (LocalREX,)))
for proto in protocols:
with self.subTest(proto=proto):
with self.assertRaises((pickle.PicklingError, AttributeError)):
with self.assertRaises(pickle.PicklingError) as cm:
self.dumps(obj, proto)
if proto >= 2:
self.assertEqual(cm.exception.__notes__, [
f'when serializing {LocalREX.__module__}.{LocalREX.__qualname__} class',
f'when serializing {LocalREX.__module__}.{LocalREX.__qualname__} object'])
else:
self.assertEqual(cm.exception.__notes__, [
'when serializing tuple item 0',
f'when serializing {LocalREX.__module__}.{LocalREX.__qualname__} reconstructor arguments',
f'when serializing {LocalREX.__module__}.{LocalREX.__qualname__} object'])
def test_unpickleable_newobj_args(self):
obj = REX((copyreg.__newobj__, (REX, 1, 2, UNPICKLEABLE)))
for proto in protocols:
with self.subTest(proto=proto):
with self.assertRaises(CustomError):
with self.assertRaises(CustomError) as cm:
self.dumps(obj, proto)
if proto >= 2:
self.assertEqual(cm.exception.__notes__, [
'when serializing tuple item 2',
'when serializing test.pickletester.REX __new__ arguments',
'when serializing test.pickletester.REX object'])
else:
self.assertEqual(cm.exception.__notes__, [
'when serializing tuple item 3',
'when serializing test.pickletester.REX reconstructor arguments',
'when serializing test.pickletester.REX object'])
def test_bad_newobj_ex_args(self):
obj = REX((copyreg.__newobj_ex__, ()))
for proto in protocols[2:]:
with self.subTest(proto=proto):
with self.assertRaises((ValueError, pickle.PicklingError)):
with self.assertRaises((ValueError, pickle.PicklingError)) as cm:
self.dumps(obj, proto)
self.assertIn(str(cm.exception), {
'not enough values to unpack (expected 3, got 0)',
'__newobj_ex__ expected 3 arguments, got 0'})
self.assertEqual(cm.exception.__notes__, [
'when serializing test.pickletester.REX object'])
obj = REX((copyreg.__newobj_ex__, 42))
for proto in protocols[2:]:
with self.subTest(proto=proto):
with self.assertRaises(pickle.PicklingError):
with self.assertRaises(pickle.PicklingError) as cm:
self.dumps(obj, proto)
self.assertEqual(str(cm.exception),
'second item of the tuple returned by __reduce__ '
'must be a tuple, not int')
self.assertEqual(cm.exception.__notes__, [
'when serializing test.pickletester.REX object'])
obj = REX((copyreg.__newobj_ex__, (REX, 42, {})))
is_py = self.pickler is pickle._Pickler
for proto in protocols[2:4] if is_py else protocols[2:]:
with self.subTest(proto=proto):
with self.assertRaises((TypeError, pickle.PicklingError)):
self.dumps(obj, proto)
if self.pickler is pickle._Pickler:
for proto in protocols[2:4]:
with self.subTest(proto=proto):
with self.assertRaises(TypeError) as cm:
self.dumps(obj, proto)
self.assertEqual(str(cm.exception),
'Value after * must be an iterable, not int')
self.assertEqual(cm.exception.__notes__, [
'when serializing test.pickletester.REX object'])
else:
for proto in protocols[2:]:
with self.subTest(proto=proto):
with self.assertRaises(pickle.PicklingError) as cm:
self.dumps(obj, proto)
self.assertEqual(str(cm.exception),
'second argument to __newobj_ex__() must be a tuple, not int')
self.assertEqual(cm.exception.__notes__, [
'when serializing test.pickletester.REX object'])
obj = REX((copyreg.__newobj_ex__, (REX, (), [])))
for proto in protocols[2:4] if is_py else protocols[2:]:
with self.subTest(proto=proto):
with self.assertRaises((TypeError, pickle.PicklingError)):
self.dumps(obj, proto)
if self.pickler is pickle._Pickler:
for proto in protocols[2:4]:
with self.subTest(proto=proto):
with self.assertRaises(TypeError) as cm:
self.dumps(obj, proto)
self.assertEqual(str(cm.exception),
'functools.partial() argument after ** must be a mapping, not list')
self.assertEqual(cm.exception.__notes__, [
'when serializing test.pickletester.REX object'])
else:
for proto in protocols[2:]:
with self.subTest(proto=proto):
with self.assertRaises(pickle.PicklingError) as cm:
self.dumps(obj, proto)
self.assertEqual(str(cm.exception),
'third argument to __newobj_ex__() must be a dict, not list')
self.assertEqual(cm.exception.__notes__, [
'when serializing test.pickletester.REX object'])
def test_bad_newobj_ex__class(self):
obj = REX((copyreg.__newobj_ex__, (NoNew(), (), {})))
for proto in protocols[2:]:
with self.subTest(proto=proto):
with self.assertRaises(pickle.PicklingError):
with self.assertRaises(pickle.PicklingError) as cm:
self.dumps(obj, proto)
self.assertIn(str(cm.exception), {
'first argument to __newobj_ex__() has no __new__',
f'first argument to __newobj_ex__() must be a class, not {__name__}.NoNew'})
self.assertEqual(cm.exception.__notes__, [
'when serializing test.pickletester.REX object'])
def test_wrong_newobj_ex_class(self):
if self.pickler is not pickle._Pickler:
@@ -1761,37 +1903,99 @@ class AbstractPicklingErrorTests:
obj = REX((copyreg.__newobj_ex__, (str, (), {})))
for proto in protocols[2:]:
with self.subTest(proto=proto):
with self.assertRaises(pickle.PicklingError):
with self.assertRaises(pickle.PicklingError) as cm:
self.dumps(obj, proto)
self.assertEqual(str(cm.exception),
f'first argument to __newobj_ex__() must be {REX}, not {str}')
self.assertEqual(cm.exception.__notes__, [
'when serializing test.pickletester.REX object'])
def test_unpickleable_newobj_ex_class(self):
class LocalREX(REX): pass
obj = LocalREX((copyreg.__newobj_ex__, (LocalREX, (), {})))
for proto in protocols:
with self.subTest(proto=proto):
with self.assertRaises((pickle.PicklingError, AttributeError)):
with self.assertRaises(pickle.PicklingError) as cm:
self.dumps(obj, proto)
if proto >= 4:
self.assertEqual(cm.exception.__notes__, [
f'when serializing {LocalREX.__module__}.{LocalREX.__qualname__} class',
f'when serializing {LocalREX.__module__}.{LocalREX.__qualname__} object'])
elif proto >= 2:
self.assertEqual(cm.exception.__notes__, [
'when serializing tuple item 0',
'when serializing tuple item 1',
'when serializing functools.partial state',
'when serializing functools.partial object',
f'when serializing {LocalREX.__module__}.{LocalREX.__qualname__} reconstructor',
f'when serializing {LocalREX.__module__}.{LocalREX.__qualname__} object'])
else:
self.assertEqual(cm.exception.__notes__, [
'when serializing tuple item 0',
f'when serializing {LocalREX.__module__}.{LocalREX.__qualname__} reconstructor arguments',
f'when serializing {LocalREX.__module__}.{LocalREX.__qualname__} object'])
def test_unpickleable_newobj_ex_args(self):
obj = REX((copyreg.__newobj_ex__, (REX, (1, 2, UNPICKLEABLE), {})))
for proto in protocols:
with self.subTest(proto=proto):
with self.assertRaises(CustomError):
with self.assertRaises(CustomError) as cm:
self.dumps(obj, proto)
if proto >= 4:
self.assertEqual(cm.exception.__notes__, [
'when serializing tuple item 2',
'when serializing test.pickletester.REX __new__ arguments',
'when serializing test.pickletester.REX object'])
elif proto >= 2:
self.assertEqual(cm.exception.__notes__, [
'when serializing tuple item 3',
'when serializing tuple item 1',
'when serializing functools.partial state',
'when serializing functools.partial object',
'when serializing test.pickletester.REX reconstructor',
'when serializing test.pickletester.REX object'])
else:
self.assertEqual(cm.exception.__notes__, [
'when serializing tuple item 2',
'when serializing tuple item 1',
'when serializing test.pickletester.REX reconstructor arguments',
'when serializing test.pickletester.REX object'])
def test_unpickleable_newobj_ex_kwargs(self):
obj = REX((copyreg.__newobj_ex__, (REX, (), {'a': UNPICKLEABLE})))
for proto in protocols:
with self.subTest(proto=proto):
with self.assertRaises(CustomError):
with self.assertRaises(CustomError) as cm:
self.dumps(obj, proto)
if proto >= 4:
self.assertEqual(cm.exception.__notes__, [
"when serializing dict item 'a'",
'when serializing test.pickletester.REX __new__ arguments',
'when serializing test.pickletester.REX object'])
elif proto >= 2:
self.assertEqual(cm.exception.__notes__, [
"when serializing dict item 'a'",
'when serializing tuple item 2',
'when serializing functools.partial state',
'when serializing functools.partial object',
'when serializing test.pickletester.REX reconstructor',
'when serializing test.pickletester.REX object'])
else:
self.assertEqual(cm.exception.__notes__, [
"when serializing dict item 'a'",
'when serializing tuple item 2',
'when serializing test.pickletester.REX reconstructor arguments',
'when serializing test.pickletester.REX object'])
def test_unpickleable_state(self):
obj = REX_state(UNPICKLEABLE)
for proto in protocols:
with self.subTest(proto=proto):
with self.assertRaises(CustomError):
with self.assertRaises(CustomError) as cm:
self.dumps(obj, proto)
self.assertEqual(cm.exception.__notes__, [
'when serializing test.pickletester.REX_state state',
'when serializing test.pickletester.REX_state object'])
def test_bad_state_setter(self):
if self.pickler is pickle._Pickler:
@@ -1799,22 +2003,33 @@ class AbstractPicklingErrorTests:
obj = REX((print, (), 'state', None, None, 42))
for proto in protocols:
with self.subTest(proto=proto):
with self.assertRaises(pickle.PicklingError):
with self.assertRaises(pickle.PicklingError) as cm:
self.dumps(obj, proto)
self.assertEqual(str(cm.exception),
'sixth item of the tuple returned by __reduce__ '
'must be callable, not int')
self.assertEqual(cm.exception.__notes__, [
'when serializing test.pickletester.REX object'])
def test_unpickleable_state_setter(self):
obj = REX((print, (), 'state', None, None, UnpickleableCallable()))
for proto in protocols:
with self.subTest(proto=proto):
with self.assertRaises(CustomError):
with self.assertRaises(CustomError) as cm:
self.dumps(obj, proto)
self.assertEqual(cm.exception.__notes__, [
'when serializing test.pickletester.REX state setter',
'when serializing test.pickletester.REX object'])
def test_unpickleable_state_with_state_setter(self):
obj = REX((print, (), UNPICKLEABLE, None, None, print))
for proto in protocols:
with self.subTest(proto=proto):
with self.assertRaises(CustomError):
with self.assertRaises(CustomError) as cm:
self.dumps(obj, proto)
self.assertEqual(cm.exception.__notes__, [
'when serializing test.pickletester.REX state',
'when serializing test.pickletester.REX object'])
def test_bad_object_list_items(self):
# Issue4176: crash when 4th and 5th items of __reduce__()
@@ -1822,23 +2037,37 @@ class AbstractPicklingErrorTests:
obj = REX((list, (), None, 42))
for proto in protocols:
with self.subTest(proto=proto):
with self.assertRaises((TypeError, pickle.PicklingError)):
with self.assertRaises((TypeError, pickle.PicklingError)) as cm:
self.dumps(obj, proto)
self.assertIn(str(cm.exception), {
"'int' object is not iterable",
'fourth item of the tuple returned by __reduce__ '
'must be an iterator, not int'})
self.assertEqual(cm.exception.__notes__, [
'when serializing test.pickletester.REX object'])
if self.pickler is not pickle._Pickler:
# Python implementation is less strict and also accepts iterables.
obj = REX((list, (), None, []))
for proto in protocols:
with self.subTest(proto=proto):
with self.assertRaises((TypeError, pickle.PicklingError)):
with self.assertRaises(pickle.PicklingError):
self.dumps(obj, proto)
self.assertEqual(str(cm.exception),
'fourth item of the tuple returned by __reduce__ '
'must be an iterator, not int')
self.assertEqual(cm.exception.__notes__, [
'when serializing test.pickletester.REX object'])
def test_unpickleable_object_list_items(self):
obj = REX_six([1, 2, UNPICKLEABLE])
for proto in protocols:
with self.subTest(proto=proto):
with self.assertRaises(CustomError):
with self.assertRaises(CustomError) as cm:
self.dumps(obj, proto)
self.assertEqual(cm.exception.__notes__, [
'when serializing test.pickletester.REX_six item 2',
'when serializing test.pickletester.REX_six object'])
def test_bad_object_dict_items(self):
# Issue4176: crash when 4th and 5th items of __reduce__()
@@ -1846,82 +2075,135 @@ class AbstractPicklingErrorTests:
obj = REX((dict, (), None, None, 42))
for proto in protocols:
with self.subTest(proto=proto):
with self.assertRaises((TypeError, pickle.PicklingError)):
with self.assertRaises((TypeError, pickle.PicklingError)) as cm:
self.dumps(obj, proto)
self.assertIn(str(cm.exception), {
"'int' object is not iterable",
'fifth item of the tuple returned by __reduce__ '
'must be an iterator, not int'})
self.assertEqual(cm.exception.__notes__, [
'when serializing test.pickletester.REX object'])
for proto in protocols:
obj = REX((dict, (), None, None, iter([('a',)])))
with self.subTest(proto=proto):
with self.assertRaises((ValueError, TypeError)):
with self.assertRaises((ValueError, TypeError)) as cm:
self.dumps(obj, proto)
self.assertIn(str(cm.exception), {
'not enough values to unpack (expected 2, got 1)',
'dict items iterator must return 2-tuples'})
self.assertEqual(cm.exception.__notes__, [
'when serializing test.pickletester.REX object'])
if self.pickler is not pickle._Pickler:
# Python implementation is less strict and also accepts iterables.
obj = REX((dict, (), None, None, []))
for proto in protocols:
with self.subTest(proto=proto):
with self.assertRaises((TypeError, pickle.PicklingError)):
with self.assertRaises(pickle.PicklingError):
self.dumps(obj, proto)
self.assertEqual(str(cm.exception),
'dict items iterator must return 2-tuples')
self.assertEqual(cm.exception.__notes__, [
'when serializing test.pickletester.REX object'])
def test_unpickleable_object_dict_items(self):
obj = REX_seven({'a': UNPICKLEABLE})
for proto in protocols:
with self.subTest(proto=proto):
with self.assertRaises(CustomError):
with self.assertRaises(CustomError) as cm:
self.dumps(obj, proto)
self.assertEqual(cm.exception.__notes__, [
"when serializing test.pickletester.REX_seven item 'a'",
'when serializing test.pickletester.REX_seven object'])
def test_unpickleable_list_items(self):
obj = [1, [2, 3, UNPICKLEABLE]]
for proto in protocols:
with self.subTest(proto=proto):
with self.assertRaises(CustomError):
with self.assertRaises(CustomError) as cm:
self.dumps(obj, proto)
self.assertEqual(cm.exception.__notes__, [
'when serializing list item 2',
'when serializing list item 1'])
for n in [0, 1, 1000, 1005]:
obj = [*range(n), UNPICKLEABLE]
for proto in protocols:
with self.subTest(proto=proto):
with self.assertRaises(CustomError):
with self.assertRaises(CustomError) as cm:
self.dumps(obj, proto)
self.assertEqual(cm.exception.__notes__, [
f'when serializing list item {n}'])
def test_unpickleable_tuple_items(self):
obj = (1, (2, 3, UNPICKLEABLE))
for proto in protocols:
with self.subTest(proto=proto):
with self.assertRaises(CustomError):
with self.assertRaises(CustomError) as cm:
self.dumps(obj, proto)
self.assertEqual(cm.exception.__notes__, [
'when serializing tuple item 2',
'when serializing tuple item 1'])
obj = (*range(10), UNPICKLEABLE)
for proto in protocols:
with self.subTest(proto=proto):
with self.assertRaises(CustomError):
with self.assertRaises(CustomError) as cm:
self.dumps(obj, proto)
self.assertEqual(cm.exception.__notes__, [
'when serializing tuple item 10'])
def test_unpickleable_dict_items(self):
obj = {'a': {'b': UNPICKLEABLE}}
for proto in protocols:
with self.subTest(proto=proto):
with self.assertRaises(CustomError):
with self.assertRaises(CustomError) as cm:
self.dumps(obj, proto)
self.assertEqual(cm.exception.__notes__, [
"when serializing dict item 'b'",
"when serializing dict item 'a'"])
for n in [0, 1, 1000, 1005]:
obj = dict.fromkeys(range(n))
obj['a'] = UNPICKLEABLE
for proto in protocols:
with self.subTest(proto=proto, n=n):
with self.assertRaises(CustomError):
with self.assertRaises(CustomError) as cm:
self.dumps(obj, proto)
self.assertEqual(cm.exception.__notes__, [
"when serializing dict item 'a'"])
def test_unpickleable_set_items(self):
obj = {UNPICKLEABLE}
for proto in protocols:
with self.subTest(proto=proto):
with self.assertRaises(CustomError):
with self.assertRaises(CustomError) as cm:
self.dumps(obj, proto)
if proto >= 4:
self.assertEqual(cm.exception.__notes__, [
'when serializing set element'])
else:
self.assertEqual(cm.exception.__notes__, [
'when serializing list item 0',
'when serializing tuple item 0',
'when serializing set reconstructor arguments'])
def test_unpickleable_frozenset_items(self):
obj = frozenset({frozenset({UNPICKLEABLE})})
for proto in protocols:
with self.subTest(proto=proto):
with self.assertRaises(CustomError):
with self.assertRaises(CustomError) as cm:
self.dumps(obj, proto)
if proto >= 4:
self.assertEqual(cm.exception.__notes__, [
'when serializing frozenset element',
'when serializing frozenset element'])
else:
self.assertEqual(cm.exception.__notes__, [
'when serializing list item 0',
'when serializing tuple item 0',
'when serializing frozenset reconstructor arguments',
'when serializing list item 0',
'when serializing tuple item 0',
'when serializing frozenset reconstructor arguments'])
def test_global_lookup_error(self):
# Global name does not exist
@@ -1929,26 +2211,42 @@ class AbstractPicklingErrorTests:
obj.__module__ = __name__
for proto in protocols:
with self.subTest(proto=proto):
with self.assertRaises(pickle.PicklingError):
with self.assertRaises(pickle.PicklingError) as cm:
self.dumps(obj, proto)
self.assertEqual(str(cm.exception),
f"Can't pickle {obj!r}: it's not found as {__name__}.spam")
self.assertEqual(str(cm.exception.__context__),
f"module '{__name__}' has no attribute 'spam'")
obj.__module__ = 'nonexisting'
for proto in protocols:
with self.subTest(proto=proto):
with self.assertRaises(pickle.PicklingError):
with self.assertRaises(pickle.PicklingError) as cm:
self.dumps(obj, proto)
self.assertEqual(str(cm.exception),
f"Can't pickle {obj!r}: No module named 'nonexisting'")
self.assertEqual(str(cm.exception.__context__),
"No module named 'nonexisting'")
obj.__module__ = ''
for proto in protocols:
with self.subTest(proto=proto):
with self.assertRaises((ValueError, pickle.PicklingError)):
with self.assertRaises(pickle.PicklingError) as cm:
self.dumps(obj, proto)
self.assertEqual(str(cm.exception),
f"Can't pickle {obj!r}: Empty module name")
self.assertEqual(str(cm.exception.__context__),
"Empty module name")
obj.__module__ = None
for proto in protocols:
with self.subTest(proto=proto):
with self.assertRaises(pickle.PicklingError):
with self.assertRaises(pickle.PicklingError) as cm:
self.dumps(obj, proto)
self.assertEqual(str(cm.exception),
f"Can't pickle {obj!r}: it's not found as __main__.spam")
self.assertEqual(str(cm.exception.__context__),
"module '__main__' has no attribute 'spam'")
def test_nonencodable_global_name_error(self):
for proto in protocols[:4]:
@@ -1957,8 +2255,11 @@ class AbstractPicklingErrorTests:
obj = REX(name)
obj.__module__ = __name__
with support.swap_item(globals(), name, obj):
with self.assertRaises((UnicodeEncodeError, pickle.PicklingError)):
with self.assertRaises(pickle.PicklingError) as cm:
self.dumps(obj, proto)
self.assertEqual(str(cm.exception),
f"can't pickle global identifier {name!r} using pickle protocol {proto}")
self.assertIsInstance(cm.exception.__context__, UnicodeEncodeError)
def test_nonencodable_module_name_error(self):
for proto in protocols[:4]:
@@ -1968,8 +2269,11 @@ class AbstractPicklingErrorTests:
obj.__module__ = name
mod = types.SimpleNamespace(test=obj)
with support.swap_item(sys.modules, name, mod):
with self.assertRaises((UnicodeEncodeError, pickle.PicklingError)):
with self.assertRaises(pickle.PicklingError) as cm:
self.dumps(obj, proto)
self.assertEqual(str(cm.exception),
f"can't pickle module identifier {name!r} using pickle protocol {proto}")
self.assertIsInstance(cm.exception.__context__, UnicodeEncodeError)
def test_nested_lookup_error(self):
# Nested name does not exist
@@ -1981,14 +2285,24 @@ class AbstractPicklingErrorTests:
obj.__module__ = __name__
for proto in protocols:
with self.subTest(proto=proto):
with self.assertRaises(pickle.PicklingError):
with self.assertRaises(pickle.PicklingError) as cm:
self.dumps(obj, proto)
self.assertEqual(str(cm.exception),
f"Can't pickle {obj!r}: "
f"it's not found as {__name__}.TestGlobal.A.B.C")
self.assertEqual(str(cm.exception.__context__),
"type object 'A' has no attribute 'B'")
obj.__module__ = None
for proto in protocols:
with self.subTest(proto=proto):
with self.assertRaises(pickle.PicklingError):
with self.assertRaises(pickle.PicklingError) as cm:
self.dumps(obj, proto)
self.assertEqual(str(cm.exception),
f"Can't pickle {obj!r}: "
f"it's not found as __main__.TestGlobal.A.B.C")
self.assertEqual(str(cm.exception.__context__),
"module '__main__' has no attribute 'TestGlobal'")
def test_wrong_object_lookup_error(self):
# Name is bound to different object
@@ -1999,14 +2313,23 @@ class AbstractPicklingErrorTests:
obj.__module__ = __name__
for proto in protocols:
with self.subTest(proto=proto):
with self.assertRaises(pickle.PicklingError):
with self.assertRaises(pickle.PicklingError) as cm:
self.dumps(obj, proto)
self.assertEqual(str(cm.exception),
f"Can't pickle {obj!r}: "
f"it's not the same object as {__name__}.TestGlobal")
self.assertIsNone(cm.exception.__context__)
obj.__module__ = None
for proto in protocols:
with self.subTest(proto=proto):
with self.assertRaises(pickle.PicklingError):
with self.assertRaises(pickle.PicklingError) as cm:
self.dumps(obj, proto)
self.assertEqual(str(cm.exception),
f"Can't pickle {obj!r}: "
f"it's not found as __main__.TestGlobal")
self.assertEqual(str(cm.exception.__context__),
"module '__main__' has no attribute 'TestGlobal'")
def test_local_lookup_error(self):
# Test that whichmodule() errors out cleanly when looking up
@@ -2016,21 +2339,27 @@ class AbstractPicklingErrorTests:
# Since the function is local, lookup will fail
for proto in protocols:
with self.subTest(proto=proto):
with self.assertRaises((AttributeError, pickle.PicklingError)):
with self.assertRaises(pickle.PicklingError) as cm:
self.dumps(f, proto)
self.assertEqual(str(cm.exception),
f"Can't pickle local object {f!r}")
# Same without a __module__ attribute (exercises a different path
# in _pickle.c).
del f.__module__
for proto in protocols:
with self.subTest(proto=proto):
with self.assertRaises((AttributeError, pickle.PicklingError)):
with self.assertRaises(pickle.PicklingError) as cm:
self.dumps(f, proto)
self.assertEqual(str(cm.exception),
f"Can't pickle local object {f!r}")
# Yet a different path.
f.__name__ = f.__qualname__
for proto in protocols:
with self.subTest(proto=proto):
with self.assertRaises((AttributeError, pickle.PicklingError)):
with self.assertRaises(pickle.PicklingError) as cm:
self.dumps(f, proto)
self.assertEqual(str(cm.exception),
f"Can't pickle local object {f!r}")
def test_reduce_ex_None(self):
c = REX_None()
@@ -2744,7 +3073,7 @@ class AbstractPickleTests:
pickled = self.dumps(None, proto)
if proto >= 2:
proto_header = pickle.PROTO + bytes([proto])
self.assertTrue(pickled.startswith(proto_header))
self.assertStartsWith(pickled, proto_header)
else:
self.assertEqual(count_opcode(pickle.PROTO, pickled), 0)
@@ -4640,8 +4969,11 @@ class AbstractHookTests:
# NotImplemented
self.assertIs(math_log, math.log)
with self.assertRaises(pickle.PicklingError):
with self.assertRaises(pickle.PicklingError) as cm:
p.dump(g)
self.assertRegex(str(cm.exception),
r'(__reduce__|<bound method .*reducer_override.*>)'
r' must return (a )?string or tuple')
with self.assertRaisesRegex(
ValueError, 'The reducer just failed'):
@@ -4680,7 +5012,7 @@ class AbstractDispatchTableTests:
p = self.pickler_class(f, 0)
with self.assertRaises(AttributeError):
p.dispatch_table
self.assertFalse(hasattr(p, 'dispatch_table'))
self.assertNotHasAttr(p, 'dispatch_table')
def test_class_dispatch_table(self):
# A dispatch_table attribute can be specified class-wide