如何在FastAPI应用的Pytest测试中添加认证Mock？

我正在用pytest测试FastAPI应用，需要集成认证Mock功能。有一个需要认证的/roles端点，但不知道怎么在pytest配置里正确添加认证Mock。

现有pytest代码 #

# pytest code for testing /roles endpoint
import pytest
from fastapi.testclient import TestClient
from unittest.mock import patch, AsyncMock
from main import app
from tests.mock_data import test_roles_data

class MockRoles:
    async def setup(self):
        pass

    async def get_json(self):
        return test_roles_data

@pytest.fixture
def client():
    with TestClient(app) as c:
        yield c

@patch('routers.base.parse_params', return_value=('custom_fields', '2023-01-01', '2023-01-31'))
@patch('services.roles.Roles', return_value=MockRoles())
def test_roles_endpoint(mock_parse_params, mock_roles, client):
    response = client.get("/roles", headers={"account": "test_account"})

    assert response.status_code == 200
    assert response.json() == test_roles_data

待集成的假认证函数 #

# Fake authentication function
async def fake_authenticate(
        apitoken: Optional[str] = None,
        account: Optional[str] = None,
        sessionHost: Optional[str] = None
):
    return {
        'apitoken': 'apitoken',
        'account': 'account',
        'sessionHost': 'sessionHost',
    }

额外信息 #

• FastAPI应用的/roles端点需要认证。
• 希望在假设认证成功的前提下测试端点响应。

相关代码片段 #

/main.py #

async def authenticate(
    apitoken: Optional[str] = Header(""),
    account: Optional[str] = Header(""),
    sessionHost: Optional[str] = Header(""),
):

    print(sessionHost)
    if (
        sessionHost
        and sessionHost[-20:] == "mydomain-staging.com"
        or sessionHost == "https://apistaging.mydomain.com"
    ):
        url = sessionHost + "/api/jml/templates"
    else:
        url = PF_API_URL + "/jml/templates"

    print(url)

    params = {"per_page": 1, "page": 1}
    accept = "application/vnd.mydomain+json;version=2"
    auth_headers = {
        "content_type": "application/json",
        "accept": accept,
        "account": account,
        "apitoken": apitoken,
    }
    r = await session.get(url, headers=auth_headers, params=params, timeout=30)

    if r.status_code != 200:
        raise HTTPException(status_code=401, detail="Unauthorized")

    app.token_cache[apitoken] = {
        "account": account,
        "ts": datetime.now(),
    }


app.include_router(
    stats.router,
    dependencies=[Depends(authenticate)],
    responses={404: {"description": "Not found"}},
)

/routers/stats.py #

@router.get("/roles")
async def roles(
    request: Request,
    account: str = Header(...),
    start_date: str = None,
    end_date: str = None,
    profile_id: int = 0,
):

    custom_fields, start_date, end_date = parse_params(
        request, start_date, end_date
    )
    new_role = Roles(
        account, start_date, end_date, profile_id, custom_fields=custom_fields
    )

    try:
        await new_role.setup()
        result = await new_role.get_json()
    except NoDataException:
        result = {"Error": "No Data"}

    return result

要Mock掉认证流程，核心是用unittest.mock.patch替换main.py里的authenticate函数，跳过真实的HTTP请求和权限校验，直接假设认证成功。

步骤1：调整假认证函数 #

原authenticate函数没有返回值，仅通过是否抛出HTTPException判断认证结果，因此修改假认证函数，只需跳过所有逻辑不抛异常即可：

from typing import Optional
from fastapi import Header

async def fake_authenticate(
    apitoken: Optional[str] = Header(""),
    account: Optional[str] = Header(""),
    sessionHost: Optional[str] = Header(""),
):
    # 模拟认证成功，跳过所有校验逻辑
    # 若需要模拟token缓存，可以添加以下代码
    # from main import app
    # from datetime import datetime
    # app.token_cache[apitoken] = {"account": account, "ts": datetime.now()}
    pass

步骤2：在测试中Patch认证函数 #

在原有测试代码中新增patch装饰器，将main.authenticate替换为假认证函数：

import pytest
from fastapi.testclient import TestClient
from unittest.mock import patch, AsyncMock
from main import app
from tests.mock_data import test_roles_data
from typing import Optional
from fastapi import Header

class MockRoles:
    async def setup(self):
        pass

    async def get_json(self):
        return test_roles_data

@pytest.fixture
def client():
    with TestClient(app) as c:
        yield c

async def fake_authenticate(
    apitoken: Optional[str] = Header(""),
    account: Optional[str] = Header(""),
    sessionHost: Optional[str] = Header(""),
):
    pass

@patch('main.authenticate', side_effect=fake_authenticate)
@patch('routers.base.parse_params', return_value=('custom_fields', '2023-01-01', '2023-01-31'))
@patch('services.roles.Roles', return_value=MockRoles())
def test_roles_endpoint(mock_auth, mock_parse_params, mock_roles, client):
    response = client.get("/roles", headers={"account": "test_account"})

    assert response.status_code == 200
    assert response.json() == test_roles_data

步骤3：可选优化——用Fixture复用Mock #

如果多个测试需要Mock认证，可以将Patch逻辑封装为pytest fixture：

@pytest.fixture
def mock_authenticate():
    async def fake_auth(
        apitoken: Optional[str] = Header(""),
        account: Optional[str] = Header(""),
        sessionHost: Optional[str] = Header(""),
    ):
        pass
    with patch('main.authenticate', side_effect=fake_auth):
        yield

@pytest.fixture
def client():
    with TestClient(app) as c:
        yield c

@patch('routers.base.parse_params', return_value=('custom_fields', '2023-01-01', '2023-01-31'))
@patch('services.roles.Roles', return_value=MockRoles())
def test_roles_endpoint(mock_parse_params, mock_roles, client, mock_authenticate):
    response = client.get("/roles", headers={"account": "test_account"})

    assert response.status_code == 200
    assert response.json() == test_roles_data

关键注意事项 #

• Patch路径必须是被依赖注入使用的路径：因为路由依赖的是main.authenticate，所以必须Patchmain.authenticate而非其他路径。
• 假认证函数需与原函数参数签名完全一致，包括Header参数的定义，否则FastAPI依赖注入会报错。

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