You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

使用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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.08 11:08:16