如何为FastAPI端点创建单元测试，使请求到另一个端点?

Question

我有一个FastAPI应用程序，它在处理特定请求的函数中向其他端点发出请求。

如何使用fastapi.testclient.TestClient为这个端点构建单元测试?

import fastapi
import requests
import os
app = fastapi.FastAPI()
# in production this will be a k8s service to allow for 
# automatic scaling and load balancing.
# The default value works when the app is run using uvicorn
service_host = os.environ.get('SERVICE_HOST', 'localhost:8000')
@app.post('/spawner')
def spawn():
# an endpoint that makes multiple requests to set of worker endpoints
for i in range(4):
url = 'http://'+service_host+f'/create/{i}'
requests.post(url)
return {'msg': 'spawned the creation/update of 4 resources'}
@app.post('/create/{index}')
def worker(index: int):
# an endpoint that does the actual work of creating/updating the resource
return {'msg': f'Created/updated resource {index}'}

Answer 1

如何编写单元测试取决于您想要检查的具体内容。

例如，从你的示例代码中，如果你只是想检查你的/spawner端点正确地调用你的/create端点一定次数，你可以使用Python的unittest.mock.patch来修补requests.post调用，这样它就不会进行实际的调用，但是你可以检查它会进行的调用。(参见我如何模拟请求和响应?有关如何模拟requests库进行的外部调用的示例。)

您仍然使用FastAPI的TestClient来调用端点，但是当.post调用被修补时:

from fastapi.testclient import TestClient
from main import app
from unittest.mock import patch
def test_spawn():
client = TestClient(app)
mocked_response = requests.Response()
mocked_response.status_code = 200
with patch("requests.post", return_value=mocked_response) as mocked_request:
response = client.post("/spawner")
# Expect that the requests.post was called 4 times
# and it called the /create/{index} endpoint
assert mocked_request.call_count == 4
called_urls = [call.args[0] for call in mocked_request.call_args_list]
assert called_urls[0].endswith("/create/0")
assert called_urls[1].endswith("/create/1")
assert called_urls[2].endswith("/create/2")
assert called_urls[3].endswith("/create/3")
# Expect to get the msg
assert response.status_code == 200
assert response.json() == {"msg": "spawned the creation/update of 4 resources"}

patch返回一个Mock对象，它的属性像call_count告诉你requests.post函数被调用了多少次，call_args_list存储每次调用requests.post的参数。然后，您的测试还可以断言，如果一切都按照预期进行，那么它应该返回预期的响应。

现在，回到我所说的，这个测试的有用性很大程度上取决于特别是您希望从编写测试中获得什么。这个测试的一个问题是，它不是严格的黑盒测试，因为测试"知道";端点应该如何在内部工作。例如，如果要点是要确保使用某些查询参数?count=N调用/spawner会产生N数量的生成资源，那么可能这种测试是有用的。 另一件事，这种测试是有意义的，如果你让实际的外部调用一些其他API，而不是你的自己的端点。用户JarroVGit在评论中提到，一个更好的实现是，你可以重构/spawner，直接调用端点函数，并将传递给/create响应:

@app.post("/spawner")
def spawn():
# an endpoint that makes multiple requests to set of worker endpoints
results = [worker(i) for i in range(4)]
return {
"msg": "spawned the creation/update of 4 resources",
"results": results,
}
@app.post("/create/{index}")
def worker(index: int):
# an endpoint that does the actual work of creating/updating the resource
worker_inst.create(index)
return {"msg": f"Created/updated resource {index}"}

可以通过将/create中的内容修补为而不是创建实际资源(在这里，我假设某种worker_inst.create(index)执行实际创建)来使测试更简单:

def test_spawn():
client = TestClient(app)
# Patch the actual create/update resources to *not* create anything
with patch.object(WorkerInst, "create"):
response = client.post("/spawner")
# Expect to get the msg
assert response.status_code == 200
assert response.json() == {
"msg": "spawned the creation/update of 4 resources",
"results": [
{"msg": "Created/updated resource 0"},
{"msg": "Created/updated resource 1"},
{"msg": "Created/updated resource 2"},
{"msg": "Created/updated resource 3"},
],
}

我不知道是什么样的"资源"。你的应用程序正在创建，但你可以将该单元测试转换为更有用的集成测试，通过删除worker的patch-ing，只是让工作器创建资源，然后让测试断言1)正确的资源被创建，2)/spawner端点返回预期的响应。同样，这取决于"资源"是什么。和测试的目的。