← Back to branch summary

~crunch.io/ubuntu/precise/pymongo/unstable

« back to all changes in this revision

Viewing changes to test/test_json_util.py

  • Committer: Joseph Tate
  • Date: 2013-01-31 08:00:57 UTC
  • mfrom: (1.1.12)
  • Revision ID: jtate@dragonstrider.com-20130131080057-y7lv17xi6x8c1j5x
New upstream release.

Show diffs side-by-side

added added

removed removed

Lines of Context:
test/test_json_util.py
18
18
import datetime
19
19
import re
20
20
import sys
21
 
json_lib = True
22
 
try:
23
 
    import json
24
 
except ImportError:
25
 
    try:
26
 
        import simplejson as json
27
 
    except ImportError:
28
 
        json_lib = False
29
 
try:
30
 
    import uuid
31
 
    should_test_uuid = True
32
 
except ImportError:
33
 
    should_test_uuid = False
34
21
 
35
22
from nose.plugins.skip import SkipTest
36
23
 
37
24
sys.path[0:0] = [""]
38
25
 
 
26
import bson
 
27
from bson.py3compat import b
 
28
from bson import json_util
 
29
from bson.binary import Binary, MD5_SUBTYPE, USER_DEFINED_SUBTYPE
 
30
from bson.code import Code
39
31
from bson.dbref import DBRef
40
 
from bson.json_util import default, object_hook
 
32
from bson.max_key import MaxKey
41
33
from bson.min_key import MinKey
42
 
from bson.max_key import MaxKey
43
34
from bson.objectid import ObjectId
44
35
from bson.timestamp import Timestamp
45
36
from bson.tz_util import utc
46
37
 
 
38
from test.test_connection import get_connection
 
39
 
47
40
PY3 = sys.version_info[0] == 3
48
41
 
49
42
 
50
43
class TestJsonUtil(unittest.TestCase):
51
44
 
52
45
    def setUp(self):
53
 
        if not json_lib:
54
 
            raise SkipTest()
 
46
        if not json_util.json_lib:
 
47
            raise SkipTest("No json or simplejson module")
 
48
 
 
49
        self.db = get_connection().pymongo_test
55
50
 
56
51
    def round_tripped(self, doc):
57
 
        return json.loads(json.dumps(doc, default=default),
58
 
                          object_hook=object_hook)
 
52
        return json_util.loads(json_util.dumps(doc))
59
53
 
60
54
    def round_trip(self, doc):
61
55
        self.assertEqual(doc, self.round_tripped(doc))
69
63
    def test_dbref(self):
70
64
        self.round_trip({"ref": DBRef("foo", 5)})
71
65
        self.round_trip({"ref": DBRef("foo", 5, "db")})
72
 
 
73
 
        # TODO this is broken when using cjson. See:
74
 
        #   http://jira.mongodb.org/browse/PYTHON-153
75
 
        #   http://bugs.python.org/issue6105
76
 
        #
77
 
        # self.assertEqual("{\"ref\": {\"$ref\": \"foo\", \"$id\": 5}}",
78
 
        #                  json.dumps({"ref": DBRef("foo", 5)},
79
 
        #                  default=default))
80
 
        # self.assertEqual("{\"ref\": {\"$ref\": \"foo\",
81
 
        #                              \"$id\": 5, \"$db\": \"bar\"}}",
82
 
        #                  json.dumps({"ref": DBRef("foo", 5, "bar")},
83
 
        #                  default=default))
 
66
        self.round_trip({"ref": DBRef("foo", ObjectId())})
 
67
        self.round_trip({"ref": DBRef("foo", ObjectId(), "db")})
84
68
 
85
69
    def test_datetime(self):
86
70
        # only millis, not micros
92
76
        self.assertEqual("a*b", res.pattern)
93
77
        if PY3:
94
78
            # re.UNICODE is a default in python 3.
95
 
            self.assertEqual(re.IGNORECASE|re.UNICODE, res.flags)
 
79
            self.assertEqual(re.IGNORECASE | re.UNICODE, res.flags)
96
80
        else:
97
81
            self.assertEqual(re.IGNORECASE, res.flags)
98
82
 
100
84
        self.round_trip({"m": MinKey()})
101
85
 
102
86
    def test_maxkey(self):
103
 
        self.round_trip({"m": MinKey()})
 
87
        self.round_trip({"m": MaxKey()})
104
88
 
105
89
    def test_timestamp(self):
106
 
        res = json.dumps({"ts": Timestamp(4, 13)}, default=default)
107
 
        dct = json.loads(res)
 
90
        res = json_util.json.dumps({"ts": Timestamp(4, 13)},
 
91
                                     default=json_util.default)
 
92
        dct = json_util.json.loads(res)
108
93
        self.assertEqual(dct['ts']['t'], 4)
109
94
        self.assertEqual(dct['ts']['i'], 13)
110
95
 
111
96
    def test_uuid(self):
112
 
        if not should_test_uuid:
113
 
            raise SkipTest()
 
97
        if not bson.has_uuid():
 
98
            raise SkipTest("No uuid module")
114
99
        self.round_trip(
115
 
                {'uuid': uuid.UUID('f47ac10b-58cc-4372-a567-0e02b2c3d479')})
 
100
                {'uuid': bson.uuid.UUID(
 
101
                            'f47ac10b-58cc-4372-a567-0e02b2c3d479')})
 
102
 
 
103
    def test_binary(self):
 
104
        bin_type_dict = {"bin": Binary(b("\x00\x01\x02\x03\x04"))}
 
105
        md5_type_dict = {
 
106
            "md5": Binary(b(' n7\x18\xaf\t/\xd1\xd1/\x80\xca\xe7q\xcc\xac'),
 
107
                MD5_SUBTYPE)}
 
108
        custom_type_dict = {"custom": Binary(b("hello"), USER_DEFINED_SUBTYPE)}
 
109
 
 
110
        self.round_trip(bin_type_dict)
 
111
        self.round_trip(md5_type_dict)
 
112
        self.round_trip(custom_type_dict)
 
113
 
 
114
        # PYTHON-443 ensure old type formats are supported
 
115
        json_bin_dump = json_util.dumps(bin_type_dict)
 
116
        self.assertTrue('"$type": "00"' in json_bin_dump)
 
117
        self.assertEqual(bin_type_dict,
 
118
            json_util.loads('{"bin": {"$type": 0, "$binary": "AAECAwQ="}}'))
 
119
 
 
120
        json_bin_dump = json_util.dumps(md5_type_dict)
 
121
        self.assertTrue('"$type": "05"' in json_bin_dump)
 
122
        self.assertEqual(md5_type_dict,
 
123
            json_util.loads('{"md5": {"$type": 5, "$binary":'
 
124
                            ' "IG43GK8JL9HRL4DK53HMrA=="}}'))
 
125
 
 
126
        json_bin_dump = json_util.dumps(custom_type_dict)
 
127
        self.assertTrue('"$type": "80"' in json_bin_dump)
 
128
        self.assertEqual(custom_type_dict,
 
129
            json_util.loads('{"custom": {"$type": 128, "$binary":'
 
130
                            ' "aGVsbG8="}}'))
 
131
 
 
132
        # Handle mongoexport where subtype >= 128
 
133
        self.assertEqual(128,
 
134
            json_util.loads('{"custom": {"$type": "ffffff80", "$binary":'
 
135
                            ' "aGVsbG8="}}')['custom'].subtype)
 
136
 
 
137
        self.assertEqual(255,
 
138
            json_util.loads('{"custom": {"$type": "ffffffff", "$binary":'
 
139
                            ' "aGVsbG8="}}')['custom'].subtype)
 
140
 
 
141
    def test_code(self):
 
142
        self.round_trip({"code": Code("function x() { return 1; }")})
 
143
        self.round_trip({"code": Code("function y() { return z; }", z=2)})
 
144
 
 
145
    def test_cursor(self):
 
146
        db = self.db
 
147
 
 
148
        db.drop_collection("test")
 
149
        docs = [
 
150
            {'foo': [1, 2]},
 
151
            {'bar': {'hello': 'world'}},
 
152
            {'code': Code("function x() { return 1; }")},
 
153
            {'bin': Binary(b("\x00\x01\x02\x03\x04"))},
 
154
            {'dbref': {'_ref': DBRef('simple',
 
155
                               ObjectId('509b8db456c02c5ab7e63c34'))}}
 
156
        ]
 
157
 
 
158
        db.test.insert(docs)
 
159
        reloaded_docs = json_util.loads(json_util.dumps(db.test.find()))
 
160
        for doc in docs:
 
161
            self.assertTrue(doc in reloaded_docs)
116
162
 
117
163
if __name__ == "__main__":
118
164
    unittest.main()