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
|
# -*- coding: utf-8 -*-
# :Project: python-rapidjson -- Tracemalloc-based leaks tests
# :Created: dom 10 feb 2019 13:47:32 CET
# :Author: Lele Gaifax <lele@metapensiero.it>
# :License: MIT License
# :Copyright: © 2019 Lele Gaifax
#
import io
import datetime
import gc
import tracemalloc
import pytest
import rapidjson as rj
def object_hook(td):
if '__td__' in td:
return datetime.timedelta(td['__td__'])
else:
return td
def default(obj):
if isinstance(obj, datetime.timedelta):
return {"__td__": obj.total_seconds()}
else:
return obj
def test_object_hook_and_default():
tracemalloc.start()
data = []
for i in range(1, 100):
data.append({"name": "a%d" % i, "timestamp": datetime.timedelta(seconds=i)})
snapshot1 = tracemalloc.take_snapshot().filter_traces((
tracemalloc.Filter(True, __file__),))
for _ in range(1000):
a = rj.dumps(data, default=default)
rj.loads(a, object_hook=object_hook)
gc.collect()
snapshot2 = tracemalloc.take_snapshot().filter_traces((
tracemalloc.Filter(True, __file__),))
top_stats = snapshot2.compare_to(snapshot1, 'lineno')
tracemalloc.stop()
for stat in top_stats[:10]:
assert stat.count_diff < 3
def test_load():
tracemalloc.start()
snapshot1 = tracemalloc.take_snapshot().filter_traces((
tracemalloc.Filter(True, __file__),))
for _ in range(10):
dct = '{' + ','.join('"foo%d":"bar%d"' % (i, i) for i in range(100)) + '}'
content = io.StringIO('[' + ','.join(dct for _ in range(100)) + ']')
rj.load(content, chunk_size=50)
del content
del _
gc.collect()
snapshot2 = tracemalloc.take_snapshot().filter_traces((
tracemalloc.Filter(True, __file__),))
top_stats = snapshot2.compare_to(snapshot1, 'lineno')
tracemalloc.stop()
for stat in top_stats[:10]:
assert stat.count_diff < 3
def test_failed_validation():
tracemalloc.start()
schema = """{
"$schema": "http://json-schema.org/draft-04/schema#",
"required": ["id", "name"],
"type": "object",
"properties": {
"id": {"type": "integer"},
"name": {"type": "string"}
}
}""".encode("utf-8")
obj = """{
"id": 50
}""".encode("utf-8")
validate = rj.Validator(schema)
snapshot1 = tracemalloc.take_snapshot().filter_traces((
tracemalloc.Filter(True, __file__),))
# start the test
for j in range(1000):
try:
validate(obj)
except rj.ValidationError:
pass
del j
gc.collect()
snapshot2 = tracemalloc.take_snapshot().filter_traces((
tracemalloc.Filter(True, __file__),))
top_stats = snapshot2.compare_to(snapshot1, 'lineno')
tracemalloc.stop()
for stat in top_stats[:10]:
assert stat.count_diff < 3
|
