def test_reduce_bad_iterator(self):
# Issue4176: crash when 4th and 5th items of __reduce__()
# are not iterators
class C(object):
def __reduce__(self):
# 4th item is not an iterator
return list, (), None, [], None
class D(object):
def __reduce__(self):
# 5th item is not an iterator
return dict, (), None, None, []
# Protocol 0 in Python implementation is less strict and also accepts
# iterables.
for proto in protocols:
try:
self.dumps(C(), proto)
except (AttributeError, pickle.PicklingError, cPickle.PicklingError):
pass
try:
self.dumps(D(), proto)
except (AttributeError, pickle.PicklingError, cPickle.PicklingError):
pass
python类PicklingError()的实例源码
def test_reduce_bad_iterator(self):
# Issue4176: crash when 4th and 5th items of __reduce__()
# are not iterators
class C(object):
def __reduce__(self):
# 4th item is not an iterator
return list, (), None, [], None
class D(object):
def __reduce__(self):
# 5th item is not an iterator
return dict, (), None, None, []
# Protocol 0 in Python implementation is less strict and also accepts
# iterables.
for proto in protocols:
try:
self.dumps(C(), proto)
except (AttributeError, pickle.PicklingError, cPickle.PicklingError):
pass
try:
self.dumps(D(), proto)
except (AttributeError, pickle.PicklingError, cPickle.PicklingError):
pass
def putmessage(self, message):
self.debug("putmessage:%d:" % message[0])
try:
s = pickle.dumps(message)
except pickle.PicklingError:
print >>sys.__stderr__, "Cannot pickle:", repr(message)
raise
s = struct.pack("<i", len(s)) + s
while len(s) > 0:
try:
r, w, x = select.select([], [self.sock], [])
n = self.sock.send(s[:BUFSIZE])
except (AttributeError, TypeError):
raise IOError, "socket no longer exists"
except socket.error:
raise
else:
s = s[n:]
def put(self, o):
"""Encode object ``o`` and write it to the pipe.
Block gevent-cooperatively until all data is written. The default
encoder is ``pickle.dumps``.
:arg o: a Python object that is encodable with the encoder of choice.
Raises:
- :exc:`GIPCError`
- :exc:`GIPCClosed`
- :exc:`pickle.PicklingError`
"""
self._validate()
with self._lock:
bindata = self._encoder(o)
self._write(struct.pack("!i", len(bindata)) + bindata)
def dump(self, value, f):
try:
pickle.dump(value, f, 2)
except pickle.PickleError:
raise
except Exception as e:
msg = "Could not serialize broadcast: " + e.__class__.__name__ + ": " + e.message
print_exec(sys.stderr)
raise pickle.PicklingError(msg)
f.close()
return f.name
def pickled(cls, obj, **kwargs):
self = cls(**kwargs)
try:
self.data = cPickle.dumps(obj, protocol=2)
except cPickle.PicklingError, e:
self.data = cPickle.dumps(CallError(e), protocol=2)
return self
def putmessage(self, message):
self.debug("putmessage:%d:" % message[0])
try:
s = pickle.dumps(message)
except pickle.PicklingError:
print >>sys.__stderr__, "Cannot pickle:", repr(message)
raise
s = struct.pack("<i", len(s)) + s
while len(s) > 0:
try:
r, w, x = select.select([], [self.sock], [])
n = self.sock.send(s[:BUFSIZE])
except (AttributeError, TypeError):
raise IOError, "socket no longer exists"
s = s[n:]
def putmessage(self, message):
self.debug("putmessage:%d:" % message[0])
try:
s = pickle.dumps(message)
except pickle.PicklingError:
print >>sys.__stderr__, "Cannot pickle:", repr(message)
raise
s = struct.pack("<i", len(s)) + s
while len(s) > 0:
try:
r, w, x = select.select([], [self.sock], [])
n = self.sock.send(s[:BUFSIZE])
except (AttributeError, TypeError):
raise IOError, "socket no longer exists"
s = s[n:]
def update_saved_parameters(training_stats, test_stack_config, args):
if training_stats['worthy_epoch']:
test_stack_config['best_epoch_id'] = training_stats['best_epoch_id']
test_stack_config['validation_result'] = training_stats['validation_result']
test_stack_config['training_result'] = training_stats['training_result']
try:
lstm_seqlabel_load_save_model.save_parameters_to_file(
test_stack_config, args.saveto)
except pickle.PicklingError as e:
print "Suffering from Pickling Error in saving \n%s \nto %s"%(
' '.join(test_stack_config.keys()),
args.saveto)
return
def test_handledByPickleModule(self):
"""
Handling L{pickle.PicklingError} handles
L{_UniversalPicklingError}.
"""
self.assertRaises(pickle.PicklingError,
self.raise_UniversalPicklingError)
def test_handledBycPickleModule(self):
"""
Handling L{cPickle.PicklingError} handles
L{_UniversalPicklingError}.
"""
try:
import cPickle
except ImportError:
raise unittest.SkipTest("cPickle not available.")
else:
self.assertRaises(cPickle.PicklingError,
self.raise_UniversalPicklingError)
def test_lambdaRaisesPicklingError(self):
"""
Pickling a C{lambda} function ought to raise a L{pickle.PicklingError}.
"""
self.assertRaises(pickle.PicklingError, pickle.dumps, lambdaExample)
try:
import cPickle
except:
pass
else:
self.assertRaises(cPickle.PicklingError, cPickle.dumps,
lambdaExample)
