测试AsyncIO MongoDB函数（Asyncio Motor Driver）遇属性错误求助

错误AttributeError: 'coroutine' object has no attribute 'to_list'的核心原因是：你错误地用AsyncMock模拟了collection.aggregate方法。

Motor的aggregate方法是同步返回游标对象的（不需要await），它本身不是异步函数。但你用AsyncMock定义后，调用collection.aggregate(...)会返回一个协程对象，而不是你预先准备好的cursor_mock，协程对象自然没有to_list方法，所以报错。

修正后的测试代码 #

@pytest.mark.asyncio
async def test_run2():
    results = [{"_id": 1, "name": "Alice", "age": 25}, {"_id": 2, "name": "Bob", "age": 30}]
    
    # 游标Mock，to_list是异步方法，用AsyncMock
    cursor_mock = mock.Mock()
    cursor_mock.to_list = mock.AsyncMock(return_value=results)
    
    # collection的aggregate是同步方法，用普通Mock返回游标Mock
    collection_mock = mock.MagicMock()
    collection_mock.aggregate = mock.Mock(return_value=cursor_mock)

    pipeline = BasePipeline2("mongodb://localhost:27017/", "test_db", "test_collection")
    with mock.patch.object(pipeline, "collection", collection_mock):
        pipeline_result = await pipeline.run()

    assert pipeline_result == results

关键调整点 #

• 将collection_mock.aggregate从AsyncMock改为普通Mock：匹配Motor中aggregate同步返回游标的行为。
• 游标对象的to_list方法保持用AsyncMock：因为它是异步方法，需要await调用。

这样修改后，collection.aggregate(...)会返回你定义的cursor_mock，调用其to_list方法时就能正确返回预设的测试结果了。

内容的提问来源于stack exchange，提问作者garyj