异步测试中Task已销毁但仍处于pending状态的问题求助

错误提示的Task was destroyed but it is pending!指向未完全执行的异步生成器任务：

• 你的make_consumer fixture返回的是异步生成器函数_make_consumer，测试中仅通过await anext(make_consumer_gen)获取了yield出来的consumer对象，但没有驱动生成器执行yield之后的清理逻辑（await consumer.stop()）。
• 这个未完成的异步生成器会被asyncio包装成任务，测试结束时该任务仍处于pending状态，最终被销毁时抛出错误。

提供三种可行修复方案，可根据实际需求选择：

方案1：手动耗尽异步生成器 #

在测试代码末尾，驱动生成器执行剩余的清理代码，捕获生成器耗尽的异常：

@pytest.mark.asyncio
async def test_pending_todo_set(make_consumer, make_publisher, make_dispatcher_service):
    make_consumer_gen = make_consumer()
    consumer_1: ReconnectingConsumer = await anext(make_consumer_gen)

    tasks: List[asyncio.Task] = []
    tasks.append(asyncio.create_task(consumer_1.run()))

    for t in tasks:
        t.cancel()

    _ = await asyncio.gather(*tasks, return_exceptions=True)
    
    # 驱动生成器执行yield后的清理逻辑
    try:
        await anext(make_consumer_gen)
    except StopAsyncIteration:
        pass

方案2：将工厂fixture改为普通异步fixture（单实例场景） #

如果测试仅需一个consumer实例，直接将fixture改为异步yield形式，让pytest自动处理清理：

@pytest.fixture
async def consumer():
    consumer = ReconnectingConsumer(
        amqp_url=AMQP_URL, queue=INPUT_QUEUE, on_message=None
    )
    yield consumer
    await consumer.stop()
    print('Consumer stopped')

测试代码简化为：

@pytest.mark.asyncio
async def test_pending_todo_set(consumer, make_publisher, make_dispatcher_service):
    tasks: List[asyncio.Task] = []
    tasks.append(asyncio.create_task(consumer.run()))

    for t in tasks:
        t.cancel()

    _ = await asyncio.gather(*tasks, return_exceptions=True)

方案3：将工厂包装为异步上下文管理器（多实例场景） #

如果需要在测试中创建多个consumer实例，将工厂函数包装为异步上下文管理器，利用async with自动处理生命周期：

from contextlib import asynccontextmanager
import pytest

@pytest.fixture
def make_consumer():
    @asynccontextmanager
    async def _make_consumer():
        consumer = ReconnectingConsumer(
            amqp_url=AMQP_URL, queue=INPUT_QUEUE, on_message=None
        )
        try:
            yield consumer
        finally:
            await consumer.stop()
            print('Consumer stopped')
    
    return _make_consumer

测试中使用async with创建实例：

@pytest.mark.asyncio
async def test_pending_todo_set(make_consumer, make_publisher, make_dispatcher_service):
    async with make_consumer() as consumer_1:
        tasks: List[asyncio.Task] = []
        tasks.append(asyncio.create_task(consumer_1.run()))

        for t in tasks:
            t.cancel()

        _ = await asyncio.gather(*tasks, return_exceptions=True)

内容的提问来源于stack exchange，提问作者asant-leitha