使用pytest搭配异步生成器,测试通过后进程挂起
问题分析与解决方案
你的测试代码执行通过后进程挂起,核心原因是pytest-asyncio在处理类内异步测试时,事件循环未被正确清理,或者异步生成器资源未妥善释放。以下是几种可行的解决方法:
方法1:将测试代码移出类结构
pytest对类内异步测试的支持存在细节兼容问题,把测试相关的fixture、工具函数和用例改为顶级定义:
from typing import AsyncIterable import pytest class Detector: async def find_events(self) -> AsyncIterable[int]: yield 1 @pytest.fixture(autouse=True) def detector(): return Detector() async def run_detector(detector): results = detector.find_events() result_list = [] async for result in results: result_list.append(result) return result_list @pytest.mark.asyncio async def test_detector(detector): results = await run_detector(detector) assert results
方法2:手动关闭异步生成器
虽然async for理论上会自动关闭异步生成器,但显式调用aclose()可确保资源释放,避免事件循环挂起:
@staticmethod async def run_detector(detector): results = detector.find_events() result_list = [] try: async for result in results: result_list.append(result) finally: await results.aclose() return result_list
方法3:升级pytest-asyncio版本
旧版本pytest-asyncio在处理类内异步测试时存在bug,升级到最新版本(推荐>=0.21.0)可解决这类兼容性问题:
pip install --upgrade pytest-asyncio
方法4:指定事件循环策略
在项目根目录创建conftest.py,添加以下代码强制使用现代事件循环策略:
import asyncio import pytest @pytest.fixture(scope="session") def event_loop(): policy = asyncio.get_event_loop_policy() loop = policy.new_event_loop() yield loop loop.close()
内容的提问来源于stack exchange,提问作者Idan
相关产品推荐
相关产品推荐

