如何模拟StreamingRespone？

Question

想象有一个具有文件结构的 FastAPI 应用程序：

src:
- app.py
- responses.py
- test_download.py

我退回定制的

StreamingRespone

：

# responses.py

from fastapi.responses import StreamingRespone


class CustomStreamingRespone(StreamingRespone):
    def __init__(self, url):
        self._url = url

我在视图中返回

CustomStreamingRespone

：

# app.py

from fastapi import FastAPI


app = FastAPI()


@app.get("/act/download/")
async def download_act(...):
    # search for url...
    return CustomStreamingRespone(url=url)


@app.get("/report/download/")
async def download_report(...):
    # search for url...
    return CustomStreamingRespone(url=url)


# there might be multiple views that return CustomStreamingRespone

我正在尝试像这样嘲笑

CustomStreamingRespone

（与

pytest

）：

# test_download.py

async def test_act_downloaded(client: httpx.AsyncClient, mocker: MockerFixture):
    # preparing data for test might be identical in part of preparing content inside CustomStreamingRespone
    mock = mocker.patch("responses.CustomStreamingRespone")
    resp = await client.get("/act/download")
    mock.assert_called()  # fails with error AssertionError: Expected 'CustomStreamingRespone' to have been called. 
    assert resp == mock.return_value


async def test_report_downloaded(client: httpx.AsyncClient, mocker: MockerFixture):
    ...

但这行不通。

我只需要测试

CustomStreamingRespone

是否被使用（使用的事实）。

测试视图时如何模拟

CustomStreamingRespone

？

附注我刚刚意识到我应该模拟

CustomStreamingRespone.__call__

方法？

Answer 1

您需要首先将 CustomStreamingResponse 导入到您的 app.py 中。这应该是文件：

from fastapi import FastAPI
from responses import CustomStreamingRespone

app = FastAPI()


@app.get("/act/download/")
async def download_act(...):
    # search for url...
    return CustomStreamingRespone(url=url)

此外，您需要调整应用程序的入口点。我不确定您的入口点是什么，因此我导入相关课程的方式可能适合您，也可能不适合您。

然后您需要模拟应用程序本身。现在到了棘手的一点，要模拟函数或类，您需要模拟从文件内部导入到文件本身的实例。因此，请注意下面代码中模拟的导入路径（再次根据您的入口点进行调整）。

from app import app
from httpx import AsyncClient
import pytest

@pytest.mark.asyncio
@mock.patch("app.CustomStreamingRespone")
async def test_download_act(mock_custom_streaming_response):
    url = "abc"
    mock_custom_streaming_response_instance = (
        mock_custom_streaming_response.return_value
    )
    mock_custom_streaming_response_instance.return_value = "abc"

    async with AsyncClient(app=app, base_url="http://test") as ac:
        response = await ac.get("/act/download/")

    assert response == mock_custom_streaming_response_instance.return_value
    mock_custom_streaming_response.assert_called_once()

这里有几件事需要您注意。由于您的端点是异步的，因此您需要 pytest asyncio 装饰器。