def test_negative_32b_binbytes(self):
# On 32-bit builds, a BINBYTES of 2**31 or more is refused
dumped = b'\x80\x03B\xff\xff\xff\xffxyzq\x00.'
self.check_unpickling_error((pickle.UnpicklingError, OverflowError),
dumped)
python类UnpicklingError()的实例源码
def test_negative_32b_binunicode(self):
# On 32-bit builds, a BINUNICODE of 2**31 or more is refused
dumped = b'\x80\x03X\xff\xff\xff\xffxyzq\x00.'
self.check_unpickling_error((pickle.UnpicklingError, OverflowError),
dumped)
def test_large_32b_binbytes8(self):
dumped = b'\x80\x04\x8e\4\0\0\0\1\0\0\0\xe2\x82\xac\x00.'
self.check_unpickling_error((pickle.UnpicklingError, OverflowError),
dumped)
def test_large_32b_binunicode8(self):
dumped = b'\x80\x04\x8d\4\0\0\0\1\0\0\0\xe2\x82\xac\x00.'
self.check_unpickling_error((pickle.UnpicklingError, OverflowError),
dumped)
def test_bad_init(self):
# Test issue3664 (pickle can segfault from a badly initialized Pickler).
# Override initialization without calling __init__() of the superclass.
class BadPickler(pickle.Pickler):
def __init__(self): pass
class BadUnpickler(pickle.Unpickler):
def __init__(self): pass
self.assertRaises(pickle.PicklingError, BadPickler().dump, 0)
self.assertRaises(pickle.UnpicklingError, BadUnpickler().load)
def pollmessage(self, wait):
packet = self.pollpacket(wait)
if packet is None:
return None
try:
message = pickle.loads(packet)
except pickle.UnpicklingError:
print("-----------------------", file=sys.__stderr__)
print("cannot unpickle packet:", repr(packet), file=sys.__stderr__)
traceback.print_stack(file=sys.__stderr__)
print("-----------------------", file=sys.__stderr__)
raise
return message
def _load_binstring(self):
len, = unpack('<i', self.read(4))
if len < 0:
raise pickle.UnpicklingError("BINSTRING pickle has negative byte count")
data = self.read(len)
try:
data = str(data, self.encoding, self.errors)
except:
pass
self.append(data)
def _GetValueAndType(self, key):
"""Fetch value from memcache and detect its type.
Args:
key: String
Returns:
(value, type), value is a Python object or None if the key was not set in
the cache, type is a string describing the type of the value.
"""
try:
value = memcache.get(key)
except (pickle.UnpicklingError, AttributeError, EOFError, ImportError,
IndexError), e:
msg = 'Failed to retrieve value from cache: %s' % e
return msg, 'error'
if value is None:
return None, self.DEFAULT_TYPESTR_FOR_NEW
for typeobj, _, typestr in self.TYPES:
if isinstance(value, typeobj):
break
else:
typestr = 'pickled'
value = pprint.pformat(value, indent=2)
return value, typestr
def persistent_load(self, pid):
type_tag = pid
if type_tag == marker:
return self.w
else:
raise pickle.UnpicklingError("unsupported persistent object")
def test_pop_empty_stack(self):
# Test issue7455
s = b'0'
self.assertRaises((pickle.UnpicklingError, IndexError), self.loads, s)
def check_negative_32b_binXXX(self, dumped):
if sys.maxsize > 2**32:
self.skipTest("test is only meaningful on 32-bit builds")
# XXX Pure Python pickle reads lengths as signed and passes
# them directly to read() (hence the EOFError)
with self.assertRaises((pickle.UnpicklingError, EOFError,
ValueError, OverflowError)):
self.loads(dumped)
def test_bad_init(self):
# Test issue3664 (pickle can segfault from a badly initialized Pickler).
# Override initialization without calling __init__() of the superclass.
class BadPickler(pickle.Pickler):
def __init__(self): pass
class BadUnpickler(pickle.Unpickler):
def __init__(self): pass
self.assertRaises(pickle.PicklingError, BadPickler().dump, 0)
self.assertRaises(pickle.UnpicklingError, BadUnpickler().load)
def pollmessage(self, wait):
packet = self.pollpacket(wait)
if packet is None:
return None
try:
message = pickle.loads(packet)
except pickle.UnpicklingError:
print("-----------------------", file=sys.__stderr__)
print("cannot unpickle packet:", repr(packet), file=sys.__stderr__)
traceback.print_stack(file=sys.__stderr__)
print("-----------------------", file=sys.__stderr__)
raise
return message
def __get__(self, obj, typ=None):
if self.name not in FileDescr.saved:
raise AttributeError, \
"%r used before assignment" % self.name
try:
f = open(self.name, 'r')
val = pickle.load(f)
f.close()
return val
except (pickle.UnpicklingError, IOError,
EOFError, AttributeError,
ImportError, IndexError), e:
raise AttributeError, \
"could not read %r: %s" % self.name
def __get__(self, obj, typ=None):
if self.name not in FileDescr.saved:
raise AttributeError, \
"%r used before assignment" % self.name
try:
f = open(self.name, 'r')
val = pickle.load(f)
f.close()
return val
except (pickle.UnpicklingError, IOError,
EOFError, AttributeError,
ImportError, IndexError), e:
raise AttributeError, \
"could not read %r: %s" % self.name
def __get__(self, obj, typ=None):
if self.name not in FileDescr.saved:
raise AttributeError, \
"%r used before assignment" % self.name
try:
f = open(self.name, 'r')
val = pickle.load(f)
f.close()
return val
except (pickle.UnpicklingError, IOError,
EOFError, AttributeError,
ImportError, IndexError), e:
raise AttributeError, \
"could not read %r: %s" % self.name
def __get__(self, obj, typ=None):
if self.name not in FileDescr.saved:
raise AttributeError, \
"%r used before assignment" % self.name
try:
f = open(self.name, 'r')
val = pickle.load(f)
f.close()
return val
except (pickle.UnpicklingError, IOError,
EOFError, AttributeError,
ImportError, IndexError), e:
raise AttributeError, \
"could not read %r: %s" % self.name
def read_device_messages(self):
while self.is_receiving or self.is_transmitting:
try:
message = self.parent_ctrl_conn.recv()
try:
action, return_code = message.split(":")
self.log_retcode(int(return_code), action)
except ValueError:
self.device_messages.append("{0}: {1}".format(self.__class__.__name__, message))
except (EOFError, UnpicklingError, OSError, ConnectionResetError) as e:
logger.info("Exiting read device message thread due to " + str(e))
break
self.is_transmitting = False
self.is_receiving = False
logger.debug("Exiting read device errors thread")
def load_tokc(self):
try:
with open(self.base_fname + '.tokc', mode='rb') as f:
self.tokc = pickle.load(f)
if not self.tokc:
raise IndexLoadError
except (IOError, pickle.UnpicklingError):
raise IndexLoadError
def findGlobal(self, module, klass):
"""Find class name."""
if (module, klass) not in self.allowedGlobals():
raise UnpicklingError("For security reasons, you can\'t unpickle"
" objects from module %s with type %s." % (module, klass))
g = {}
exec 'from %s import %s as theClass' % (module, klass) in g
return g['theClass']
