testing.py 文件源码

def test_put_g2g(self):
        gm_client = gearman.GearmanClient(['localhost:4730'])
        agg = Aggregator(gm_client, None)

#        add_request = bson.BSON.encode({
#            'database':'feedlark',
#            'collecion':'g2g',
#            'data':{
#                'username':'__test123__',
#                'test_parameter':'NOLO'
#                }
#            })
#        gm_client.submit_job('db-add',str(add_request))

        test_document = {
            'username': 'iandioch',
            'test_parameter': 'YOLO',
            }
        agg.put_g2g('iandioch', test_document)

        get_request = bson.BSON.encode({
            'database': 'feedlark',
            'collection': 'g2g',
            'query': {
                'username': 'iandioch',
                },
            'projection': {
                'test_parameter': 1,
                },
            })
        g2g_data = gm_client.submit_job('db-get', str(get_request)).result
        self.assertEqual(
            bson.BSON(g2g_data).decode()['docs'][0]['test_parameter'], 'YOLO')