如何使用pytest测试带有数据库依赖的FastAPI路由？

测试这类路由的核心是完全隔离真实数据库，不需要建立实际连接。根据测试粒度，你有两种可行的方案：

方案1：Mock数据库会话（延续Repository测试逻辑） #

这种方案和你已有的Repository测试思路一致，通过覆盖FastAPI的依赖注入，返回模拟的数据库会话，再配置会话的返回值，验证路由是否正确处理Repository的结果。

代码示例 #

from fastapi.testclient import AsyncTestClient
from your_module import router, get_session  # 替换为你的实际模块路径
import pytest
from unittest.mock import Mock
from sqlalchemy.ext.asyncio import AsyncSession

# 复用你已有的mock会话fixture
@pytest.fixture
def mock_postgres_async_session():
    yield Mock(spec=AsyncSession)

# 创建测试客户端，覆盖get_session依赖
@pytest.fixture
def client(mock_postgres_async_session):
    async def override_get_session():
        yield mock_postgres_async_session
    router.dependency_overrides[get_session] = override_get_session
    yield AsyncTestClient(router)
    # 测试后清理依赖覆盖，避免影响其他测试
    router.dependency_overrides.clear()

@pytest.mark.asyncio
async def test_get_cities_route_success(client, city_object, mock_postgres_async_session):
    # 配置mock会话的返回值，和测试Repository时的逻辑一致
    mock_result = Mock()
    mock_result.all.return_value = [city_object]
    mock_postgres_async_session.scalars.return_value = mock_result

    # 发送GET请求
    response = await client.get("/cities")

    # 验证响应状态和内容
    assert response.status_code == 200
    assert response.json() == {
        "cities": [
            {
                "id": 1,
                "city_name": "city_name",
                "sensor_list": [1]
            }
        ]
    }

    # 验证数据库会话的调用逻辑
    mock_postgres_async_session.scalars.assert_called_once()

方案2：直接Mock Repository方法 #

既然你已经单独测试过CitiesRepository，路由测试可以聚焦于路由自身的逻辑（比如数据格式化、异常处理），直接mockCitiesRepository.get_cities方法即可，无需再处理数据库会话的细节。

代码示例 #

from fastapi.testclient import AsyncTestClient
from your_module import router, CitiesRepository  # 替换为实际模块路径
import pytest
from unittest.mock import patch

@pytest.fixture
def client():
    yield AsyncTestClient(router)

@pytest.mark.asyncio
@patch("your_module.CitiesRepository.get_cities")  # 注意路径要匹配路由模块中导入的CitiesRepository
async def test_get_cities_route_success(mock_get_cities, client, city_object):
    # 配置mock方法返回预设数据
    mock_get_cities.return_value = [city_object]

    response = await client.get("/cities")

    # 验证响应
    assert response.status_code == 200
    assert response.json() == {
        "cities": [
            {
                "id": 1,
                "city_name": "city_name",
                "sensor_list": [1]
            }
        ]
    }

    # 验证Repository方法被正确调用
    mock_get_cities.assert_awaited_once()

异常场景测试示例 #

以404异常为例，验证路由是否正确捕获自定义异常并返回对应状态码：

@pytest.mark.asyncio
@patch("your_module.CitiesRepository.get_cities")
async def test_get_cities_route_not_found(mock_get_cities, client):
    from your_module import PostgreSQLNotFoundError  # 导入自定义异常类
    mock_get_cities.side_effect = PostgreSQLNotFoundError("Cities not found")

    response = await client.get("/cities")

    assert response.status_code == 404
    assert response.json()["detail"] == "Cities not found"

方案选择建议 #

• 方案1适合需要验证路由与Repository、数据库会话完整交互流程的场景
• 方案2更轻量化，适合已充分测试Repository后，仅验证路由数据转换、异常处理逻辑的场景

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