00001
00002
00003
00004
00005
00006
00007
00008
00009
00010
00011
00012
00013
00014
00015 """Tests for SONManipulators.
00016 """
00017
00018 import unittest
00019 import sys
00020 sys.path[0:0] = [""]
00021
00022 from bson.objectid import ObjectId
00023 from bson.son import SON
00024 from pymongo.database import Database
00025 from pymongo.son_manipulator import (NamespaceInjector,
00026 ObjectIdInjector,
00027 ObjectIdShuffler,
00028 SONManipulator)
00029 from test_connection import get_connection
00030 import qcheck
00031
00032
00033 class TestSONManipulator(unittest.TestCase):
00034
00035 def setUp(self):
00036 self.db = Database(get_connection(), "pymongo_test")
00037
00038 def test_basic(self):
00039 manip = SONManipulator()
00040 collection = self.db.test
00041
00042 def incoming_is_identity(son):
00043 return son == manip.transform_incoming(son, collection)
00044 qcheck.check_unittest(self, incoming_is_identity,
00045 qcheck.gen_mongo_dict(3))
00046
00047 def outgoing_is_identity(son):
00048 return son == manip.transform_outgoing(son, collection)
00049 qcheck.check_unittest(self, outgoing_is_identity,
00050 qcheck.gen_mongo_dict(3))
00051
00052 def test_id_injection(self):
00053 manip = ObjectIdInjector()
00054 collection = self.db.test
00055
00056 def incoming_adds_id(son):
00057 son = manip.transform_incoming(son, collection)
00058 assert "_id" in son
00059 return True
00060 qcheck.check_unittest(self, incoming_adds_id,
00061 qcheck.gen_mongo_dict(3))
00062
00063 def outgoing_is_identity(son):
00064 return son == manip.transform_outgoing(son, collection)
00065 qcheck.check_unittest(self, outgoing_is_identity,
00066 qcheck.gen_mongo_dict(3))
00067
00068 def test_id_shuffling(self):
00069 manip = ObjectIdShuffler()
00070 collection = self.db.test
00071
00072 def incoming_moves_id(son_in):
00073 son = manip.transform_incoming(son_in, collection)
00074 if not "_id" in son:
00075 return True
00076 for (k, v) in son.items():
00077 self.assertEqual(k, "_id")
00078 break
00079 return son_in == son
00080
00081 self.assert_(incoming_moves_id({}))
00082 self.assert_(incoming_moves_id({"_id": 12}))
00083 self.assert_(incoming_moves_id({"hello": "world", "_id": 12}))
00084 self.assert_(incoming_moves_id(SON([("hello", "world"),
00085 ("_id", 12)])))
00086
00087 def outgoing_is_identity(son):
00088 return son == manip.transform_outgoing(son, collection)
00089 qcheck.check_unittest(self, outgoing_is_identity,
00090 qcheck.gen_mongo_dict(3))
00091
00092 def test_ns_injection(self):
00093 manip = NamespaceInjector()
00094 collection = self.db.test
00095
00096 def incoming_adds_ns(son):
00097 son = manip.transform_incoming(son, collection)
00098 assert "_ns" in son
00099 return son["_ns"] == collection.name
00100 qcheck.check_unittest(self, incoming_adds_ns,
00101 qcheck.gen_mongo_dict(3))
00102
00103 def outgoing_is_identity(son):
00104 return son == manip.transform_outgoing(son, collection)
00105 qcheck.check_unittest(self, outgoing_is_identity,
00106 qcheck.gen_mongo_dict(3))
00107
00108 if __name__ == "__main__":
00109 unittest.main()