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
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
|
from __future__ import absolute_import, unicode_literals
from datetime import datetime
from decimal import Decimal
from math import ceil
import pytest
from amqp.basic_message import Message
from amqp.exceptions import FrameSyntaxError
from amqp.platform import pack
from amqp.serialization import GenericContent, _read_item, dumps, loads
class _ANY(object):
def __eq__(self, other):
return other is not None
def __ne__(self, other):
return other is None
class test_serialization:
@pytest.mark.parametrize('descr,frame,expected,cast', [
('S', b's8thequick', 'thequick', None),
('x', b'x\x00\x00\x00\x09thequick\xffIGNORED', b'thequick\xff', None),
('b', b'b' + pack('>B', True), True, None),
('B', b'B' + pack('>b', 123), 123, None),
('U', b'U' + pack('>h', -321), -321, None),
('u', b'u' + pack('>H', 321), 321, None),
('i', b'i' + pack('>I', 1234), 1234, None),
('L', b'L' + pack('>q', -32451), -32451, None),
('l', b'l' + pack('>Q', 32451), 32451, None),
('f', b'f' + pack('>f', 33.3), 34.0, ceil),
])
def test_read_item(self, descr, frame, expected, cast):
actual = _read_item(frame, 0)[0]
actual = cast(actual) if cast else actual
assert actual == expected
def test_read_item_V(self):
assert _read_item(b'V', 0)[0] is None
def test_roundtrip(self):
format = b'bobBlLbsbSTx'
x = dumps(format, [
True, 32, False, 3415, 4513134, 13241923419,
True, b'thequickbrownfox', False, 'jumpsoverthelazydog',
datetime(2015, 3, 13, 10, 23),
b'thequick\xff'
])
y = loads(format, x, 0)
assert [
True, 32, False, 3415, 4513134, 13241923419,
True, 'thequickbrownfox', False, 'jumpsoverthelazydog',
datetime(2015, 3, 13, 10, 23), b'thequick\xff'
] == y[0]
def test_int_boundaries(self):
format = b'F'
x = dumps(format, [
{'a': -2147483649, 'b': 2147483648}, # celery/celery#3121
])
y = loads(format, x, 0)
assert y[0] == [{
'a': -2147483649, 'b': 2147483648, # celery/celery#3121
}]
def test_loads_unknown_type(self):
with pytest.raises(FrameSyntaxError):
loads('y', 'asdsad', 0)
def test_float(self):
data = int(loads(b'fb', dumps(b'fb', [32.31, False]), 0)[0][0] * 100)
assert(data == 3231)
def test_table(self):
table = {
'foo': 32,
'bar': 'baz',
'nil': None,
'array': [
1, True, 'bar'
]
}
assert loads(b'F', dumps(b'F', [table]), 0)[0][0] == table
def test_array(self):
array = [
'A', 1, True, 33.3,
Decimal('55.5'), Decimal('-3.4'),
datetime(2015, 3, 13, 10, 23),
{'quick': 'fox', 'amount': 1},
[3, 'hens'],
None,
]
expected = list(array)
expected[6] = _ANY()
assert expected == loads('A', dumps('A', [array]), 0)[0][0]
def test_array_unknown_type(self):
with pytest.raises(FrameSyntaxError):
dumps('A', [[object()]])
def test_bit_offset_adjusted_correctly(self):
expected = [50, "quick", "fox", True,
False, False, True, True, {"prop1": True}]
buf = dumps('BssbbbbbF', expected)
actual, _ = loads('BssbbbbbF', buf, 0)
assert actual == expected
class test_GenericContent:
@pytest.fixture(autouse=True)
def setup_content(self):
self.g = GenericContent()
def test_getattr(self):
self.g.properties['foo'] = 30
with pytest.raises(AttributeError):
self.g.__setstate__
assert self.g.foo == 30
with pytest.raises(AttributeError):
self.g.bar
def test_load_properties(self):
m = Message()
m.properties = {
'content_type': 'application/json',
'content_encoding': 'utf-8',
'application_headers': {
'foo': 1,
'id': 'id#1',
},
'delivery_mode': 1,
'priority': 255,
'correlation_id': 'df31-142f-34fd-g42d',
'reply_to': 'cosmo',
'expiration': '2015-12-23',
'message_id': '3312',
'timestamp': 3912491234,
'type': 'generic',
'user_id': 'george',
'app_id': 'vandelay',
'cluster_id': 'NYC',
}
s = m._serialize_properties()
m2 = Message()
m2._load_properties(m2.CLASS_ID, s)
assert m2.properties == m.properties
def test_load_properties__some_missing(self):
m = Message()
m.properties = {
'content_type': 'application/json',
'content_encoding': 'utf-8',
'delivery_mode': 1,
'correlation_id': 'df31-142f-34fd-g42d',
'reply_to': 'cosmo',
'expiration': '2015-12-23',
'message_id': '3312',
'type': None,
'app_id': None,
'cluster_id': None,
}
s = m._serialize_properties()
m2 = Message()
m2._load_properties(m2.CLASS_ID, s)
def test_inbound_header(self):
m = Message()
m.properties = {
'content_type': 'application/json',
'content_encoding': 'utf-8',
}
body = 'the quick brown fox'
buf = b'\0' * 30 + pack('>HxxQ', m.CLASS_ID, len(body))
buf += m._serialize_properties()
assert m.inbound_header(buf, offset=30) == 42
assert m.body_size == len(body)
assert m.properties['content_type'] == 'application/json'
assert not m.ready
def test_inbound_header__empty_body(self):
m = Message()
m.properties = {}
buf = pack('>HxxQ', m.CLASS_ID, 0)
buf += m._serialize_properties()
assert m.inbound_header(buf, offset=0) == 12
assert m.ready
def test_inbound_body(self):
m = Message()
m.body_size = 16
m.body_received = 8
m._pending_chunks = [b'the', b'quick']
m.inbound_body(b'brown')
assert not m.ready
m.inbound_body(b'fox')
assert m.ready
assert m.body == b'thequickbrownfox'
def test_inbound_body__no_chunks(self):
m = Message()
m.body_size = 16
m.inbound_body('thequickbrownfox')
assert m.ready
|
