FastAPI依赖错误无法Mock：健康检查端点测试遇阻

相关代码 #

健康检查端点（app/routes/api_v1/status.py） #

@router.get("/", response_model=HealthStatus, summary="Database Health Check")
async def db_health_check(_: None = Depends(check_project_db_status_dependency)):
    return HealthStatus(status="Running")

数据库健康检查依赖（app/dependencies.py） #

async def check_project_db_status_dependency(session: AsyncSession = Depends(get_project_async_session)) -> None:
    try:
        result = await session.execute(text("SELECT 1"))
        if result.scalar() != 1:
            raise DatabaseNotAccessible(ErrorType.INTERNAL_ERROR.value.DB_NOT_RESPONDING_CORRECTLY)
    except OperationalError as exc:
        raise DatabaseNotAccessible(ErrorType.INTERNAL_ERROR.value.FAILED_TO_CONNECT_TO_DB) from exc

自定义错误处理器 #

async def database_not_accessible_error_handler(request: Request, exc: DatabaseNotAccessible) -> JSONResponse:
    """
    Handler for DatabaseNotAccessible exception.

    Args:
        request (Request): Request object.
        exc (DatabaseNotAccessible): Exception object.

    Returns:
        JSONResponse: Response object.
    """
    return JSONResponse(
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        content={"detail": str(exc)},
    )

现有测试的问题 #

目前尝试了两种方式：

• 依赖覆盖：能得到预期的500状态码，但完全跳过了实际的依赖逻辑，导致代码覆盖率工具标记这部分未覆盖。
• pytest patch：使用patch.object、mocker.patch.object直接patch依赖路径，均返回200状态码，Mock未生效。

现有依赖覆盖测试代码：

@pytest.mark.parametrize("project_name", PROJECTS)
def test_project_health_check_db_unreachable(
    async_main_session: IAsyncSession,
    project_name: str,
):
    def raise_error():
        raise DatabaseNotAccessible(ErrorType.INTERNAL_ERROR.value.DB_NOT_RESPONDING_CORRECTLY)

    app.dependency_overrides[check_project_db_status_dependency] = raise_error
    response = client.get(build_health_check_url(project_name))
    assert response.status_code == 500
    assert response.json()["detail"] == ErrorType.INTERNAL_ERROR.value.DB_NOT_RESPONDING_CORRECTLY.value


@pytest.mark.parametrize("project_name", PROJECTS)
def test_project_health_check_fail_to_connect(
    async_main_session: IAsyncSession,
    project_name: str,
):
    def raise_error():
        raise DatabaseNotAccessible(ErrorType.INTERNAL_ERROR.value.FAILED_TO_CONNECT_TO_DB)

    app.dependency_overrides[check_project_db_status_dependency] = raise_error
    response = client.get(build_health_check_url(project_name))
    assert response.status_code == 500
    assert response.json()["detail"] == ErrorType.INTERNAL_ERROR.value.FAILED_TO_CONNECT_TO_DB.value

项目结构：

main.py
app/
├── main.py
├── dependencies.py
└── routes/
    └── api_v1/
        └── status.py

正确的Mock方案 #

要覆盖实际依赖逻辑，需要Mock依赖中数据库会话的行为，而不是直接替换整个依赖。以下是针对两种错误场景的测试代码：

1. 测试数据库连接失败（OperationalError） #

MockAsyncSession.execute方法，使其抛出OperationalError，触发依赖中的异常捕获逻辑：

import pytest
from sqlalchemy.exc import OperationalError
from sqlalchemy import text
from app.dependencies import get_project_async_session

@pytest.mark.parametrize("project_name", PROJECTS)
async def test_project_health_check_fail_to_connect(
    mocker,
    project_name: str,
):
    # Mock数据库会话对象
    mock_session = mocker.Mock()
    # 让execute方法抛出OperationalError
    mock_session.execute.side_effect = OperationalError("Connection failed", {}, None)
    # 替换依赖中的get_project_async_session，返回Mock会话
    mocker.patch("app.dependencies.get_project_async_session", return_value=mock_session)

    response = client.get(build_health_check_url(project_name))
    assert response.status_code == 500
    assert response.json()["detail"] == ErrorType.INTERNAL_ERROR.value.FAILED_TO_CONNECT_TO_DB.value
    # 验证execute方法被正确调用
    mock_session.execute.assert_called_once_with(text("SELECT 1"))

2. 测试数据库返回异常结果（SELECT 1 != 1） #

Mocksession.execute返回的结果，使其scalar()方法返回非1的值：

import pytest
from sqlalchemy import text
from app.dependencies import get_project_async_session

@pytest.mark.parametrize("project_name", PROJECTS)
async def test_project_health_check_db_unreachable(
    mocker,
    project_name: str,
):
    # Mock会话和执行结果对象
    mock_session = mocker.Mock()
    mock_result = mocker.Mock()
    # 让scalar返回0，触发异常分支
    mock_result.scalar.return_value = 0
    mock_session.execute.return_value = mock_result
    # 替换依赖中的会话获取函数
    mocker.patch("app.dependencies.get_project_async_session", return_value=mock_session)

    response = client.get(build_health_check_url(project_name))
    assert response.status_code == 500
    assert response.json()["detail"] == ErrorType.INTERNAL_ERROR.value.DB_NOT_RESPONDING_CORRECTLY.value
    # 验证execute和scalar方法都被调用
    mock_session.execute.assert_called_once_with(text("SELECT 1"))
    mock_result.scalar.assert_called_once()

关键说明 #

• Patch路径正确性：必须patch依赖中实际导入并使用的对象路径（比如app.dependencies.get_project_async_session），而非端点中导入依赖的路径，这是之前Mock失效的核心原因。
• 保留实际逻辑：通过Mock会话的行为，让依赖代码完整执行，既触发了异常逻辑，又能被覆盖率工具统计到。
• 异步测试适配：测试函数需定义为async，确保异步Mock和请求执行流程正确。

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