我模拟异步生成器并收到错误

Question

失败的测试/test_main.py::test_main - TypeError: 'async for' 需要一个带有 aiter 方法的对象，得到协程

with patch('url_shortener_service.main.MessageBroker') as MockMessageBroker, \
         patch('url_shortener_service.main.Cache') as MockCache, \
         patch('url_shortener_service.main.DataBase') as MockDataBase, \
         patch('url_shortener_service.main.check_short_url', new_callable=AsyncMock) as mock_check_short_url:

        mock_broker_instance = MockMessageBroker.return_value.__aenter__.return_value 
        mock_broker_instance.consume_data.return_value.__aiter__.return_value = iter([
            json.dumps({"url": "http://example.com", "expiration": 3600, "prefix": "short"})
        ])```


tried all possible options

Answer 1

这是我测试过的代码

from .message_broker import MessageBroker
from .shortener_service import check_short_url
from .db import DataBase
from .cache import Cache
import asyncio
import json


async def main() -> None:
    async with MessageBroker() as broker:
        async for message in broker.consume_data():
            data = json.loads(message)
            long_url = data["url"]
            expiration = data["expiration"]
            short_url = await check_short_url(data["prefix"])
            async with Cache() as cache:
                await cache.create_recording(short_url, long_url, expiration)
            async with DataBase() as db:
                await db.create_recording(short_url, long_url, expiration)


def run() -> None:
    asyncio.run(main())

追溯：

__________________________________________________ test_main __________________________________________________

    @pytest.mark.asyncio
    async def test_main():
        with patch('url_shortener_service.main.MessageBroker') as MockMessageBroker, \
             patch('url_shortener_service.main.Cache') as MockCache, \
             patch('url_shortener_service.main.DataBase') as MockDataBase, \
             patch('url_shortener_service.main.check_short_url', new_callable=AsyncMock) as mock_check_short_url:
    
            mock_broker_instance = MockMessageBroker.return_value.__aenter__.return_value
            mock_broker_instance.consume_data.return_value.__aiter__.return_value = iter([
                json.dumps({"url": "http://example.com", "expiration": 3600, "prefix": "short"})
            ])
    
            mock_cache_instance = MockCache.return_value.__aenter__.return_valun
            mock_cache_instance.create_recording = AsyncMock()
    
            # Настройка заглушки для DataBase
            mock_db_instance = MockDataBase.return_value.__aenter__.return_value
            mock_db_instance.create_recording = AsyncMock()
    
            # Настройка возвращаемого значения для check_short_url
            mock_check_short_url.return_value = "shortened_url"
    
            # Запуск тестируемой функции
>           await main()

tests/test_main.py:29: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

    async def main() -> None:
        async with MessageBroker() as broker:
>           async for message in broker.consume_data():
E           TypeError: 'async for' requires an object with __aiter__ method, got coroutine

url_shortener_service/main.py:11: TypeError
============================================== warnings summary ===============================================
tests/test_main.py::test_main
  ~/My_Project/url_shortener_service/url_shortener_service/main.py:11: RuntimeWarning: coroutine 'AsyncMockMixin._execute_mock_call' was never awaited
    async for message in broker.consume_data():
  Enable tracemalloc to get traceback where the object was allocated.
  See https://docs.pytest.org/en/stable/how-to/capture-warnings.html#resource-warnings for more info.

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html

---------- coverage: platform darwin, python 3.12.5-final-0 ----------
Name                                         Stmts   Miss  Cover   Missing
--------------------------------------------------------------------------
url_shortener_service/__init__.py                0      0   100%
url_shortener_service/cache.py                  32      0   100%
url_shortener_service/db.py                     67      0   100%
url_shortener_service/main.py                   19      9    53%   12-19, 23
url_shortener_service/message_broker.py         22      0   100%
url_shortener_service/shortener_service.py      29      0   100%
--------------------------------------------------------------------------
TOTAL                                          169      9    95%

FAIL Required test coverage of 100% not reached. Total coverage: 94.67%
=========================================== short test summary info ===========================================
FAILED tests/test_main.py::test_main - TypeError: 'async for' requires an object with __aiter__ method, got coroutine