summaryrefslogtreecommitdiff
path: root/lib/python2.7/test/pickletester.py
diff options
context:
space:
mode:
Diffstat (limited to 'lib/python2.7/test/pickletester.py')
-rw-r--r--lib/python2.7/test/pickletester.py1319
1 files changed, 0 insertions, 1319 deletions
diff --git a/lib/python2.7/test/pickletester.py b/lib/python2.7/test/pickletester.py
deleted file mode 100644
index 34cafcb..0000000
--- a/lib/python2.7/test/pickletester.py
+++ /dev/null
@@ -1,1319 +0,0 @@
-import unittest
-import pickle
-import cPickle
-import StringIO
-import cStringIO
-import pickletools
-import copy_reg
-
-from test.test_support import TestFailed, verbose, have_unicode, TESTFN
-try:
- from test.test_support import _2G, _1M, precisionbigmemtest
-except ImportError:
- # this import might fail when run on older Python versions by test_xpickle
- _2G = _1M = 0
- def precisionbigmemtest(*args, **kwargs):
- return lambda self: None
-
-# Tests that try a number of pickle protocols should have a
-# for proto in protocols:
-# kind of outer loop.
-assert pickle.HIGHEST_PROTOCOL == cPickle.HIGHEST_PROTOCOL == 2
-protocols = range(pickle.HIGHEST_PROTOCOL + 1)
-
-# Copy of test.test_support.run_with_locale. This is needed to support Python
-# 2.4, which didn't include it. This is all to support test_xpickle, which
-# bounces pickled objects through older Python versions to test backwards
-# compatibility.
-def run_with_locale(catstr, *locales):
- def decorator(func):
- def inner(*args, **kwds):
- try:
- import locale
- category = getattr(locale, catstr)
- orig_locale = locale.setlocale(category)
- except AttributeError:
- # if the test author gives us an invalid category string
- raise
- except:
- # cannot retrieve original locale, so do nothing
- locale = orig_locale = None
- else:
- for loc in locales:
- try:
- locale.setlocale(category, loc)
- break
- except:
- pass
-
- # now run the function, resetting the locale on exceptions
- try:
- return func(*args, **kwds)
- finally:
- if locale and orig_locale:
- locale.setlocale(category, orig_locale)
- inner.func_name = func.func_name
- inner.__doc__ = func.__doc__
- return inner
- return decorator
-
-
-# Return True if opcode code appears in the pickle, else False.
-def opcode_in_pickle(code, pickle):
- for op, dummy, dummy in pickletools.genops(pickle):
- if op.code == code:
- return True
- return False
-
-# Return the number of times opcode code appears in pickle.
-def count_opcode(code, pickle):
- n = 0
- for op, dummy, dummy in pickletools.genops(pickle):
- if op.code == code:
- n += 1
- return n
-
-# We can't very well test the extension registry without putting known stuff
-# in it, but we have to be careful to restore its original state. Code
-# should do this:
-#
-# e = ExtensionSaver(extension_code)
-# try:
-# fiddle w/ the extension registry's stuff for extension_code
-# finally:
-# e.restore()
-
-class ExtensionSaver:
- # Remember current registration for code (if any), and remove it (if
- # there is one).
- def __init__(self, code):
- self.code = code
- if code in copy_reg._inverted_registry:
- self.pair = copy_reg._inverted_registry[code]
- copy_reg.remove_extension(self.pair[0], self.pair[1], code)
- else:
- self.pair = None
-
- # Restore previous registration for code.
- def restore(self):
- code = self.code
- curpair = copy_reg._inverted_registry.get(code)
- if curpair is not None:
- copy_reg.remove_extension(curpair[0], curpair[1], code)
- pair = self.pair
- if pair is not None:
- copy_reg.add_extension(pair[0], pair[1], code)
-
-class C:
- def __cmp__(self, other):
- return cmp(self.__dict__, other.__dict__)
-
-import __main__
-__main__.C = C
-C.__module__ = "__main__"
-
-class myint(int):
- def __init__(self, x):
- self.str = str(x)
-
-class initarg(C):
-
- def __init__(self, a, b):
- self.a = a
- self.b = b
-
- def __getinitargs__(self):
- return self.a, self.b
-
-class metaclass(type):
- pass
-
-class use_metaclass(object):
- __metaclass__ = metaclass
-
-class pickling_metaclass(type):
- def __eq__(self, other):
- return (type(self) == type(other) and
- self.reduce_args == other.reduce_args)
-
- def __reduce__(self):
- return (create_dynamic_class, self.reduce_args)
-
- __hash__ = None
-
-def create_dynamic_class(name, bases):
- result = pickling_metaclass(name, bases, dict())
- result.reduce_args = (name, bases)
- return result
-
-# DATA0 .. DATA2 are the pickles we expect under the various protocols, for
-# the object returned by create_data().
-
-# break into multiple strings to avoid confusing font-lock-mode
-DATA0 = """(lp1
-I0
-aL1L
-aF2
-ac__builtin__
-complex
-p2
-""" + \
-"""(F3
-F0
-tRp3
-aI1
-aI-1
-aI255
-aI-255
-aI-256
-aI65535
-aI-65535
-aI-65536
-aI2147483647
-aI-2147483647
-aI-2147483648
-a""" + \
-"""(S'abc'
-p4
-g4
-""" + \
-"""(i__main__
-C
-p5
-""" + \
-"""(dp6
-S'foo'
-p7
-I1
-sS'bar'
-p8
-I2
-sbg5
-tp9
-ag9
-aI5
-a.
-"""
-
-# Disassembly of DATA0.
-DATA0_DIS = """\
- 0: ( MARK
- 1: l LIST (MARK at 0)
- 2: p PUT 1
- 5: I INT 0
- 8: a APPEND
- 9: L LONG 1L
- 13: a APPEND
- 14: F FLOAT 2.0
- 17: a APPEND
- 18: c GLOBAL '__builtin__ complex'
- 39: p PUT 2
- 42: ( MARK
- 43: F FLOAT 3.0
- 46: F FLOAT 0.0
- 49: t TUPLE (MARK at 42)
- 50: R REDUCE
- 51: p PUT 3
- 54: a APPEND
- 55: I INT 1
- 58: a APPEND
- 59: I INT -1
- 63: a APPEND
- 64: I INT 255
- 69: a APPEND
- 70: I INT -255
- 76: a APPEND
- 77: I INT -256
- 83: a APPEND
- 84: I INT 65535
- 91: a APPEND
- 92: I INT -65535
- 100: a APPEND
- 101: I INT -65536
- 109: a APPEND
- 110: I INT 2147483647
- 122: a APPEND
- 123: I INT -2147483647
- 136: a APPEND
- 137: I INT -2147483648
- 150: a APPEND
- 151: ( MARK
- 152: S STRING 'abc'
- 159: p PUT 4
- 162: g GET 4
- 165: ( MARK
- 166: i INST '__main__ C' (MARK at 165)
- 178: p PUT 5
- 181: ( MARK
- 182: d DICT (MARK at 181)
- 183: p PUT 6
- 186: S STRING 'foo'
- 193: p PUT 7
- 196: I INT 1
- 199: s SETITEM
- 200: S STRING 'bar'
- 207: p PUT 8
- 210: I INT 2
- 213: s SETITEM
- 214: b BUILD
- 215: g GET 5
- 218: t TUPLE (MARK at 151)
- 219: p PUT 9
- 222: a APPEND
- 223: g GET 9
- 226: a APPEND
- 227: I INT 5
- 230: a APPEND
- 231: . STOP
-highest protocol among opcodes = 0
-"""
-
-DATA1 = (']q\x01(K\x00L1L\nG@\x00\x00\x00\x00\x00\x00\x00'
- 'c__builtin__\ncomplex\nq\x02(G@\x08\x00\x00\x00\x00\x00'
- '\x00G\x00\x00\x00\x00\x00\x00\x00\x00tRq\x03K\x01J\xff\xff'
- '\xff\xffK\xffJ\x01\xff\xff\xffJ\x00\xff\xff\xffM\xff\xff'
- 'J\x01\x00\xff\xffJ\x00\x00\xff\xffJ\xff\xff\xff\x7fJ\x01\x00'
- '\x00\x80J\x00\x00\x00\x80(U\x03abcq\x04h\x04(c__main__\n'
- 'C\nq\x05oq\x06}q\x07(U\x03fooq\x08K\x01U\x03barq\tK\x02ubh'
- '\x06tq\nh\nK\x05e.'
- )
-
-# Disassembly of DATA1.
-DATA1_DIS = """\
- 0: ] EMPTY_LIST
- 1: q BINPUT 1
- 3: ( MARK
- 4: K BININT1 0
- 6: L LONG 1L
- 10: G BINFLOAT 2.0
- 19: c GLOBAL '__builtin__ complex'
- 40: q BINPUT 2
- 42: ( MARK
- 43: G BINFLOAT 3.0
- 52: G BINFLOAT 0.0
- 61: t TUPLE (MARK at 42)
- 62: R REDUCE
- 63: q BINPUT 3
- 65: K BININT1 1
- 67: J BININT -1
- 72: K BININT1 255
- 74: J BININT -255
- 79: J BININT -256
- 84: M BININT2 65535
- 87: J BININT -65535
- 92: J BININT -65536
- 97: J BININT 2147483647
- 102: J BININT -2147483647
- 107: J BININT -2147483648
- 112: ( MARK
- 113: U SHORT_BINSTRING 'abc'
- 118: q BINPUT 4
- 120: h BINGET 4
- 122: ( MARK
- 123: c GLOBAL '__main__ C'
- 135: q BINPUT 5
- 137: o OBJ (MARK at 122)
- 138: q BINPUT 6
- 140: } EMPTY_DICT
- 141: q BINPUT 7
- 143: ( MARK
- 144: U SHORT_BINSTRING 'foo'
- 149: q BINPUT 8
- 151: K BININT1 1
- 153: U SHORT_BINSTRING 'bar'
- 158: q BINPUT 9
- 160: K BININT1 2
- 162: u SETITEMS (MARK at 143)
- 163: b BUILD
- 164: h BINGET 6
- 166: t TUPLE (MARK at 112)
- 167: q BINPUT 10
- 169: h BINGET 10
- 171: K BININT1 5
- 173: e APPENDS (MARK at 3)
- 174: . STOP
-highest protocol among opcodes = 1
-"""
-
-DATA2 = ('\x80\x02]q\x01(K\x00\x8a\x01\x01G@\x00\x00\x00\x00\x00\x00\x00'
- 'c__builtin__\ncomplex\nq\x02G@\x08\x00\x00\x00\x00\x00\x00G\x00'
- '\x00\x00\x00\x00\x00\x00\x00\x86Rq\x03K\x01J\xff\xff\xff\xffK'
- '\xffJ\x01\xff\xff\xffJ\x00\xff\xff\xffM\xff\xffJ\x01\x00\xff\xff'
- 'J\x00\x00\xff\xffJ\xff\xff\xff\x7fJ\x01\x00\x00\x80J\x00\x00\x00'
- '\x80(U\x03abcq\x04h\x04(c__main__\nC\nq\x05oq\x06}q\x07(U\x03foo'
- 'q\x08K\x01U\x03barq\tK\x02ubh\x06tq\nh\nK\x05e.')
-
-# Disassembly of DATA2.
-DATA2_DIS = """\
- 0: \x80 PROTO 2
- 2: ] EMPTY_LIST
- 3: q BINPUT 1
- 5: ( MARK
- 6: K BININT1 0
- 8: \x8a LONG1 1L
- 11: G BINFLOAT 2.0
- 20: c GLOBAL '__builtin__ complex'
- 41: q BINPUT 2
- 43: G BINFLOAT 3.0
- 52: G BINFLOAT 0.0
- 61: \x86 TUPLE2
- 62: R REDUCE
- 63: q BINPUT 3
- 65: K BININT1 1
- 67: J BININT -1
- 72: K BININT1 255
- 74: J BININT -255
- 79: J BININT -256
- 84: M BININT2 65535
- 87: J BININT -65535
- 92: J BININT -65536
- 97: J BININT 2147483647
- 102: J BININT -2147483647
- 107: J BININT -2147483648
- 112: ( MARK
- 113: U SHORT_BINSTRING 'abc'
- 118: q BINPUT 4
- 120: h BINGET 4
- 122: ( MARK
- 123: c GLOBAL '__main__ C'
- 135: q BINPUT 5
- 137: o OBJ (MARK at 122)
- 138: q BINPUT 6
- 140: } EMPTY_DICT
- 141: q BINPUT 7
- 143: ( MARK
- 144: U SHORT_BINSTRING 'foo'
- 149: q BINPUT 8
- 151: K BININT1 1
- 153: U SHORT_BINSTRING 'bar'
- 158: q BINPUT 9
- 160: K BININT1 2
- 162: u SETITEMS (MARK at 143)
- 163: b BUILD
- 164: h BINGET 6
- 166: t TUPLE (MARK at 112)
- 167: q BINPUT 10
- 169: h BINGET 10
- 171: K BININT1 5
- 173: e APPENDS (MARK at 5)
- 174: . STOP
-highest protocol among opcodes = 2
-"""
-
-def create_data():
- c = C()
- c.foo = 1
- c.bar = 2
- x = [0, 1L, 2.0, 3.0+0j]
- # Append some integer test cases at cPickle.c's internal size
- # cutoffs.
- uint1max = 0xff
- uint2max = 0xffff
- int4max = 0x7fffffff
- x.extend([1, -1,
- uint1max, -uint1max, -uint1max-1,
- uint2max, -uint2max, -uint2max-1,
- int4max, -int4max, -int4max-1])
- y = ('abc', 'abc', c, c)
- x.append(y)
- x.append(y)
- x.append(5)
- return x
-
-class AbstractPickleTests(unittest.TestCase):
- # Subclass must define self.dumps, self.loads, self.error.
-
- _testdata = create_data()
-
- def setUp(self):
- pass
-
- def test_misc(self):
- # test various datatypes not tested by testdata
- for proto in protocols:
- x = myint(4)
- s = self.dumps(x, proto)
- y = self.loads(s)
- self.assertEqual(x, y)
-
- x = (1, ())
- s = self.dumps(x, proto)
- y = self.loads(s)
- self.assertEqual(x, y)
-
- x = initarg(1, x)
- s = self.dumps(x, proto)
- y = self.loads(s)
- self.assertEqual(x, y)
-
- # XXX test __reduce__ protocol?
-
- def test_roundtrip_equality(self):
- expected = self._testdata
- for proto in protocols:
- s = self.dumps(expected, proto)
- got = self.loads(s)
- self.assertEqual(expected, got)
-
- def test_load_from_canned_string(self):
- expected = self._testdata
- for canned in DATA0, DATA1, DATA2:
- got = self.loads(canned)
- self.assertEqual(expected, got)
-
- # There are gratuitous differences between pickles produced by
- # pickle and cPickle, largely because cPickle starts PUT indices at
- # 1 and pickle starts them at 0. See XXX comment in cPickle's put2() --
- # there's a comment with an exclamation point there whose meaning
- # is a mystery. cPickle also suppresses PUT for objects with a refcount
- # of 1.
- def dont_test_disassembly(self):
- from pickletools import dis
-
- for proto, expected in (0, DATA0_DIS), (1, DATA1_DIS):
- s = self.dumps(self._testdata, proto)
- filelike = cStringIO.StringIO()
- dis(s, out=filelike)
- got = filelike.getvalue()
- self.assertEqual(expected, got)
-
- def test_recursive_list(self):
- l = []
- l.append(l)
- for proto in protocols:
- s = self.dumps(l, proto)
- x = self.loads(s)
- self.assertEqual(len(x), 1)
- self.assertTrue(x is x[0])
-
- def test_recursive_tuple(self):
- t = ([],)
- t[0].append(t)
- for proto in protocols:
- s = self.dumps(t, proto)
- x = self.loads(s)
- self.assertEqual(len(x), 1)
- self.assertEqual(len(x[0]), 1)
- self.assertTrue(x is x[0][0])
-
- def test_recursive_dict(self):
- d = {}
- d[1] = d
- for proto in protocols:
- s = self.dumps(d, proto)
- x = self.loads(s)
- self.assertEqual(x.keys(), [1])
- self.assertTrue(x[1] is x)
-
- def test_recursive_inst(self):
- i = C()
- i.attr = i
- for proto in protocols:
- s = self.dumps(i, proto)
- x = self.loads(s)
- self.assertEqual(dir(x), dir(i))
- self.assertIs(x.attr, x)
-
- def test_recursive_multi(self):
- l = []
- d = {1:l}
- i = C()
- i.attr = d
- l.append(i)
- for proto in protocols:
- s = self.dumps(l, proto)
- x = self.loads(s)
- self.assertEqual(len(x), 1)
- self.assertEqual(dir(x[0]), dir(i))
- self.assertEqual(x[0].attr.keys(), [1])
- self.assertTrue(x[0].attr[1] is x)
-
- def test_garyp(self):
- self.assertRaises(self.error, self.loads, 'garyp')
-
- def test_insecure_strings(self):
- insecure = ["abc", "2 + 2", # not quoted
- #"'abc' + 'def'", # not a single quoted string
- "'abc", # quote is not closed
- "'abc\"", # open quote and close quote don't match
- "'abc' ?", # junk after close quote
- "'\\'", # trailing backslash
- "'", # issue #17710
- "' ", # issue #17710
- # some tests of the quoting rules
- #"'abc\"\''",
- #"'\\\\a\'\'\'\\\'\\\\\''",
- ]
- for s in insecure:
- buf = "S" + s + "\012p0\012."
- self.assertRaises(ValueError, self.loads, buf)
-
- if have_unicode:
- def test_unicode(self):
- endcases = [u'', u'<\\u>', u'<\\\u1234>', u'<\n>',
- u'<\\>', u'<\\\U00012345>']
- for proto in protocols:
- for u in endcases:
- p = self.dumps(u, proto)
- u2 = self.loads(p)
- self.assertEqual(u2, u)
-
- def test_unicode_high_plane(self):
- t = u'\U00012345'
- for proto in protocols:
- p = self.dumps(t, proto)
- t2 = self.loads(p)
- self.assertEqual(t2, t)
-
- def test_ints(self):
- import sys
- for proto in protocols:
- n = sys.maxint
- while n:
- for expected in (-n, n):
- s = self.dumps(expected, proto)
- n2 = self.loads(s)
- self.assertEqual(expected, n2)
- n = n >> 1
-
- def test_maxint64(self):
- maxint64 = (1L << 63) - 1
- data = 'I' + str(maxint64) + '\n.'
- got = self.loads(data)
- self.assertEqual(got, maxint64)
-
- # Try too with a bogus literal.
- data = 'I' + str(maxint64) + 'JUNK\n.'
- self.assertRaises(ValueError, self.loads, data)
-
- def test_long(self):
- for proto in protocols:
- # 256 bytes is where LONG4 begins.
- for nbits in 1, 8, 8*254, 8*255, 8*256, 8*257:
- nbase = 1L << nbits
- for npos in nbase-1, nbase, nbase+1:
- for n in npos, -npos:
- pickle = self.dumps(n, proto)
- got = self.loads(pickle)
- self.assertEqual(n, got)
- # Try a monster. This is quadratic-time in protos 0 & 1, so don't
- # bother with those.
- nbase = long("deadbeeffeedface", 16)
- nbase += nbase << 1000000
- for n in nbase, -nbase:
- p = self.dumps(n, 2)
- got = self.loads(p)
- self.assertEqual(n, got)
-
- def test_float(self):
- test_values = [0.0, 4.94e-324, 1e-310, 7e-308, 6.626e-34, 0.1, 0.5,
- 3.14, 263.44582062374053, 6.022e23, 1e30]
- test_values = test_values + [-x for x in test_values]
- for proto in protocols:
- for value in test_values:
- pickle = self.dumps(value, proto)
- got = self.loads(pickle)
- self.assertEqual(value, got)
-
- @run_with_locale('LC_ALL', 'de_DE', 'fr_FR')
- def test_float_format(self):
- # make sure that floats are formatted locale independent
- self.assertEqual(self.dumps(1.2)[0:3], 'F1.')
-
- def test_reduce(self):
- pass
-
- def test_getinitargs(self):
- pass
-
- def test_metaclass(self):
- a = use_metaclass()
- for proto in protocols:
- s = self.dumps(a, proto)
- b = self.loads(s)
- self.assertEqual(a.__class__, b.__class__)
-
- def test_dynamic_class(self):
- a = create_dynamic_class("my_dynamic_class", (object,))
- copy_reg.pickle(pickling_metaclass, pickling_metaclass.__reduce__)
- for proto in protocols:
- s = self.dumps(a, proto)
- b = self.loads(s)
- self.assertEqual(a, b)
-
- def test_structseq(self):
- import time
- import os
-
- t = time.localtime()
- for proto in protocols:
- s = self.dumps(t, proto)
- u = self.loads(s)
- self.assertEqual(t, u)
- if hasattr(os, "stat"):
- t = os.stat(os.curdir)
- s = self.dumps(t, proto)
- u = self.loads(s)
- self.assertEqual(t, u)
- if hasattr(os, "statvfs"):
- t = os.statvfs(os.curdir)
- s = self.dumps(t, proto)
- u = self.loads(s)
- self.assertEqual(t, u)
-
- # Tests for protocol 2
-
- def test_proto(self):
- build_none = pickle.NONE + pickle.STOP
- for proto in protocols:
- expected = build_none
- if proto >= 2:
- expected = pickle.PROTO + chr(proto) + expected
- p = self.dumps(None, proto)
- self.assertEqual(p, expected)
-
- oob = protocols[-1] + 1 # a future protocol
- badpickle = pickle.PROTO + chr(oob) + build_none
- try:
- self.loads(badpickle)
- except ValueError, detail:
- self.assertTrue(str(detail).startswith(
- "unsupported pickle protocol"))
- else:
- self.fail("expected bad protocol number to raise ValueError")
-
- def test_long1(self):
- x = 12345678910111213141516178920L
- for proto in protocols:
- s = self.dumps(x, proto)
- y = self.loads(s)
- self.assertEqual(x, y)
- self.assertEqual(opcode_in_pickle(pickle.LONG1, s), proto >= 2)
-
- def test_long4(self):
- x = 12345678910111213141516178920L << (256*8)
- for proto in protocols:
- s = self.dumps(x, proto)
- y = self.loads(s)
- self.assertEqual(x, y)
- self.assertEqual(opcode_in_pickle(pickle.LONG4, s), proto >= 2)
-
- def test_short_tuples(self):
- # Map (proto, len(tuple)) to expected opcode.
- expected_opcode = {(0, 0): pickle.TUPLE,
- (0, 1): pickle.TUPLE,
- (0, 2): pickle.TUPLE,
- (0, 3): pickle.TUPLE,
- (0, 4): pickle.TUPLE,
-
- (1, 0): pickle.EMPTY_TUPLE,
- (1, 1): pickle.TUPLE,
- (1, 2): pickle.TUPLE,
- (1, 3): pickle.TUPLE,
- (1, 4): pickle.TUPLE,
-
- (2, 0): pickle.EMPTY_TUPLE,
- (2, 1): pickle.TUPLE1,
- (2, 2): pickle.TUPLE2,
- (2, 3): pickle.TUPLE3,
- (2, 4): pickle.TUPLE,
- }
- a = ()
- b = (1,)
- c = (1, 2)
- d = (1, 2, 3)
- e = (1, 2, 3, 4)
- for proto in protocols:
- for x in a, b, c, d, e:
- s = self.dumps(x, proto)
- y = self.loads(s)
- self.assertEqual(x, y, (proto, x, s, y))
- expected = expected_opcode[proto, len(x)]
- self.assertEqual(opcode_in_pickle(expected, s), True)
-
- def test_singletons(self):
- # Map (proto, singleton) to expected opcode.
- expected_opcode = {(0, None): pickle.NONE,
- (1, None): pickle.NONE,
- (2, None): pickle.NONE,
-
- (0, True): pickle.INT,
- (1, True): pickle.INT,
- (2, True): pickle.NEWTRUE,
-
- (0, False): pickle.INT,
- (1, False): pickle.INT,
- (2, False): pickle.NEWFALSE,
- }
- for proto in protocols:
- for x in None, False, True:
- s = self.dumps(x, proto)
- y = self.loads(s)
- self.assertTrue(x is y, (proto, x, s, y))
- expected = expected_opcode[proto, x]
- self.assertEqual(opcode_in_pickle(expected, s), True)
-
- def test_newobj_tuple(self):
- x = MyTuple([1, 2, 3])
- x.foo = 42
- x.bar = "hello"
- for proto in protocols:
- s = self.dumps(x, proto)
- y = self.loads(s)
- self.assertEqual(tuple(x), tuple(y))
- self.assertEqual(x.__dict__, y.__dict__)
-
- def test_newobj_list(self):
- x = MyList([1, 2, 3])
- x.foo = 42
- x.bar = "hello"
- for proto in protocols:
- s = self.dumps(x, proto)
- y = self.loads(s)
- self.assertEqual(list(x), list(y))
- self.assertEqual(x.__dict__, y.__dict__)
-
- def test_newobj_generic(self):
- for proto in protocols:
- for C in myclasses:
- B = C.__base__
- x = C(C.sample)
- x.foo = 42
- s = self.dumps(x, proto)
- y = self.loads(s)
- detail = (proto, C, B, x, y, type(y))
- self.assertEqual(B(x), B(y), detail)
- self.assertEqual(x.__dict__, y.__dict__, detail)
-
- # Register a type with copy_reg, with extension code extcode. Pickle
- # an object of that type. Check that the resulting pickle uses opcode
- # (EXT[124]) under proto 2, and not in proto 1.
-
- def produce_global_ext(self, extcode, opcode):
- e = ExtensionSaver(extcode)
- try:
- copy_reg.add_extension(__name__, "MyList", extcode)
- x = MyList([1, 2, 3])
- x.foo = 42
- x.bar = "hello"
-
- # Dump using protocol 1 for comparison.
- s1 = self.dumps(x, 1)
- self.assertIn(__name__, s1)
- self.assertIn("MyList", s1)
- self.assertEqual(opcode_in_pickle(opcode, s1), False)
-
- y = self.loads(s1)
- self.assertEqual(list(x), list(y))
- self.assertEqual(x.__dict__, y.__dict__)
-
- # Dump using protocol 2 for test.
- s2 = self.dumps(x, 2)
- self.assertNotIn(__name__, s2)
- self.assertNotIn("MyList", s2)
- self.assertEqual(opcode_in_pickle(opcode, s2), True)
-
- y = self.loads(s2)
- self.assertEqual(list(x), list(y))
- self.assertEqual(x.__dict__, y.__dict__)
-
- finally:
- e.restore()
-
- def test_global_ext1(self):
- self.produce_global_ext(0x00000001, pickle.EXT1) # smallest EXT1 code
- self.produce_global_ext(0x000000ff, pickle.EXT1) # largest EXT1 code
-
- def test_global_ext2(self):
- self.produce_global_ext(0x00000100, pickle.EXT2) # smallest EXT2 code
- self.produce_global_ext(0x0000ffff, pickle.EXT2) # largest EXT2 code
- self.produce_global_ext(0x0000abcd, pickle.EXT2) # check endianness
-
- def test_global_ext4(self):
- self.produce_global_ext(0x00010000, pickle.EXT4) # smallest EXT4 code
- self.produce_global_ext(0x7fffffff, pickle.EXT4) # largest EXT4 code
- self.produce_global_ext(0x12abcdef, pickle.EXT4) # check endianness
-
- def test_list_chunking(self):
- n = 10 # too small to chunk
- x = range(n)
- for proto in protocols:
- s = self.dumps(x, proto)
- y = self.loads(s)
- self.assertEqual(x, y)
- num_appends = count_opcode(pickle.APPENDS, s)
- self.assertEqual(num_appends, proto > 0)
-
- n = 2500 # expect at least two chunks when proto > 0
- x = range(n)
- for proto in protocols:
- s = self.dumps(x, proto)
- y = self.loads(s)
- self.assertEqual(x, y)
- num_appends = count_opcode(pickle.APPENDS, s)
- if proto == 0:
- self.assertEqual(num_appends, 0)
- else:
- self.assertTrue(num_appends >= 2)
-
- def test_dict_chunking(self):
- n = 10 # too small to chunk
- x = dict.fromkeys(range(n))
- for proto in protocols:
- s = self.dumps(x, proto)
- y = self.loads(s)
- self.assertEqual(x, y)
- num_setitems = count_opcode(pickle.SETITEMS, s)
- self.assertEqual(num_setitems, proto > 0)
-
- n = 2500 # expect at least two chunks when proto > 0
- x = dict.fromkeys(range(n))
- for proto in protocols:
- s = self.dumps(x, proto)
- y = self.loads(s)
- self.assertEqual(x, y)
- num_setitems = count_opcode(pickle.SETITEMS, s)
- if proto == 0:
- self.assertEqual(num_setitems, 0)
- else:
- self.assertTrue(num_setitems >= 2)
-
- def test_simple_newobj(self):
- x = object.__new__(SimpleNewObj) # avoid __init__
- x.abc = 666
- for proto in protocols:
- s = self.dumps(x, proto)
- self.assertEqual(opcode_in_pickle(pickle.NEWOBJ, s), proto >= 2)
- y = self.loads(s) # will raise TypeError if __init__ called
- self.assertEqual(y.abc, 666)
- self.assertEqual(x.__dict__, y.__dict__)
-
- def test_newobj_list_slots(self):
- x = SlotList([1, 2, 3])
- x.foo = 42
- x.bar = "hello"
- s = self.dumps(x, 2)
- y = self.loads(s)
- self.assertEqual(list(x), list(y))
- self.assertEqual(x.__dict__, y.__dict__)
- self.assertEqual(x.foo, y.foo)
- self.assertEqual(x.bar, y.bar)
-
- def test_reduce_overrides_default_reduce_ex(self):
- for proto in protocols:
- x = REX_one()
- self.assertEqual(x._reduce_called, 0)
- s = self.dumps(x, proto)
- self.assertEqual(x._reduce_called, 1)
- y = self.loads(s)
- self.assertEqual(y._reduce_called, 0)
-
- def test_reduce_ex_called(self):
- for proto in protocols:
- x = REX_two()
- self.assertEqual(x._proto, None)
- s = self.dumps(x, proto)
- self.assertEqual(x._proto, proto)
- y = self.loads(s)
- self.assertEqual(y._proto, None)
-
- def test_reduce_ex_overrides_reduce(self):
- for proto in protocols:
- x = REX_three()
- self.assertEqual(x._proto, None)
- s = self.dumps(x, proto)
- self.assertEqual(x._proto, proto)
- y = self.loads(s)
- self.assertEqual(y._proto, None)
-
- def test_reduce_ex_calls_base(self):
- for proto in protocols:
- x = REX_four()
- self.assertEqual(x._proto, None)
- s = self.dumps(x, proto)
- self.assertEqual(x._proto, proto)
- y = self.loads(s)
- self.assertEqual(y._proto, proto)
-
- def test_reduce_calls_base(self):
- for proto in protocols:
- x = REX_five()
- self.assertEqual(x._reduce_called, 0)
- s = self.dumps(x, proto)
- self.assertEqual(x._reduce_called, 1)
- y = self.loads(s)
- self.assertEqual(y._reduce_called, 1)
-
- 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 is less strict and also accept iterables.
- for proto in protocols:
- try:
- self.dumps(C(), proto)
- except (AttributeError, pickle.PickleError, cPickle.PickleError):
- pass
- try:
- self.dumps(D(), proto)
- except (AttributeError, pickle.PickleError, cPickle.PickleError):
- pass
-
- def test_many_puts_and_gets(self):
- # Test that internal data structures correctly deal with lots of
- # puts/gets.
- keys = ("aaa" + str(i) for i in xrange(100))
- large_dict = dict((k, [4, 5, 6]) for k in keys)
- obj = [dict(large_dict), dict(large_dict), dict(large_dict)]
-
- for proto in protocols:
- dumped = self.dumps(obj, proto)
- loaded = self.loads(dumped)
- self.assertEqual(loaded, obj,
- "Failed protocol %d: %r != %r"
- % (proto, obj, loaded))
-
- def test_attribute_name_interning(self):
- # Test that attribute names of pickled objects are interned when
- # unpickling.
- for proto in protocols:
- x = C()
- x.foo = 42
- x.bar = "hello"
- s = self.dumps(x, proto)
- y = self.loads(s)
- x_keys = sorted(x.__dict__)
- y_keys = sorted(y.__dict__)
- for x_key, y_key in zip(x_keys, y_keys):
- self.assertIs(x_key, y_key)
-
-
-# Test classes for reduce_ex
-
-class REX_one(object):
- _reduce_called = 0
- def __reduce__(self):
- self._reduce_called = 1
- return REX_one, ()
- # No __reduce_ex__ here, but inheriting it from object
-
-class REX_two(object):
- _proto = None
- def __reduce_ex__(self, proto):
- self._proto = proto
- return REX_two, ()
- # No __reduce__ here, but inheriting it from object
-
-class REX_three(object):
- _proto = None
- def __reduce_ex__(self, proto):
- self._proto = proto
- return REX_two, ()
- def __reduce__(self):
- raise TestFailed, "This __reduce__ shouldn't be called"
-
-class REX_four(object):
- _proto = None
- def __reduce_ex__(self, proto):
- self._proto = proto
- return object.__reduce_ex__(self, proto)
- # Calling base class method should succeed
-
-class REX_five(object):
- _reduce_called = 0
- def __reduce__(self):
- self._reduce_called = 1
- return object.__reduce__(self)
- # This one used to fail with infinite recursion
-
-# Test classes for newobj
-
-class MyInt(int):
- sample = 1
-
-class MyLong(long):
- sample = 1L
-
-class MyFloat(float):
- sample = 1.0
-
-class MyComplex(complex):
- sample = 1.0 + 0.0j
-
-class MyStr(str):
- sample = "hello"
-
-class MyUnicode(unicode):
- sample = u"hello \u1234"
-
-class MyTuple(tuple):
- sample = (1, 2, 3)
-
-class MyList(list):
- sample = [1, 2, 3]
-
-class MyDict(dict):
- sample = {"a": 1, "b": 2}
-
-myclasses = [MyInt, MyLong, MyFloat,
- MyComplex,
- MyStr, MyUnicode,
- MyTuple, MyList, MyDict]
-
-
-class SlotList(MyList):
- __slots__ = ["foo"]
-
-class SimpleNewObj(object):
- def __init__(self, a, b, c):
- # raise an error, to make sure this isn't called
- raise TypeError("SimpleNewObj.__init__() didn't expect to get called")
-
-class AbstractPickleModuleTests(unittest.TestCase):
-
- def test_dump_closed_file(self):
- import os
- f = open(TESTFN, "w")
- try:
- f.close()
- self.assertRaises(ValueError, self.module.dump, 123, f)
- finally:
- os.remove(TESTFN)
-
- def test_load_closed_file(self):
- import os
- f = open(TESTFN, "w")
- try:
- f.close()
- self.assertRaises(ValueError, self.module.dump, 123, f)
- finally:
- os.remove(TESTFN)
-
- def test_load_from_and_dump_to_file(self):
- stream = cStringIO.StringIO()
- data = [123, {}, 124]
- self.module.dump(data, stream)
- stream.seek(0)
- unpickled = self.module.load(stream)
- self.assertEqual(unpickled, data)
-
- def test_highest_protocol(self):
- # Of course this needs to be changed when HIGHEST_PROTOCOL changes.
- self.assertEqual(self.module.HIGHEST_PROTOCOL, 2)
-
- def test_callapi(self):
- f = cStringIO.StringIO()
- # With and without keyword arguments
- self.module.dump(123, f, -1)
- self.module.dump(123, file=f, protocol=-1)
- self.module.dumps(123, -1)
- self.module.dumps(123, protocol=-1)
- self.module.Pickler(f, -1)
- self.module.Pickler(f, protocol=-1)
-
- def test_incomplete_input(self):
- s = StringIO.StringIO("X''.")
- self.assertRaises(EOFError, self.module.load, s)
-
- def test_restricted(self):
- # issue7128: cPickle failed in restricted mode
- builtins = {self.module.__name__: self.module,
- '__import__': __import__}
- d = {}
- teststr = "def f(): {0}.dumps(0)".format(self.module.__name__)
- exec teststr in {'__builtins__': builtins}, d
- d['f']()
-
- def test_bad_input(self):
- # Test issue4298
- s = '\x58\0\0\0\x54'
- self.assertRaises(EOFError, self.module.loads, s)
- # Test issue7455
- s = '0'
- # XXX Why doesn't pickle raise UnpicklingError?
- self.assertRaises((IndexError, cPickle.UnpicklingError),
- self.module.loads, s)
-
-class AbstractPersistentPicklerTests(unittest.TestCase):
-
- # This class defines persistent_id() and persistent_load()
- # functions that should be used by the pickler. All even integers
- # are pickled using persistent ids.
-
- def persistent_id(self, object):
- if isinstance(object, int) and object % 2 == 0:
- self.id_count += 1
- return str(object)
- else:
- return None
-
- def persistent_load(self, oid):
- self.load_count += 1
- object = int(oid)
- assert object % 2 == 0
- return object
-
- def test_persistence(self):
- self.id_count = 0
- self.load_count = 0
- L = range(10)
- self.assertEqual(self.loads(self.dumps(L)), L)
- self.assertEqual(self.id_count, 5)
- self.assertEqual(self.load_count, 5)
-
- def test_bin_persistence(self):
- self.id_count = 0
- self.load_count = 0
- L = range(10)
- self.assertEqual(self.loads(self.dumps(L, 1)), L)
- self.assertEqual(self.id_count, 5)
- self.assertEqual(self.load_count, 5)
-
-class AbstractPicklerUnpicklerObjectTests(unittest.TestCase):
-
- pickler_class = None
- unpickler_class = None
-
- def setUp(self):
- assert self.pickler_class
- assert self.unpickler_class
-
- def test_clear_pickler_memo(self):
- # To test whether clear_memo() has any effect, we pickle an object,
- # then pickle it again without clearing the memo; the two serialized
- # forms should be different. If we clear_memo() and then pickle the
- # object again, the third serialized form should be identical to the
- # first one we obtained.
- data = ["abcdefg", "abcdefg", 44]
- f = cStringIO.StringIO()
- pickler = self.pickler_class(f)
-
- pickler.dump(data)
- first_pickled = f.getvalue()
-
- # Reset StringIO object.
- f.seek(0)
- f.truncate()
-
- pickler.dump(data)
- second_pickled = f.getvalue()
-
- # Reset the Pickler and StringIO objects.
- pickler.clear_memo()
- f.seek(0)
- f.truncate()
-
- pickler.dump(data)
- third_pickled = f.getvalue()
-
- self.assertNotEqual(first_pickled, second_pickled)
- self.assertEqual(first_pickled, third_pickled)
-
- def test_priming_pickler_memo(self):
- # Verify that we can set the Pickler's memo attribute.
- data = ["abcdefg", "abcdefg", 44]
- f = cStringIO.StringIO()
- pickler = self.pickler_class(f)
-
- pickler.dump(data)
- first_pickled = f.getvalue()
-
- f = cStringIO.StringIO()
- primed = self.pickler_class(f)
- primed.memo = pickler.memo
-
- primed.dump(data)
- primed_pickled = f.getvalue()
-
- self.assertNotEqual(first_pickled, primed_pickled)
-
- def test_priming_unpickler_memo(self):
- # Verify that we can set the Unpickler's memo attribute.
- data = ["abcdefg", "abcdefg", 44]
- f = cStringIO.StringIO()
- pickler = self.pickler_class(f)
-
- pickler.dump(data)
- first_pickled = f.getvalue()
-
- f = cStringIO.StringIO()
- primed = self.pickler_class(f)
- primed.memo = pickler.memo
-
- primed.dump(data)
- primed_pickled = f.getvalue()
-
- unpickler = self.unpickler_class(cStringIO.StringIO(first_pickled))
- unpickled_data1 = unpickler.load()
-
- self.assertEqual(unpickled_data1, data)
-
- primed = self.unpickler_class(cStringIO.StringIO(primed_pickled))
- primed.memo = unpickler.memo
- unpickled_data2 = primed.load()
-
- primed.memo.clear()
-
- self.assertEqual(unpickled_data2, data)
- self.assertTrue(unpickled_data2 is unpickled_data1)
-
- def test_reusing_unpickler_objects(self):
- data1 = ["abcdefg", "abcdefg", 44]
- f = cStringIO.StringIO()
- pickler = self.pickler_class(f)
- pickler.dump(data1)
- pickled1 = f.getvalue()
-
- data2 = ["abcdefg", 44, 44]
- f = cStringIO.StringIO()
- pickler = self.pickler_class(f)
- pickler.dump(data2)
- pickled2 = f.getvalue()
-
- f = cStringIO.StringIO()
- f.write(pickled1)
- f.seek(0)
- unpickler = self.unpickler_class(f)
- self.assertEqual(unpickler.load(), data1)
-
- f.seek(0)
- f.truncate()
- f.write(pickled2)
- f.seek(0)
- self.assertEqual(unpickler.load(), data2)
-
-class BigmemPickleTests(unittest.TestCase):
-
- # Memory requirements: 1 byte per character for input strings, 1 byte
- # for pickled data, 1 byte for unpickled strings, 1 byte for internal
- # buffer and 1 byte of free space for resizing of internal buffer.
-
- @precisionbigmemtest(size=_2G + 100*_1M, memuse=5)
- def test_huge_strlist(self, size):
- chunksize = 2**20
- data = []
- while size > chunksize:
- data.append('x' * chunksize)
- size -= chunksize
- chunksize += 1
- data.append('y' * size)
-
- try:
- for proto in protocols:
- try:
- pickled = self.dumps(data, proto)
- res = self.loads(pickled)
- self.assertEqual(res, data)
- finally:
- res = None
- pickled = None
- finally:
- data = None
generated by cgit v1.2.3 (git 2.25.1) at 2024-06-07 23:12:43 +0000