test_database.py 文件源码

def test_insert_find_one(self, test_db):
        a_doc = SON({'hello': 'world'})
        a_key = (await test_db.test.insert_one(a_doc)).inserted_id
        assert isinstance(a_doc['_id'], ObjectId)
        assert a_doc['_id'] == a_key
        assert a_doc == await test_db.test.find_one({'_id': a_doc['_id']})
        assert a_doc == await test_db.test.find_one(a_key)
        assert await test_db.test.find_one(ObjectId()) is None
        assert a_doc == await test_db.test.find_one({'hello': 'world'})
        assert await test_db.test.find_one({'hello': 'test'}) is None

        b = await test_db.test.find_one()
        b['hello'] = 'mike'
        await test_db.test.replace_one({'_id': b['_id']}, b)

        assert a_doc != await test_db.test.find_one(a_key)
        assert b == await test_db.test.find_one(a_key)
        assert b == await test_db.test.find_one()

        count = 0

        async with test_db.test.find() as cursor:
            async for _ in cursor:
                count += 1
        assert count == 1