dbtools_test.py 文件源码

def test_updater(self):
        req = {"database":"testing", "collection":"unit_tests", "data":{"inserttime":time(), "has_been_updated":False, "test":"updater"}}
        bsonReq = bson.BSON.encode(req)
        raw_response = self.client.submit_job('db-add', str(bsonReq))
        resp = bson.BSON.decode(bson.BSON(raw_response.result))
        ident = resp["_id"]
        req = {"database":"testing", "collection":"unit_tests", "data":{"selector":{"_id":ident}, "updates":{"has_been_updated":True}}}
        bsonReq = bson.BSON.encode(req)
        raw_response = self.client.submit_job('db-update', str(bsonReq))
        resp = bson.BSON.decode(bson.BSON(raw_response.result))
        self.assertTrue("status" in resp)
        self.assertEquals(resp["status"], "ok")