如何在 FastAPI 应用程序中向 pytest 添加身份验证模拟？

问：

我正在使用 pytest 测试 FastAPI 应用程序，并且需要集成身份验证模拟。我有一个需要身份验证的端点，但我不确定如何在我的 pytest 设置中正确添加身份验证模拟。/roles

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',
    }

我的目标是模拟身份验证过程，以便在没有实际身份验证的情况下可以测试端点。有人可以指导我如何正确地将此函数添加到我的pytest设置中吗？/rolesfake_authenticate

附加信息：

• 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"}},
)

/路由器/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