失败的测试/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
这是我测试过的代码
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