测试
测试客户端允许您使用 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!"