异步FastAPI应用pytest测试配置问题及报错排查

问题背景 #

在使用pytest测试异步FastAPI+PostgreSQL集成应用时，测试数据库相关接口出现报错，可正常测试健康检查等无数据库依赖的简单接口，但集成测试失败。

现有conftest.py配置 #

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from fastapi.testclient import TestClient
from httpx import AsyncClient
import asyncio
import pytest
from your_app_module import app, get_db, Base, settings  # 替换为实际导入路径

DATABASE_URL = (
    "postgresql+asyncpg://"
    f"{settings.POSTGRES_USER}:{settings.POSTGRES_PASSWORD}@"
    f"{settings.POSTGRES_HOST}:{settings.POSTGRES_PORT}/"
    f"{settings.POSTGRES_DB_TEST}"
)

engine = create_async_engine(DATABASE_URL, echo=True)
async_session = sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)


async def override_get_db() -> AsyncGenerator[AsyncSession, None]:
    async with async_session() as session:
        yield session


app.dependency_overrides[get_db] = override_get_db


@pytest.fixture(autouse=True, scope="session")
async def prepare_database():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)


@pytest.fixture(scope="session")
def event_loop(request):
    """Create an instance of the default event loop for each test case."""
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()


client = TestClient(app)


@pytest.fixture(scope="session")
async def ac() -> AsyncGenerator[AsyncClient, None]:
    async with AsyncClient(app=app, base_url="http://test") as ac:
        yield ac

测试函数代码 #

import pytest
from httpx import AsyncClient

@pytest.mark.asyncio
async def test_get_country(ac: AsyncClient):
    response = await ac.get("/api/v1/countries/1")

    response_data = response.json()

    assert response.status_code == 200
    assert response["name"] == "Test Country"

报错信息 #

FAILED tests/test_country.py::test_get_country - AttributeError: 'async_generator' object has no attribute 'get'

错误原因与修复方案 #

错误原因 #

1. 异步fixture装饰器错误：使用普通的@pytest.fixture装饰异步fixture，pytest无法正确解析异步生成器，导致注入测试函数的是async_generator对象而非AsyncClient实例。
2. 断言逻辑错误：直接尝试从Response对象response中取name字段，正确做法是先调用.json()转为字典后再取值。

修复步骤 #

1. 安装pytest-asyncio
   执行命令安装依赖：

   pip install pytest-asyncio
   

2. 替换异步fixture的装饰器：将conftest.py中所有异步fixture的@pytest.fixture替换为@pytest_asyncio.fixture（包括prepare_database和ac）。
3. 移除自定义event_loop fixture：pytest-asyncio会自动管理事件循环，无需手动定义。
4. 修正测试函数的断言：将assert response["name"] == "Test Country"改为assert response_data["name"] == "Test Country"。

修正后的核心代码示例 #

修正后的conftest.py关键片段

# 导入pytest_asyncio
import pytest_asyncio

# 替换异步fixture装饰器
@pytest_asyncio.fixture(autouse=True, scope="session")
async def prepare_database():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)

@pytest_asyncio.fixture(scope="session")
async def ac() -> AsyncGenerator[AsyncClient, None]:
    async with AsyncClient(app=app, base_url="http://test") as ac:
        yield ac

修正后的测试函数

@pytest.mark.asyncio
async def test_get_country(ac: AsyncClient):
    response = await ac.get("/api/v1/countries/1")

    response_data = response.json()

    assert response.status_code == 200
    assert response_data["name"] == "Test Country"

额外注意事项 #

• 确保测试数据库POSTGRES_DB_TEST已提前创建，且数据库服务正常运行。
• 若需要每次测试前清空数据，可将prepare_database的scope改为function，并在yield前添加数据清空逻辑。

内容的提问来源于stack exchange，提问作者123 123