FastAPI异步测试完成后pytest进程无法退出的问题求助

Docker部署FastAPI服务后，执行测试命令：

docker exec container_name pytest /app/tests/ -W ignore -s --capture=no

测试显示通过（日志：============================== 1 passed in 36.76s ==============================），但pytest进程始终无法正常退出。

测试代码如下：

from ... import fastapp
import pytest
from async_asgi_testclient import TestClient as AsyncClient
from fastapi.testclient import TestClient

client = TestClient(fastapp)

@pytest.mark.django_db(transaction=True)
@pytest.mark.asyncio
async def test_something():
    async def fetch_stream():
        async with AsyncClient(fastapp) as ac:
            url = f"/api/..."
            response = await ac.post(
                url,
                json={...},
                stream=True,
            )
            result = ""
            async for line in response.iter_content(2):
                line = line.decode("utf-8").strip()
                result += line

            if (
                len(result) > 0
                and "failed" not in result
            ):
                print("success!")
                assert True

            else:
                print("false!")
                assert False

    await fetch_stream()

当前依赖版本：

async-asgi-testclient==1.4.11
pytest-asyncio==0.14.0
pytest-timeout==2.3.1
pytest==7.4.3

已尝试无效方案：

• 给测试用例加@pytest.mark.timeout(60)标记，仅限制测试执行时间，无法让pytest退出；
• 在docker exec命令中添加timeout，虽能终止进程但退出码非0，无法用于GitHub Actions；
• 在测试中添加exit(code=0)，测试失败且pytest仍无法退出；
• 使用httpx调用真实API，无法使用测试数据库。

1. 依赖版本不兼容：async-asgi-testclient 1.4.11是较旧版本，与pytest-asyncio 0.14.0、pytest 7.4.3的事件循环管理逻辑存在冲突，导致异步任务残留；
2. 异步资源未完全释放：流式响应未被正确关闭，导致事件循环中存在未完成的任务；
3. 全局TestClient残留：全局创建的TestClient实例未被正确销毁，持有事件循环资源。

1. 升级依赖版本 #

更新async-asgi-testclient和pytest-asyncio到兼容版本，解决底层兼容性问题：

pip install --upgrade async-asgi-testclient pytest-asyncio

建议升级到：

• async-asgi-testclient >= 1.5.0
• pytest-asyncio >= 0.21.0（适配pytest 7.x系列）

2. 确保流式响应完全关闭 #

在处理完响应后，主动调用aclose()关闭响应对象，释放异步资源：

async with AsyncClient(fastapp) as ac:
    url = f"/api/..."
    response = await ac.post(
        url,
        json={...},
        stream=True,
    )
    result = ""
    async for line in response.iter_content(2):
        line = line.decode("utf-8").strip()
        result += line
    # 主动关闭响应
    await response.aclose()

3. 避免全局创建TestClient #

将全局TestClient改为测试内局部创建，或用pytest fixture管理生命周期：

# 移除全局client定义，改用fixture
@pytest.fixture
def sync_client():
    with TestClient(fastapp) as client:
        yield client

@pytest.mark.django_db(transaction=True)
@pytest.mark.asyncio
async def test_something():
    # 测试内创建AsyncClient，而非全局
    async def fetch_stream():
        async with AsyncClient(fastapp) as ac:
            # ... 原有逻辑 ...
    await fetch_stream()

4. 替换为httpx.AsyncClient测试ASGI应用 #

httpx原生支持测试ASGI应用，无需调用真实API，可直接使用测试数据库，避免async-asgi-testclient的潜在问题：

from ... import fastapp
import pytest
from httpx import AsyncClient

@pytest.mark.django_db(transaction=True)
@pytest.mark.asyncio
async def test_something():
    async with AsyncClient(app=fastapp, base_url="http://testserver") as ac:
        url = f"/api/..."
        response = await ac.post(
            url,
            json={...},
            stream=True,
        )
        result = ""
        async for line in response.aiter_raw():
            line = line.decode("utf-8").strip()
            result += line
        
        assert len(result) > 0 and "failed" not in result
        await response.aclose()

5. 手动清理事件循环（兜底方案） #

在测试结束后手动关闭当前事件循环，强制释放资源：

import asyncio

@pytest.mark.django_db(transaction=True)
@pytest.mark.asyncio
async def test_something():
    # ... 原有测试逻辑 ...
    
    # 测试结束后关闭事件循环
    loop = asyncio.get_event_loop()
    loop.close()

内容的提问来源于stack exchange，提问作者김정빈