table_test.py 文件源码

def test_bson(self):
        import pymongo
        outputs = run(
            self.analysis,
            inputs={
                'a': {
                    'format': 'objectlist.bson',
                    'mode': 'mongodb',
                    'db': 'test',
                    'collection': 'a'
                },
                'b': {
                    'format': 'objectlist.bson',
                    'mode': 'mongodb',
                    'db': 'test',
                    'collection': 'b'
                }
            },
            outputs={
                'c': {
                    'format': 'objectlist.bson',
                    'mode': 'mongodb',
                    'db': 'test',
                    'collection': 'temp'
                }
            })
        self.assertEqual(outputs['c']['format'], 'objectlist.bson')
        coll = pymongo.MongoClient('mongodb://localhost')['test']['temp']
        self.assertEqual([d for d in coll.find()], [self.aobj, self.bobj])