summaryrefslogtreecommitdiff
path: root/kombu/tests/utils/test_json.py
blob: 1ad8d1c962c3e6b0c0a5d87b2df66bf35777931f (plain)
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'})