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
|
from __future__ import absolute_import, unicode_literals
from datetime import datetime
from decimal import Decimal
from uuid import uuid4
import pytest
import pytz
from case import MagicMock, Mock, skip
from kombu.five import text_t
from kombu.utils.encoding import str_to_bytes
from kombu.utils.json import _DecodeError, dumps, loads
class Custom(object):
def __init__(self, data):
self.data = data
def __json__(self):
return self.data
class test_JSONEncoder:
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()},
))
assert 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')
assert loads(dumps({'d': d})), {'d': text_t(d)}
def test_UUID(self):
id = uuid4()
assert loads(dumps({'u': id})), {'u': text_t(id)}
def test_default(self):
with pytest.raises(TypeError):
dumps({'o': object()})
class test_dumps_loads:
def test_dumps_custom_object(self):
x = {'foo': Custom({'a': 'b'})}
assert loads(dumps(x)) == {'foo': x['foo'].__json__()}
def test_dumps_custom_object_no_json(self):
x = {'foo': object()}
with pytest.raises(TypeError):
dumps(x)
def test_loads_memoryview(self):
assert loads(
memoryview(bytearray(dumps({'x': 'z'}), encoding='utf-8'))
) == {'x': 'z'}
def test_loads_bytearray(self):
assert loads(
bytearray(dumps({'x': 'z'}), encoding='utf-8')
) == {'x': 'z'}
def test_loads_bytes(self):
assert loads(
str_to_bytes(dumps({'x': 'z'})),
decode_bytes=True) == {'x': 'z'}
@skip.if_python3()
def test_loads_buffer(self):
assert loads(buffer(dumps({'x': 'z'}))) == {'x': 'z'}
def test_loads_DecodeError(self):
_loads = Mock(name='_loads')
_loads.side_effect = _DecodeError(
MagicMock(), MagicMock(), MagicMock())
assert loads(dumps({'x': 'z'}), _loads=_loads) == {'x': 'z'}
|
