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
|
from __future__ import absolute_import, unicode_literals
import pytz
from datetime import datetime
from decimal import Decimal
from uuid import uuid4
from kombu.five import text_t
from kombu.utils.encoding import str_to_bytes
from kombu.utils.json import _DecodeError, dumps, loads
from kombu.tests.case import Case, MagicMock, Mock, skip
class Custom(object):
def __init__(self, data):
self.data = data
def __json__(self):
return self.data
class test_JSONEncoder(Case):
def test_datetime(self):
now = datetime.utcnow()
now_utc = now.replace(tzinfo=pytz.utc)
stripped = datetime(*now.timetuple()[:3])
serialized = loads(dumps({
'datetime': now,
'tz': now_utc,
'date': now.date(),
'time': now.time()},
))
self.assertDictEqual(serialized, {
'datetime': now.isoformat(),
'tz': '{0}Z'.format(now_utc.isoformat().split('+', 1)[0]),
'time': now.time().isoformat(),
'date': stripped.isoformat(),
})
def test_Decimal(self):
d = Decimal('3314132.13363235235324234123213213214134')
self.assertDictEqual(loads(dumps({'d': d})), {
'd': text_t(d),
})
def test_UUID(self):
id = uuid4()
self.assertDictEqual(loads(dumps({'u': id})), {
'u': text_t(id),
})
def test_default(self):
with self.assertRaises(TypeError):
dumps({'o': object()})
class test_dumps_loads(Case):
def test_dumps_custom_object(self):
x = {'foo': Custom({'a': 'b'})}
self.assertEqual(loads(dumps(x)), {'foo': x['foo'].__json__()})
def test_dumps_custom_object_no_json(self):
x = {'foo': object()}
with self.assertRaises(TypeError):
dumps(x)
def test_loads_memoryview(self):
self.assertEqual(
loads(memoryview(bytearray(dumps({'x': 'z'}), encoding='utf-8'))),
{'x': 'z'},
)
def test_loads_bytearray(self):
self.assertEqual(
loads(bytearray(dumps({'x': 'z'}), encoding='utf-8')),
{'x': 'z'})
def test_loads_bytes(self):
self.assertEqual(
loads(str_to_bytes(dumps({'x': 'z'})), decode_bytes=True),
{'x': 'z'},
)
@skip.if_python3()
def test_loads_buffer(self):
self.assertEqual(loads(buffer(dumps({'x': 'z'}))), {'x': 'z'})
def test_loads_DecodeError(self):
_loads = Mock(name='_loads')
_loads.side_effect = _DecodeError(
MagicMock(), MagicMock(), MagicMock())
self.assertEqual(loads(dumps({'x': 'z'}), _loads=_loads), {'x': 'z'})
|