1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
|
#!/usr/bin/env python
# coding: utf-8
import six
import struct
from nose import main
from nose.tools import *
from nose.plugins.skip import SkipTest
from msgpack import packb, unpackb, Unpacker, Packer
from io import BytesIO
def check(data, use_list=False):
re = unpackb(packb(data), use_list=use_list)
assert_equal(re, data)
def testPack():
test_data = [
0, 1, 127, 128, 255, 256, 65535, 65536,
-1, -32, -33, -128, -129, -32768, -32769,
1.0,
b"", b"a", b"a"*31, b"a"*32,
None, True, False,
(), ((),), ((), None,),
{None: 0},
(1<<23),
]
for td in test_data:
check(td)
def testPackUnicode():
test_data = [
six.u(""), six.u("abcd"), [six.u("defgh")], six.u("Русский текст"),
]
for td in test_data:
re = unpackb(packb(td, encoding='utf-8'), use_list=1, encoding='utf-8')
assert_equal(re, td)
packer = Packer(encoding='utf-8')
data = packer.pack(td)
re = Unpacker(BytesIO(data), encoding='utf-8', use_list=1).unpack()
assert_equal(re, td)
def testPackUTF32():
try:
test_data = [
six.u(""),
six.u("abcd"),
[six.u("defgh")],
six.u("Русский текст"),
]
for td in test_data:
re = unpackb(packb(td, encoding='utf-32'), use_list=1, encoding='utf-32')
assert_equal(re, td)
except LookupError:
raise SkipTest
def testPackBytes():
test_data = [
b"", b"abcd", (b"defgh",),
]
for td in test_data:
check(td)
def testIgnoreUnicodeErrors():
re = unpackb(packb(b'abc\xeddef'), encoding='utf-8', unicode_errors='ignore', use_list=1)
assert_equal(re, "abcdef")
@raises(UnicodeDecodeError)
def testStrictUnicodeUnpack():
unpackb(packb(b'abc\xeddef'), encoding='utf-8', use_list=1)
@raises(UnicodeEncodeError)
def testStrictUnicodePack():
packb(six.u("abc\xeddef"), encoding='ascii', unicode_errors='strict')
def testIgnoreErrorsPack():
re = unpackb(packb(six.u("abcФФФdef"), encoding='ascii', unicode_errors='ignore'), encoding='utf-8', use_list=1)
assert_equal(re, six.u("abcdef"))
@raises(TypeError)
def testNoEncoding():
packb(six.u("abc"), encoding=None)
def testDecodeBinary():
re = unpackb(packb("abc"), encoding=None, use_list=1)
assert_equal(re, b"abc")
def testPackFloat():
assert_equal(packb(1.0, use_single_float=True), b'\xca' + struct.pack('>f', 1.0))
assert_equal(packb(1.0, use_single_float=False), b'\xcb' + struct.pack('>d', 1.0))
def testArraySize(sizes=[0, 5, 50, 1000]):
bio = six.BytesIO()
packer = Packer()
for size in sizes:
bio.write(packer.pack_array_header(size))
for i in range(size):
bio.write(packer.pack(i))
bio.seek(0)
unpacker = Unpacker(bio)
for size in sizes:
assert unpacker.unpack() == tuple(range(size))
def testMapSize(sizes=[0, 5, 50, 1000]):
bio = six.BytesIO()
packer = Packer()
for size in sizes:
bio.write(packer.pack_map_header(size))
for i in range(size):
bio.write(packer.pack(i)) # key
bio.write(packer.pack(i * 2)) # value
bio.seek(0)
unpacker = Unpacker(bio)
for size in sizes:
assert unpacker.unpack() == {i: i * 2 for i in range(size)}
class odict(dict):
'''Reimplement OrderedDict to run test on Python 2.6'''
def __init__(self, seq):
self._seq = seq
dict.__init__(self, seq)
def items(self):
return self._seq[:]
def iteritems(self):
return iter(self._seq)
def keys(self):
return [x[0] for x in self._seq]
def test_odict():
seq = [(b'one', 1), (b'two', 2), (b'three', 3), (b'four', 4)]
od = odict(seq)
assert_equal(unpackb(packb(od), use_list=1), dict(seq))
def pair_hook(seq):
return seq
assert_equal(unpackb(packb(od), object_pairs_hook=pair_hook, use_list=1), seq)
if __name__ == '__main__':
main()
|
