FastAPI异步测试覆盖率异常：代码未被标记为已执行

我有一个FastAPI应用，使用异步模式的SQLAlchemy 2.0连接PostgreSQL数据库，测试代码能正常通过，但Coverage工具显示部分代码未被覆盖，怀疑是conftest.py中的数据库依赖配置存在问题。

端点代码 #

@router.get("", response_model=HelloResponse)
async def get_hello(
    current_user: UserResponse = Security(
        get_current_active_user,
        scopes=[Scoper.admin, Scoper.staff],
    ),
    db=Depends(get_session),
):
    hello = await HelloManager(db).get(id=1)
    if hello.foo == "bar":
        return hello
    return await HelloManager(db).get(id=2)

测试代码（test_hello.py） #

@pytest.mark.asyncio
async def test_hello_with_mocked_security(async_client: AsyncClient):
    app.dependency_overrides[get_current_user] = mock_get_current_user

    response = await async_client.get("/api/hello", headers=headers_hocked_token)

    assert response.status_code == 200
    data = response.json()
    assert data["id"] == 2
    assert data["name"] == "Hello2 !!"  # 注：原代码此处为赋值符号=，应为==

覆盖问题 #

测试已成功通过，但Coverage工具标记hello = await HelloManager(db).get(id=1)之后的3行代码未被测试覆盖：

if hello.foo == "bar":
        return hello
    return await HelloManager(db).get(id=2)

同时多处Manager相关代码都被标记为未测试，但测试能返回正确结果。

conftest.py配置 #

from typing import AsyncIterator
from my_app.db.database import get_session
from httpx import AsyncClient
import pytest
from my_app.db.models.base_model import BaseModelORM
from my_app.settings import settings
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from my_app.main import app

from my_app.settings import get_settings as original_get_settings


@pytest.fixture(scope="session", autouse=True)
async def init_database() -> None:
    """
    Initialize the database before running the tests.
    """

    engine = create_async_engine(
        settings.sqlalchemy_database_uri_async or "",
        echo=False,
    )
    async with engine.begin() as conn:
        await conn.run_sync(BaseModelORM.metadata.drop_all)
        await conn.run_sync(BaseModelORM.metadata.create_all)

@pytest.fixture(scope="function")
async def async_engine():
    engine = create_async_engine(
        settings.sqlalchemy_database_uri_async,  # type: ignore
        echo=False,
    )
    return engine


@pytest.fixture(scope="function")
def TestingSessionLocalAsync(async_engine):
    return async_sessionmaker(
        async_engine,
        class_=AsyncSession,
        expire_on_commit=False,
    )


@pytest.fixture(scope="function", autouse=True)
async def db(async_engine, TestingSessionLocalAsync) -> AsyncIterator[AsyncSession]:
    async with async_engine.connect() as conn:
        async_session = TestingSessionLocalAsync()

        yield async_session

    await async_session.close()
    await conn.close()


@pytest.fixture(scope="session", autouse=True)
def get_settings():
    settings.fapi_mode = "TESTING"
    settings.sqlalchemy_database_uri_async = (
        f"postgresql+asyncpg://postgres:postgres@localhost:5432/test_myapp"
    )
    return settings


@pytest.fixture(scope="function")
def override_get_db(db: AsyncSession):
    async def _override_get_db():
        yield db

    return _override_get_db


@pytest.fixture(scope="function")
async def async_client(override_get_db, get_settings):
    app.dependency_overrides[get_session] = override_get_db
    app.dependency_overrides[original_get_settings] = get_settings
    async with AsyncClient(app=app, base_url="http://test.test") as ac:
        yield ac

解决方法 #

1. 修复Coverage对异步代码的追踪 #

在项目根目录创建.coveragerc文件，添加异步支持配置：

[run]
concurrency = asyncio
parallel = True
source = my_app  # 替换为你的应用模块名

运行测试时使用coverage run -m pytest，再生成报告。

2. 修正数据库会话fixture的生命周期 #

当前db fixture的会话管理存在问题，修改为事务化的会话，确保测试数据隔离且能被正确读取：

@pytest.fixture(scope="function")
async def db(async_engine, TestingSessionLocalAsync) -> AsyncIterator[AsyncSession]:
    async with async_engine.begin() as conn:
        # 每次测试前创建表（可选，若init_database已处理可跳过）
        await conn.run_sync(BaseModelORM.metadata.create_all)
        # 开启事务，测试后回滚
        await conn.begin_nested()
        async with TestingSessionLocalAsync(bind=conn) as session:
            # 可选：在此插入测试基础数据
            # from my_app.db.models import Hello
            # session.add(Hello(id=1, foo="not_bar", name="Hello1"))
            # session.add(Hello(id=2, foo="test", name="Hello2 !!"))
            # await session.commit()
            yield session
            # 回滚事务，清除测试数据
            await session.rollback()
    # 销毁引擎释放连接
    await async_engine.dispose()

3. 确保测试数据正确初始化 #

在测试函数中插入预期的测试数据，保证代码路径被真实执行：

@pytest.mark.asyncio
async def test_hello_with_mocked_security(async_client: AsyncClient, db: AsyncSession):
    from my_app.db.models import Hello
    # 插入测试数据，确保id=1的foo不是"bar"
    db.add(Hello(id=1, foo="not_bar", name="Hello1"))
    db.add(Hello(id=2, foo="test", name="Hello2 !!"))
    await db.commit()
    
    app.dependency_overrides[get_current_user] = mock_get_current_user

    response = await async_client.get("/api/hello", headers=headers_hocked_token)

    assert response.status_code == 200
    data = response.json()
    assert data["id"] == 2
    assert data["name"] == "Hello2 !!"

4. 清理依赖覆盖 #

在async_client fixture中，测试结束后清除依赖覆盖，避免影响其他测试：

@pytest.fixture(scope="function")
async def async_client(override_get_db, get_settings):
    original_overrides = app.dependency_overrides.copy()
    app.dependency_overrides[get_session] = override_get_db
    app.dependency_overrides[original_get_settings] = get_settings
    async with AsyncClient(app=app, base_url="http://test.test") as ac:
        yield ac
    # 恢复原始依赖
    app.dependency_overrides = original_overrides

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