使用pytest-asyncio测试异步aiohttp客户端时触发RuntimeError,但直接执行Python代码正常
我明白你遇到的问题了——在pytest环境下运行异步aiohttp代码时触发RuntimeError: Timeout context manager should be used inside a task,但直接用asyncio.run()执行就完全没问题。这个问题的根源其实和pytest-asyncio的事件循环管理以及aiohttp ClientSession的初始化时机有关,下面我来帮你分析并给出具体的修复方案:
问题根源
当你在pytest的fixture或者测试函数中初始化ClientSession时,有时候事件循环的上下文没有正确绑定,导致aiohttp的超时管理器找不到当前的任务上下文。而直接用asyncio.run()时,整个代码都运行在由它创建的任务上下文里,所有异步操作的上下文都是明确的,所以不会触发这个错误。
修复方案
方案1:确保在任务上下文内创建ClientSession
修改api_client.py中的_ensure_session方法,显式确保创建ClientSession时处于任务上下文:
import asyncio # 确保已经导入asyncio async def _ensure_session(self) -> aiohttp.ClientSession: """Ensure we have an active session, creating one if necessary.""" if self._session is None or self._session.closed: # 显式触发任务上下文绑定 if asyncio.current_task() is None: await asyncio.create_task(asyncio.sleep(0)) connector = aiohttp.TCPConnector(force_close=True) self._session = aiohttp.ClientSession( connector=connector, timeout=aiohttp.ClientTimeout(total=self.timeout) ) return self._session
这个方法通过创建一个空任务并等待它完成,强制当前代码进入任务上下文,这样aiohttp的超时管理器就能正确找到所需的上下文了。
方案2:调整pytest fixture的事件循环管理
修改conftest.py,显式创建并绑定事件循环,避免pytest-asyncio自动模式的上下文问题:
import pytest_asyncio import asyncio from loguru import logger from api_client import DemoAPIClient @pytest_asyncio.fixture(scope="module") async def event_loop(): # 为整个测试模块创建独立的事件循环 loop = asyncio.new_event_loop() yield loop loop.close() @pytest_asyncio.fixture(scope="module") async def api_client(event_loop): """Fixture to provide an authenticated API client.""" logger.info("Setting up API client") # 绑定事件循环到当前上下文 asyncio.set_event_loop(event_loop) client = DemoAPIClient() try: await client.login() logger.info("API client logged in successfully") yield client finally: await client.close() logger.info("API client closed")
这种方式确保fixture和所有测试函数都使用同一个明确的事件循环,避免了循环切换导致的上下文丢失。
方案3:使用pytest-aiohttp插件优化测试环境
如果上面的方法都不生效,可以试试专门为aiohttp测试设计的pytest-aiohttp插件:
- 先安装插件:
pip install pytest-aiohttp
- 修改测试函数,利用插件提供的
aiohttp_clientfixture来确保正确的上下文:
# test_api_client.py @pytest.mark.asyncio async def test_create_post(api_client, aiohttp_client): """Test creating a new post.""" try: new_post = { "title": "Test Post", "body": "Test Content", "userId": 1 } response = await api_client.rest("/posts", "POST", new_post) assert response is not None assert "id" in response assert response["title"] == "Test Post" except Exception as e: logger.error(f"Test failed: {str(e)}") raise
这个插件会自动处理aiohttp所需的事件循环和任务上下文,减少手动配置的麻烦。
为什么直接运行main()没问题?
因为asyncio.run()会自动创建一个新的事件循环,并把整个代码块包装在一个顶级任务中执行,所有的异步操作都处于明确的任务上下文里。而pytest-asyncio在处理fixture和测试函数时,有时候会因为循环的切换或者上下文绑定的延迟,导致创建ClientSession时没有处于任务上下文,从而触发超时管理器的错误。
你可以先试试方案1,它最直接地解决了ClientSession初始化时的上下文问题,应该能快速解决你的报错。
备注:内容来源于stack exchange,提问作者bcherb2

