00001
00002
00003
00004
00005
00006
00007
00008
00009
00010
00011
00012
00013
00014
00015
00016
00017 """Test the bson module."""
00018
00019 import unittest
00020 import datetime
00021 import re
00022 import sys
00023 try:
00024 import uuid
00025 should_test_uuid = True
00026 except ImportError:
00027 should_test_uuid = False
00028 sys.path[0:0] = [""]
00029
00030 from nose.plugins.skip import SkipTest
00031
00032 import bson
00033 from bson import (BSON,
00034 decode_all,
00035 is_valid)
00036 from bson.binary import Binary
00037 from bson.code import Code
00038 from bson.objectid import ObjectId
00039 from bson.dbref import DBRef
00040 from bson.son import SON
00041 from bson.timestamp import Timestamp
00042 from bson.errors import (InvalidDocument,
00043 InvalidStringData)
00044 from bson.max_key import MaxKey
00045 from bson.min_key import MinKey
00046 from bson.tz_util import (FixedOffset,
00047 utc)
00048 import pymongo
00049 import qcheck
00050
00051 class TestBSON(unittest.TestCase):
00052
00053 def setUp(self):
00054 pass
00055
00056 def test_basic_validation(self):
00057 self.assertRaises(TypeError, is_valid, 100)
00058 self.assertRaises(TypeError, is_valid, u"test")
00059 self.assertRaises(TypeError, is_valid, 10.4)
00060
00061 self.assertFalse(is_valid("test"))
00062
00063
00064 self.assert_(is_valid("\x05\x00\x00\x00\x00"))
00065 self.assert_(is_valid(BSON("\x05\x00\x00\x00\x00")))
00066 self.assertFalse(is_valid("\x04\x00\x00\x00\x00"))
00067 self.assertFalse(is_valid("\x05\x00\x00\x00\x01"))
00068 self.assertFalse(is_valid("\x05\x00\x00\x00"))
00069 self.assertFalse(is_valid("\x05\x00\x00\x00\x00\x00"))
00070
00071 def test_random_data_is_not_bson(self):
00072 qcheck.check_unittest(self, qcheck.isnt(is_valid),
00073 qcheck.gen_string(qcheck.gen_range(0, 40)))
00074
00075 def test_basic_decode(self):
00076 self.assertEqual({"test": u"hello world"},
00077 BSON("\x1B\x00\x00\x00\x0E\x74\x65\x73\x74\x00\x0C"
00078 "\x00\x00\x00\x68\x65\x6C\x6C\x6F\x20\x77\x6F"
00079 "\x72\x6C\x64\x00\x00").decode())
00080 self.assertEqual([{"test": u"hello world"}, {}],
00081 decode_all("\x1B\x00\x00\x00\x0E\x74\x65\x73\x74\x00"
00082 "\x0C\x00\x00\x00\x68\x65\x6C\x6C\x6F\x20"
00083 "\x77\x6F\x72\x6C\x64\x00\x00\x05\x00\x00"
00084 "\x00\x00"))
00085
00086 def test_data_timestamp(self):
00087 self.assertEqual({"test": Timestamp(4, 20)},
00088 BSON("\x13\x00\x00\x00\x11\x74\x65\x73\x74\x00\x14"
00089 "\x00\x00\x00\x04\x00\x00\x00\x00").decode())
00090
00091 def test_basic_encode(self):
00092 self.assertRaises(TypeError, BSON.encode, 100)
00093 self.assertRaises(TypeError, BSON.encode, "hello")
00094 self.assertRaises(TypeError, BSON.encode, None)
00095 self.assertRaises(TypeError, BSON.encode, [])
00096
00097 self.assertEqual(BSON.encode({}), BSON("\x05\x00\x00\x00\x00"))
00098 self.assertEqual(BSON.encode({"test": u"hello world"}),
00099 "\x1B\x00\x00\x00\x02\x74\x65\x73\x74\x00\x0C\x00\x00"
00100 "\x00\x68\x65\x6C\x6C\x6F\x20\x77\x6F\x72\x6C\x64\x00"
00101 "\x00")
00102 self.assertEqual(BSON.encode({u"mike": 100}),
00103 "\x0F\x00\x00\x00\x10\x6D\x69\x6B\x65\x00\x64\x00\x00"
00104 "\x00\x00")
00105 self.assertEqual(BSON.encode({"hello": 1.5}),
00106 "\x14\x00\x00\x00\x01\x68\x65\x6C\x6C\x6F\x00\x00\x00"
00107 "\x00\x00\x00\x00\xF8\x3F\x00")
00108 self.assertEqual(BSON.encode({"true": True}),
00109 "\x0C\x00\x00\x00\x08\x74\x72\x75\x65\x00\x01\x00")
00110 self.assertEqual(BSON.encode({"false": False}),
00111 "\x0D\x00\x00\x00\x08\x66\x61\x6C\x73\x65\x00\x00"
00112 "\x00")
00113 self.assertEqual(BSON.encode({"empty": []}),
00114 "\x11\x00\x00\x00\x04\x65\x6D\x70\x74\x79\x00\x05\x00"
00115 "\x00\x00\x00\x00")
00116 self.assertEqual(BSON.encode({"none": {}}),
00117 "\x10\x00\x00\x00\x03\x6E\x6F\x6E\x65\x00\x05\x00\x00"
00118 "\x00\x00\x00")
00119 self.assertEqual(BSON.encode({"test": Binary("test", 0)}),
00120 "\x14\x00\x00\x00\x05\x74\x65\x73\x74\x00\x04\x00\x00"
00121 "\x00\x00\x74\x65\x73\x74\x00")
00122 self.assertEqual(BSON.encode({"test": Binary("test")}),
00123 "\x18\x00\x00\x00\x05\x74\x65\x73\x74\x00\x08\x00\x00"
00124 "\x00\x02\x04\x00\x00\x00\x74\x65\x73\x74\x00")
00125 self.assertEqual(BSON.encode({"test": Binary("test", 128)}),
00126 "\x14\x00\x00\x00\x05\x74\x65\x73\x74\x00\x04\x00\x00"
00127 "\x00\x80\x74\x65\x73\x74\x00")
00128 self.assertEqual(BSON.encode({"test": None}),
00129 "\x0B\x00\x00\x00\x0A\x74\x65\x73\x74\x00\x00")
00130 self.assertEqual(BSON.encode({"date": datetime.datetime(2007, 1, 8,
00131 0, 30, 11)}),
00132 "\x13\x00\x00\x00\x09\x64\x61\x74\x65\x00\x38\xBE\x1C"
00133 "\xFF\x0F\x01\x00\x00\x00")
00134 self.assertEqual(BSON.encode({"regex": re.compile("a*b",
00135 re.IGNORECASE)}),
00136 "\x12\x00\x00\x00\x0B\x72\x65\x67\x65\x78\x00\x61\x2A"
00137 "\x62\x00\x69\x00\x00")
00138 self.assertEqual(BSON.encode({"$where": Code("test")}),
00139 "\x1F\x00\x00\x00\x0F\x24\x77\x68\x65\x72\x65\x00\x12"
00140 "\x00\x00\x00\x05\x00\x00\x00\x74\x65\x73\x74\x00\x05"
00141 "\x00\x00\x00\x00\x00")
00142 a = ObjectId("\x00\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0A\x0B")
00143 self.assertEqual(BSON.encode({"oid": a}),
00144 "\x16\x00\x00\x00\x07\x6F\x69\x64\x00\x00\x01\x02\x03"
00145 "\x04\x05\x06\x07\x08\x09\x0A\x0B\x00")
00146 self.assertEqual(BSON.encode({"ref": DBRef("coll", a)}),
00147 "\x2F\x00\x00\x00\x03ref\x00\x25\x00\x00\x00\x02$ref"
00148 "\x00\x05\x00\x00\x00coll\x00\x07$id\x00\x00\x01\x02"
00149 "\x03\x04\x05\x06\x07\x08\x09\x0A\x0B\x00\x00")
00150
00151 def test_encode_then_decode(self):
00152
00153 def helper(dict):
00154 self.assertEqual(dict, (BSON.encode(dict)).decode())
00155 helper({})
00156 helper({"test": u"hello"})
00157 self.assert_(isinstance(BSON.encode({"hello": "world"})
00158 .decode()["hello"],
00159 unicode))
00160 helper({"mike": -10120})
00161 helper({"long": long(10)})
00162 helper({"really big long": 2147483648})
00163 helper({u"hello": 0.0013109})
00164 helper({"something": True})
00165 helper({"false": False})
00166 helper({"an array": [1, True, 3.8, u"world"]})
00167 helper({"an object": {"test": u"something"}})
00168 helper({"a binary": Binary("test", 100)})
00169 helper({"a binary": Binary("test", 128)})
00170 helper({"a binary": Binary("test", 254)})
00171 helper({"another binary": Binary("test")})
00172 helper(SON([(u'test dst', datetime.datetime(1993, 4, 4, 2))]))
00173 helper({"big float": float(10000000000)})
00174 helper({"ref": DBRef("coll", 5)})
00175 helper({"ref": DBRef("coll", 5, foo="bar", bar=4)})
00176 helper({"ref": DBRef("coll", 5, "foo")})
00177 helper({"ref": DBRef("coll", 5, "foo", foo="bar")})
00178 helper({"ref": Timestamp(1,2)})
00179 helper({"foo": MinKey()})
00180 helper({"foo": MaxKey()})
00181
00182 def encode_then_decode(dict):
00183 return dict == (BSON.encode(dict)).decode()
00184
00185 qcheck.check_unittest(self, encode_then_decode,
00186 qcheck.gen_mongo_dict(3))
00187
00188 def test_aware_datetime(self):
00189 aware = datetime.datetime(1993, 4, 4, 2, tzinfo=FixedOffset(555, "SomeZone"))
00190 as_utc = (aware - aware.utcoffset()).replace(tzinfo=utc)
00191 self.assertEqual(datetime.datetime(1993, 4, 3, 16, 45, tzinfo=utc), as_utc)
00192 after = BSON.encode({"date": aware}).decode(tz_aware=True)["date"]
00193 self.assertEqual(utc, after.tzinfo)
00194 self.assertEqual(as_utc, after)
00195
00196 def test_naive_decode(self):
00197 aware = datetime.datetime(1993, 4, 4, 2, tzinfo=FixedOffset(555, "SomeZone"))
00198 naive_utc = (aware - aware.utcoffset()).replace(tzinfo=None)
00199 self.assertEqual(datetime.datetime(1993, 4, 3, 16, 45), naive_utc)
00200 after = BSON.encode({"date": aware}).decode()["date"]
00201 self.assertEqual(None, after.tzinfo)
00202 self.assertEqual(naive_utc, after)
00203
00204 def test_dst(self):
00205 d = {"x": datetime.datetime(1993, 4, 4, 2)}
00206 self.assertEqual(d, BSON.encode(d).decode())
00207
00208 def test_bad_encode(self):
00209 self.assertRaises(InvalidStringData, BSON.encode,
00210 {"lalala": '\xf4\xe0\xf0\xe1\xc0 Color Touch'})
00211 evil_list = {'a' : []}
00212 evil_list['a'].append(evil_list)
00213 evil_dict = {}
00214 evil_dict['a'] = evil_dict
00215 for evil_data in [evil_dict, evil_list]:
00216 self.assertRaises(RuntimeError, BSON.encode, evil_data)
00217
00218 def test_overflow(self):
00219 self.assert_(BSON.encode({"x": 9223372036854775807L}))
00220 self.assertRaises(OverflowError, BSON.encode, {"x": 9223372036854775808L})
00221
00222 self.assert_(BSON.encode({"x": -9223372036854775808L}))
00223 self.assertRaises(OverflowError, BSON.encode, {"x": -9223372036854775809L})
00224
00225 def test_tuple(self):
00226 self.assertEqual({"tuple": [1, 2]},
00227 BSON.encode({"tuple": (1, 2)}).decode())
00228
00229 def test_uuid(self):
00230 if not should_test_uuid:
00231 raise SkipTest()
00232
00233 id = uuid.uuid4()
00234 transformed_id = (BSON.encode({"id": id})).decode()["id"]
00235
00236 self.assert_(isinstance(transformed_id, uuid.UUID))
00237 self.assertEqual(id, transformed_id)
00238 self.assertNotEqual(uuid.uuid4(), transformed_id)
00239
00240
00241
00242 def test_unicode_regex(self):
00243 regex = re.compile(u'revisi\xf3n')
00244 BSON.encode({"regex": regex}).decode()
00245
00246 def test_non_string_keys(self):
00247 self.assertRaises(InvalidDocument, BSON.encode, {8.9: "test"})
00248
00249 def test_utf8(self):
00250 w = {u"aéあ": u"aéあ"}
00251 self.assertEqual(w, BSON.encode(w).decode())
00252
00253 x = {u"aéあ".encode("utf-8"): u"aéあ".encode("utf-8")}
00254 self.assertEqual(w, BSON.encode(x).decode())
00255
00256 y = {"hello": u"aé".encode("iso-8859-1")}
00257 self.assertRaises(InvalidStringData, BSON.encode, y)
00258
00259 z = {u"aé".encode("iso-8859-1"): "hello"}
00260 self.assertRaises(InvalidStringData, BSON.encode, z)
00261
00262 def test_null_character(self):
00263 doc = {"a": "\x00"}
00264 self.assertEqual(doc, BSON.encode(doc).decode())
00265
00266 doc = {"a": u"\x00"}
00267 self.assertEqual(doc, BSON.encode(doc).decode())
00268
00269 self.assertRaises(InvalidDocument, BSON.encode, {"\x00": "a"})
00270 self.assertRaises(InvalidDocument, BSON.encode, {u"\x00": "a"})
00271
00272 self.assertRaises(InvalidDocument, BSON.encode, {"a": re.compile("ab\x00c")})
00273 self.assertRaises(InvalidDocument, BSON.encode, {"a": re.compile(u"ab\x00c")})
00274
00275 def test_move_id(self):
00276 self.assertEqual("\x19\x00\x00\x00\x02_id\x00\x02\x00\x00\x00a\x00"
00277 "\x02a\x00\x02\x00\x00\x00a\x00\x00",
00278 BSON.encode(SON([("a", "a"), ("_id", "a")])))
00279
00280 self.assertEqual("\x2c\x00\x00\x00"
00281 "\x02_id\x00\x02\x00\x00\x00b\x00"
00282 "\x03b\x00"
00283 "\x19\x00\x00\x00\x02a\x00\x02\x00\x00\x00a\x00"
00284 "\x02_id\x00\x02\x00\x00\x00a\x00\x00\x00",
00285 BSON.encode(SON([("b", SON([("a", "a"), ("_id", "a")])),
00286 ("_id", "b")])))
00287
00288 def test_dates(self):
00289 doc = {"early": datetime.datetime(1686, 5, 5),
00290 "late": datetime.datetime(2086, 5, 5)}
00291 try:
00292 self.assertEqual(doc, BSON.encode(doc).decode())
00293 except ValueError:
00294
00295
00296
00297 if bson.has_c():
00298 raise
00299
00300 def test_custom_class(self):
00301 self.assert_(isinstance(BSON.encode({}).decode(), dict))
00302 self.assertFalse(isinstance(BSON.encode({}).decode(), SON))
00303 self.assert_(isinstance(BSON.encode({}).decode(SON), SON))
00304
00305 self.assertEqual(1, BSON.encode({"x": 1}).decode(SON)["x"])
00306
00307 x = BSON.encode({"x": [{"y": 1}]})
00308 self.assert_(isinstance(x.decode(SON)["x"][0], SON))
00309
00310 def test_subclasses(self):
00311
00312 class _myint(int):
00313 pass
00314 class _myfloat(float):
00315 pass
00316 class _myunicode(unicode):
00317 pass
00318 d = {'a' : _myint(42), 'b' : _myfloat(63.9),
00319 'c' : _myunicode('hello world')
00320 }
00321 d2 = BSON.encode(d).decode()
00322 for key, value in d2.iteritems():
00323 orig_value = d[key]
00324 orig_type = orig_value.__class__.__bases__[0]
00325 self.assertEqual(type(value), orig_type)
00326 self.assertEqual(value, orig_type(value))
00327
00328 def test_ordered_dict(self):
00329 try:
00330 from collections import OrderedDict
00331 except ImportError:
00332 raise SkipTest()
00333 d = OrderedDict([("one", 1), ("two", 2), ("three", 3), ("four", 4)])
00334 self.assertEqual(d, BSON.encode(d).decode(as_class=OrderedDict))
00335
00336
00337 if __name__ == "__main__":
00338 unittest.main()
