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
|
# -*- coding: utf-8 -*-
# :Project: python-rapidjson -- Refs leaks tests
# :Author: Lele Gaifax <lele@metapensiero.it>
# :License: MIT License
# :Copyright: © 2017, 2018, 2020 Lele Gaifax
#
# NB: this is a simplicistic test that uses sys.gettotalrefcount(), available
# when the interpreter is built --with-pydebug, that tries to assert that
# repeated calls to dumps() and loads() does not leak object references.
# Since it's not an exact science, it should be taken with a grain of salt.
import datetime
import io
import sys
import uuid
import pytest
import rapidjson as rj
class Foo:
def __init__(self, foo):
self.foo = foo
def hook(d):
if 'foo' in d:
return Foo(d['foo'])
return d
def default(obj):
return {'foo': obj.foo}
# Empirical tolerance used to test the refcount growing
THRESHOLD = 7
plain_string = 'plain-string'
bigint = 123456789
pi = 3.1415926
right_now = datetime.datetime.now()
date = right_now.date()
time = right_now.time()
nil = uuid.UUID(int=0)
foo = Foo('foo')
array = [dict(index=j, array=[dict(index=i,
plain_string=plain_string,
bigint=bigint,
pi=pi,
right_now=right_now,
date=date,
time=time,
nil=nil,
foos=[foo, foo, foo, foo, foo])
for i in range(10)])
for j in range(10)]
NO_OPTION = {}
DATETIMES = {'datetime_mode': rj.DM_ISO8601}
UUIDS = {'uuid_mode': rj.UM_CANONICAL}
FOOS_DUMP = {'default': default}
FOOS_LOAD = {'object_hook': hook}
ARRAY_DUMP = {'datetime_mode': rj.DM_ISO8601,
'uuid_mode': rj.UM_CANONICAL,
'default': default}
ARRAY_LOAD = {'datetime_mode': rj.DM_ISO8601,
'uuid_mode': rj.UM_CANONICAL,
'object_hook': hook}
@pytest.mark.skipif(not hasattr(sys, 'gettotalrefcount'), reason='Non-debug Python')
@pytest.mark.parametrize('value,dumps_options,loads_options', [
( plain_string, NO_OPTION, NO_OPTION ),
( bigint, NO_OPTION, NO_OPTION ),
( pi, NO_OPTION, NO_OPTION ),
( rj.RawJSON(' "foo" '), NO_OPTION, NO_OPTION),
( right_now, DATETIMES, DATETIMES ),
( date, DATETIMES, DATETIMES ),
( time, DATETIMES, DATETIMES ),
( nil, UUIDS, UUIDS ),
( foo, FOOS_DUMP, FOOS_LOAD ),
( array, ARRAY_DUMP, ARRAY_LOAD ),
])
def test_leaks(value, dumps_options, loads_options):
rc0 = sys.gettotalrefcount()
for _ in range(1000):
asjson = rj.dumps(value, **dumps_options)
aspython = rj.loads(asjson, **loads_options)
del asjson
del aspython
rc1 = sys.gettotalrefcount()
assert (rc1 - rc0) < THRESHOLD
rc0 = sys.gettotalrefcount()
for _ in range(1000):
stream = io.BytesIO()
none = rj.dump(value, stream, **dumps_options)
stream.seek(0)
aspython = rj.load(stream, **loads_options)
del none
del aspython
del stream
rc1 = sys.gettotalrefcount()
assert (rc1 - rc0) < THRESHOLD
@pytest.mark.skipif(not hasattr(sys, 'gettotalrefcount'), reason='Non-debug Python')
def test_dump_iterator_leaks():
rc0 = sys.gettotalrefcount()
value = iter(range(10000))
asjson = rj.dumps(value)
del asjson
del value
rc1 = sys.gettotalrefcount()
assert (rc1 - rc0) < THRESHOLD
@pytest.mark.skipif(not hasattr(sys, 'gettotalrefcount'), reason='Non-debug Python')
def test_decoder_call_leaks():
class MyDecoder(rj.Decoder):
def start_object(self):
return {}
def end_object(self, obj):
return obj
def end_array(self, array):
return array
def string(self, string):
return string
decoder = MyDecoder()
rc0 = sys.gettotalrefcount()
for _ in range(1000):
value = decoder('["foo", {"foo": "bar"}]')
del value
rc1 = sys.gettotalrefcount()
assert (rc1 - rc0) < THRESHOLD
@pytest.mark.parametrize('value', ['Foo', rj.RawJSON('Foo')])
@pytest.mark.skipif(not hasattr(sys, 'gettotalrefcount'), reason='Non-debug Python')
def test_encoder_call_leaks(value):
class MyEncoder(rj.Encoder):
def __init__(self, value):
self.value = value
def default(self, obj):
return self.value
class Foo:
pass
encoder = MyEncoder(value)
foo = Foo()
rc0 = sys.gettotalrefcount()
for _ in range(1000):
value = encoder(foo)
del value
rc1 = sys.gettotalrefcount()
assert (rc1 - rc0) < THRESHOLD
@pytest.mark.skipif(not hasattr(sys, 'gettotalrefcount'), reason='Non-debug Python')
def test_rawjson_constructor():
raw_json = rj.RawJSON('["foo", "bar"]')
rc0 = sys.gettotalrefcount()
for _ in range(1000):
value = '"foobar"'
raw_json.__init__(value)
del value
rc1 = sys.gettotalrefcount()
assert (rc1 - rc0) < THRESHOLD
@pytest.mark.skipif(not hasattr(sys, 'gettotalrefcount'), reason='Non-debug Python')
def test_rawjson_new():
rc0 = sys.gettotalrefcount()
for _ in range(1000):
raw_json = rj.RawJSON('["foo", "bar"]')
del raw_json
rc1 = sys.gettotalrefcount()
assert (rc1 - rc0) < THRESHOLD
@pytest.mark.skipif(not hasattr(sys, 'gettotalrefcount'), reason='Non-debug Python')
def test_invalid_json_load_leaks():
# See issue #148
value = '{"a":{"b":}}'
rc0 = sys.gettotalrefcount()
for _ in range(1000):
try:
rj.loads(value)
except rj.JSONDecodeError:
pass
del _
rc1 = sys.gettotalrefcount()
assert (rc1 - rc0) < THRESHOLD
|
