Тестирование функции AsyncIO MongoDB (драйвер двигателя Asyncio)

Question

Я новичок в AsyncIO и мне нужна помощь. У меня есть базовый класс, который запускает конвейер агрегации, который отлично работает в рабочей среде:

      class BasePipeline2:
    client: motor.AsyncIOMotorClient
    db: motor.AsyncIOMotorDatabase
    collection: motor.AsyncIOMotorCollection

    def __init__(self, uri: str, database: str, collection: str):
        self.client = get_client(uri)
        self.db = self.client[database]
        self.collection = self.db[collection]

    async def run(self):
        """Runs the pipeline and returns the results as a list"""
        result = await self.collection.aggregate({"$match": {"foo": "bar"}}).to_list(length=None)
        return result

Я представил следующий тест:

      @pytest.mark.asyncio
async def test_run2(self):
    results = [{"_id": 1, "name": "Alice", "age": 25}, {"_id": 2, "name": "Bob", "age": 30}]
    cursor_mock = mock.AsyncMock()
    cursor_mock.to_list = mock.AsyncMock(return_value=results)

    collection_mock = mock.MagicMock()
    collection_mock.aggregate = mock.AsyncMock(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

Я пытаюсь высмеятьto_listметод для возврата результатов теста. Но что бы я ни пытался, я получаю сообщение об ошибке:

      AttributeError: 'coroutine' object has no attribute 'to_list'

Кажется, я не могу найти решение и чувствую, что проблема где-то в том, как я настроил свои макеты.

Answer 1

Если кто-нибудь еще столкнется с этим: проблема была в том, как были настроены макеты. Правильный тест, который мне помог:

      @pytest.mark.asyncio
async def test_run2(self):
    results = [{"_id": 1, "name": "Alice", "age": 25}, {"_id": 2, "name": "Bob", "age": 30}]
    cursor_mock = mock.AsyncMock()
    cursor_mock.to_list = mock.AsyncMock(return_value=results)

    collection_mock = mock.MagicMock()
    collection_mock.aggregate = mock.AsyncMock(return_value=cursor_mock)

    pipeline = BasePipeline2("mongodb://localhost:27017/", "test_db", "test_collection")
    with mock.patch.object(pipeline.collection, "aggregate", return_value=cursor_mock) as aggregate_mock:
        pipeline_result = await pipeline.run()

    aggregate_mock.assert_called_once_with({"$match": {"foo": "bar"}})
    cursor_mock.to_list.assert_awaited()
    assert pipeline_result == results

Обратите внимание на исправленияwith mock.patch.object(pipeline.collection, "aggregate", return_value=cursor_mock), который исправляетaggregateметод для возврата макета курсора.