Skip to content

测试

测试客户端允许您使用 httpx 库针对您的 ASGI 应用程序发出请求。

from starlette.responses import HTMLResponse
from starlette.testclient import TestClient


async def app(scope, receive, send):
    assert scope['type'] == 'http'
    response = HTMLResponse('<html><body>Hello, world!</body></html>')
    await response(scope, receive, send)


def test_app():
    client = TestClient(app)
    response = client.get('/')
    assert response.status_code == 200

测试客户端所暴露的接口与任何其他 httpx 会话的接口相同。特别要注意的是,发出请求的调用只是标准函数调用,而不是可等待对象。

您可以使用任何 httpx 标准 API,例如身份验证、会话 Cookie 处理或文件上传。

例如,要在 TestClient 上设置标头,您可以这样做:

client = TestClient(app)

# Set headers on the client for future requests
client.headers = {"Authorization": "..."}
response = client.get("/")

# Set headers for each request separately
response = client.get("/", headers={"Authorization": "..."})

例如,使用 TestClient 发送文件:

client = TestClient(app)

# Send a single file
with open("example.txt", "rb") as f:
    response = client.post("/form", files={"file": f})

# Send multiple files
with open("example.txt", "rb") as f1:
    with open("example.png", "rb") as f2:
        files = {"file1": f1, "file2": ("filename", f2, "image/png")}
        response = client.post("/form", files=files)

如需更多信息,您可以查看 httpx 文档。

默认情况下, TestClient 会引发应用程序中出现的任何异常。偶尔,您可能想要测试 500 错误响应的内容,而不是允许客户端引发服务器异常。在这种情况下,您应该使用 client = TestClient(app, raise_server_exceptions=False)

注意

If you want the TestClient to run the lifespan handler, you will need to use the TestClient as a context manager. It will not be triggered when the TestClient is instantiated. You can learn more about it here.

选择异步后端

TestClient 接受参数 backend (一个字符串)和 backend_options (一个字典)。这些选项会被传递给 anyio.start_blocking_portal() 。有关可接受的后端选项的更多信息,请参阅 anyio 文档。默认情况下,使用 asyncio 及其默认选项。

抱歉,原文中的“ Trio ”和“ backend="trio" ”含义不明,无法准确进行翻译。如果您能提供更多关于这两个符号的信息,我将尽力为您提供更准确的翻译。

def test_app()
    with TestClient(app, backend="trio") as client:
       ...

要使用 uvloop 运行 asyncio ,需传递 backend_options={"use_uvloop": True} 。例如:

def test_app()
    with TestClient(app, backend_options={"use_uvloop": True}) as client:
       ...

测试 WebSocket 会话

您还可以使用测试客户端测试 WebSocket 会话。

httpx 库将用于构建初始握手,这意味着您可以在 http 和 websocket 测试之间使用相同的身份验证选项和其他标头。

from starlette.testclient import TestClient
from starlette.websockets import WebSocket


async def app(scope, receive, send):
    assert scope['type'] == 'websocket'
    websocket = WebSocket(scope, receive=receive, send=send)
    await websocket.accept()
    await websocket.send_text('Hello, world!')
    await websocket.close()


def test_app():
    client = TestClient(app)
    with client.websocket_connect('/') as websocket:
        data = websocket.receive_text()
        assert data == 'Hello, world!'

会话上的操作是标准函数调用,而非可等待对象。

在上下文管理的 with 块内使用会话很重要。这确保了 ASGI 应用程序所在的后台线程能正确终止,并且应用程序内发生的任何异常都会始终由测试客户端引发。

建立一个测试会话

  • .websocket_connect(url, subprotocols=None, **options) - 与 httpx.get() 采用相同的一组参数。

如果应用程序不接受 WebSocket 连接,可能会引发 starlette.websockets.WebSocketDisconnect

websocket_connect() 必须用作上下文管理器(在 with 块中)。

!!! 注意 params 参数不受 websocket_connect 支持。如果您需要传递查询参数,请直接在 URL 中硬编码。

```python
with client.websocket_connect('/path?foo=bar') as websocket:
    ...
```

发送数据

  • .send_text(data) - 将给定文本发送到应用程序。
  • .send_bytes(data) - 将给定的字节发送到应用程序。
  • .send_json(data, mode="text") - 将给定数据发送到应用程序。使用 mode="binary" 通过二进制数据帧发送 JSON。

接收数据

  • .receive_text() - 等待应用程序发送的传入文本并返回它。
  • .receive_bytes() - 等待应用程序发送的传入字节串并返回它。
  • .receive_json(mode="text") - 等待应用程序发送的传入 json 数据并返回。使用 mode="binary" 通过二进制数据帧接收 JSON。

该文本“May raise starlette.websockets.WebSocketDisconnect .”似乎不太符合常规的语义表达,难以进行准确的中文翻译。如果您有更多的上下文信息或确认文本内容的准确性,我将尽力为您提供更满意的翻译。目前按照字面直译的话,只能译为:“可能提高 starlette.websockets.WebSocketDisconnect 。”但这个翻译可能不太符合实际意义。

关闭连接

  • .close(code=1000) - 执行客户端对 WebSocket 连接的关闭操作。

异步测试

有时您会想要在应用程序之外执行异步操作。例如,您可能希望在使用现有的异步数据库客户端/基础设施调用您的应用程序后检查数据库的状态。

对于这些情况,使用 TestClient 是困难的,因为它会创建自己的事件循环,并且异步资源(如数据库连接)通常无法在事件循环之间共享。解决此问题的最简单方法是使您的整个测试成为异步的,并使用异步客户端,如 httpx.AsyncClient 。

这里有一个这样的测试的例子:

from httpx import AsyncClient
from starlette.applications import Starlette
from starlette.routing import Route
from starlette.requests import Request
from starlette.responses import PlainTextResponse


def hello(request: Request) -> PlainTextResponse:
    return PlainTextResponse("Hello World!")


app = Starlette(routes=[Route("/", hello)])


# if you're using pytest, you'll need to to add an async marker like:
# @pytest.mark.anyio  # using https://github.com/agronholm/anyio
# or install and configure pytest-asyncio (https://github.com/pytest-dev/pytest-asyncio)
async def test_app() -> None:
    # note: you _must_ set `base_url` for relative urls like "/" to work
    async with AsyncClient(app=app, base_url="http://testserver") as client:
        r = await client.get("/")
        assert r.status_code == 200
        assert r.text == "Hello World!"