企业官网建设流程全解析

Litestar 测试工具链完全指南：TestClient、RequestFactory 与 WebSocket 测试实践

【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestar

本文围绕 Litestar 框架内置的litestar.testing模块展开，系统讲解其提供的同步/异步测试客户端、create_test_client快捷函数、RequestFactory请求工厂、WebSocket 测试会话、子进程真实服务器客户端以及生命周期处理器等核心工具。读者阅读后将掌握为 Litestar 应用编写单元测试与集成测试的完整方法，包括如何选择测试客户端、如何隔离测试应用、如何模拟请求与会话数据，以及如何在真实服务器环境下验证 SSE 等流式行为。

litestar.testing 模块总览

Litestar 的测试工具全部集中在litestar/testing/__init__.py中对外暴露，其核心成员包括：

从源码入口 litestar/testing/init.py 可以看到，litestar.testing基于httpx构建，httpx并非默认依赖，而是包含在testingextra 中。若未安装，导入时会在 litestar/testing/init.py 处抛出MissingDependencyException。

安装与前置条件

httpx需要通过testingextra 安装：

pip install 'litestar[testing]'

安装完成后，即可从litestar.testing导入所有测试工具。测试工具与 httpx 客户端保持一致的接口风格，可无缝融入 pytest 等测试框架。

同步与异步测试客户端

Litestar 提供两个测试客户端，两者的差别在于运行方式而非功能：

• TestClient：同步客户端，在新建的独立线程中创建一个事件循环并运行应用。适合不需要测试异步行为、测试框架也未提供外部事件循环的场景。其实现位于 litestar/testing/client/sync_client.py，内部通过anyio.from_thread.start_blocking_portal启动一个BlockingPortal来桥接同步与异步世界。
• AsyncTestClient：异步客户端，在外部托管的事件循环上同时运行应用与客户端，见 litestar/testing/client/async_client.py。适合测试异步行为、或测试环境中存在异步资源（如异步数据库连接）的场景。

一个健康检查端点的两种测试写法

假设我们有如下应用（对应仓库示例 docs/examples/testing/test_health_check_sync.py）：

from litestar import Litestar, MediaType, get @get(path="/health-check", media_type=MediaType.TEXT, sync_to_thread=False) def health_check() -> str: return "healthy" app = Litestar(route_handlers=[health_check], debug=True)

同步测试使用TestClient：

from litestar.status_codes import HTTP_200_OK from litestar.testing import TestClient from my_app.main import app def test_health_check(): with TestClient(app=app) as client: response = client.get("/health-check") assert response.status_code == HTTP_200_OK assert response.text == "healthy"

异步测试使用AsyncTestClient：

from litestar.status_codes import HTTP_200_OK from litestar.testing import AsyncTestClient from my_app.main import app async def test_health_check(): async with AsyncTestClient(app=app) as client: response = await client.get("/health-check") assert response.status_code == HTTP_200_OK assert response.text == "healthy"

两个客户端都是上下文管理器，__enter__/__aenter__阶段会通过LifeSpanHandler触发应用的 startup，退出时触发 shutdown，确保on_startup/on_shutdown生命周期钩子被正确执行（见 litestar/testing/client/sync_client.py）。

如何选择测试客户端

多数情况下两者的功能等价，选择取决于个人偏好。但有一个关键差异需要特别注意：事件循环归属。

TestClient在独立线程的新事件循环中运行应用，而测试本身可能运行在另一个事件循环（如 pytest-asyncio 或 anyio pytest 插件提供的事件循环）中。当测试中存在跨事件循环共享的异步资源时，就会出现问题。仓库中的示例 docs/examples/testing/async_resource_test_issue.py 展示了典型故障：共享的httpx.AsyncClient在一个 fixture 的事件循环（事件循环 A）中创建，但请求经由TestClient时连接被绑定到应用所在的事件循环 B；测试结束后 B 先关闭，fixture 清理阶段在 A 中调用aclose()时便会抛出RuntimeError: Event is closed。

解决办法是改用AsyncTestClient（见 docs/examples/testing/async_resource_test_issue_fix.py），让 fixture、测试与应用运行在同一个事件循环中，资源即可正常回收。总结一句：只要测试环境本身提供了事件循环（异步测试），就应优先使用AsyncTestClient。

将测试客户端做成 pytest fixture

由于客户端在多个测试中复用，推荐封装为 fixture（见 docs/examples/testing/test_health_check_sync.py）：

import pytest from litestar import Litestar from litestar.testing import TestClient from my_app.main import app @pytest.fixture(scope="function") def test_client() -> Iterator[TestClient[Litestar]]: with TestClient(app=app) as client: yield client def test_health_check_with_fixture(test_client: TestClient[Litestar]) -> None: response = test_client.get("/health-check") assert response.status_code == HTTP_200_OK assert response.text == "healthy"

异步版本只需将 fixture 改为async def并配合AsyncTestClient。

create_test_client：一行代码创建隔离测试应用

create_test_client与其异步版本create_async_test_client（实现于 litestar/testing/helpers.py 与 litestar/testing/helpers.py）会先基于传入的路由处理器构造一个独立的Litestar实例，再为其创建测试客户端。适合验证与应用实例无关的通用逻辑，或希望将端点隔离测试的场景。

from litestar.status_codes import HTTP_200_OK from litestar.testing import create_test_client from my_app.main import health_check def test_health_check(): with create_test_client([health_check]) as client: response = client.get("/health-check") assert response.status_code == HTTP_200_OK assert response.text == "healthy"

这个函数的核心价值在于：它把Litestar(app, **kwargs)的全部构造参数（middleware、guards、dependencies、on_startup、cors_config、csrf_config、session_config、exception_handlers、plugins、signature_namespace等）直接暴露为函数参数，使你可以在不修改生产应用代码的前提下，为测试临时注入中间件、守卫、依赖或自定义request_class/response_class。例如设置debug=True获取带堆栈的 HTML 错误页，或传入pdb_on_exception=True在异常时进入 PDB 调试器。

需要特别说明的是，create_test_client是上下文管理器，必须用with调用，否则 async 的 startup/shutdown 无法被正确触发（源码 docstring 中对此有明确说明，见 litestar/testing/helpers.py）。

RequestFactory：不经过网络构造 Request 对象

RequestFactory（实现于 litestar/testing/request_factory.py）用于直接创建Request实例，绕过真实的 HTTP 往返。它最适合对接收Request对象的纯逻辑进行单元测试，例如守卫（guard）、依赖函数或自定义的请求处理逻辑。

构造参数

RequestFactory提供get、post、put、patch、delete五个方法（见 litestar/testing/request_factory.py），它们共享一组参数：path、headers、cookies、session、user、auth、query_params、state、path_params、http_version、route_handler；带请求体的方法还额外支持request_media_type（默认RequestEncodingType.JSON，可选MULTI_PART与URL_ENCODED）和data。

基础用法

from litestar import Litestar from litestar.enums import RequestEncodingType from litestar.testing import RequestFactory my_app = Litestar(route_handlers=[]) my_server = "litestar.org" # GET 请求 + 查询参数 query_params = {"id": 1} get_user_request = RequestFactory(app=my_app, server=my_server).get( "/person", query_params=query_params ) # POST 请求 + JSON 数据 create_user_request = RequestFactory(app=my_app, server=my_server).post( "/person", data=person ) # 携带自定义头 request_with_header = RequestFactory(app=my_app, server=my_server).get( "/person", query_params=query_params, headers={"header1": "value1"} ) # 指定 multipart 媒体类型 request_with_media_type = RequestFactory(app=my_app, server=my_server).post( "/person", data=person, request_media_type=RequestEncodingType.MULTI_PART )

实战：隔离测试守卫函数

以守卫函数为例。假设我们有如下应用代码：

from litestar import Request from litestar.exceptions import NotAuthorizedException from litestar.handlers.base import BaseRouteHandler def secret_token_guard(request: Request, route_handler: BaseRouteHandler) -> None: if ( route_handler.opt.get("secret") and not request.headers.get("Secret-Header", "") == route_handler.opt["secret"] ): raise NotAuthorizedException()

以及使用该守卫的端点：

from litestar import get from my_app.guards import secret_token_guard @get(path="/secret", guards=[secret_token_guard], opt={"secret": "super-secret"}) def secret_endpoint() -> None: ...

使用RequestFactory即可在不启动服务器的情况下直接验证守卫的两种场景：

import pytest from litestar.exceptions import NotAuthorizedException from litestar.testing import RequestFactory from my_app.guards import secret_token_guard from my_app.secret import secret_endpoint request = RequestFactory().get("/") def test_secret_token_guard_failure_scenario(): copied_endpoint_handler = secret_endpoint.copy() copied_endpoint_handler.opt["secret"] = None with pytest.raises(NotAuthorizedException): secret_token_guard(request=request, route_handler=copied_endpoint_handler) def test_secret_token_guard_success_scenario(): copied_endpoint_handler = secret_endpoint.copy() copied_endpoint_handler.opt["secret"] = "super-secret" secret_token_guard(request=request, route_handler=copied_endpoint_handler)

底层实现要点

从源码看，RequestFactory的请求构造分为两条路径：

• 无请求体的方法（get/delete）：直接调用_create_scope构建 ASGI scope，再将 headers 编码后构造Request(scope=scope)（见 litestar/testing/request_factory.py）。
• 带请求体的方法（post/put/patch）：经_create_request_with_data处理（见 litestar/testing/request_factory.py）。数据首先被序列化为 JSON，然后根据request_media_type选择 httpx 的encode_json、encode_multipart_data或encode_urlencoded_data生成编码后的 body 流，并把Content-Type等编码头合并进请求头；body 最终写入 scope state 的body字段。

scope 中还注入了session、user、auth、state、path_params等键，因此request.session、request.user、request.auth在测试中可以直接读取或断言，无需经过会话中间件。

WebSocket 测试

Litestar 在 httpx 客户端之上扩展了 WebSocket 支持。通过客户端的websocket_connect方法建立连接（内部实现见 litestar/testing/client/sync_client.py 与 litestar/testing/client/async_client.py），该方法在底层 ASGI 通信抛出ConnectionUpgradeExceptionError（表示升级成功）后返回一个 WebSocket 测试会话对象。

会话对象（同步版WebSocketTestSession与异步版AsyncWebSocketTestSession，定义于 litestar/testing/websocket_test_session.py）提供成对的收发 API：

• 发送：send、send_text、send_bytes、send_json、send_msgpack、close
• 接收：receive、receive_text、receive_bytes、receive_json、receive_msgpack
• 连接信息：accepted_subprotocol、extra_headers、scope

一个完整的 WebSocket 往返测试（对应仓库示例 docs/examples/testing/test_websocket_sync.py）：

from typing import Any from litestar import WebSocket, websocket from litestar.testing import create_test_client def test_websocket() -> None: @websocket(path="/ws") async def websocket_handler(socket: WebSocket[Any, Any, Any]) -> None: await socket.accept() recv = await socket.receive_json() await socket.send_json({"message": recv}) await socket.close() with create_test_client(route_handlers=[websocket_handler]) as client, client.websocket_connect("/ws") as ws: ws.send_json({"hello": "world"}) data = ws.receive_json() assert data == {"message": {"hello": "world"}}

websocket_connect同样支持subprotocols、params、headers、cookies、auth、timeout等参数，可用于验证自定义连接头与子协议协商。

会话数据的注入与读取

当应用使用 Session 中间件时，测试中常常需要绕过 HTTP 流程直接注入或检查会话内容。两个客户端为此提供了set_session_data与get_session_data方法（见 litestar/testing/client/sync_client.py 与 litestar/testing/client/async_client.py），前提是在构造客户端时传入session_config。

异步版本示例（源码 docstring 中的用例）：

from litestar import Litestar, post from litestar.middleware.session.memory_backend import MemoryBackendConfig session_config = MemoryBackendConfig() @post(path="/test") def set_session_data(request: Request) -> None: request.session["foo"] = "bar" app = Litestar(route_handlers=[set_session_data], middleware=[session_config.middleware]) async with AsyncTestClient(app=app, session_config=session_config) as client: await client.post("/test") assert await client.get_session_data() == {"foo": "bar"}

反向用例——先注入再请求：

async with AsyncTestClient(app=app, session_config=session_config) as client: await client.set_session_data({"foo": "bar"}) assert await client.get("/test").json() == {"foo": "bar"}

从实现上看，session_config会在客户端初始化时被解析为对应的BaseSessionBackend实例（litestar/testing/client/sync_client.py），会话数据经由 blocking portal 或直接异步调用写入/读取后端，因此可以配合任意 Litestar 支持的会话后端使用。

在同步客户端上运行异步代码：blocking_portal

同步TestClient在独立线程中运行事件循环，这个桥接机制基于anyio.BlockingPortal。该 portal 被客户端以blocking_portal属性公开（见 litestar/testing/client/sync_client.py），因此可以在同步测试中执行并等待任意异步函数，让它们运行在与应用相同的循环上。

仓库示例 docs/examples/testing/test_with_portal.py 展示了两种用法：

from concurrent.futures import Future, wait import anyio from litestar.testing import create_test_client def test_with_portal() -> None: async def get_float(value: float) -> float: await anyio.sleep(value) return value with create_test_client(route_handlers=[]) as test_client: # 1) 启动后台任务 future: Future[float] = test_client.blocking_portal.start_task_soon(get_float, 0.25) # 2) 同步阻塞地调用异步函数 assert test_client.blocking_portal.call(get_float, 0.1) == 0.1 wait([future]) assert future.result() == 0.25

blocking_portal.call用于同步等待某个异步调用完成；start_task_soon则把任务放入后台并发执行并返回Future，适合在测试中模拟并发的后台工作。

子进程真实服务器：subprocess_sync_client 与 subprocess_async_client

测试客户端的默认模式是让 httpx 直接调用 ASGI 应用（走内存中的 transport），不监听真实端口。这在绝大多数场景下足够，但存在局限：例如带有无限生成器的 Server-Sent Events（SSE）端点，httpx 会等待整个响应体读取完毕才返回，导致测试客户端卡死。

此时应使用subprocess_sync_client/subprocess_async_client（实现于 litestar/testing/client/subprocess_client.py）。它们的工作流程是：

1. _get_available_port通过绑定("localhost", 0)获取一个空闲端口；
2. 以litestar --app <app> run --port <port>命令在子进程中启动真实服务器；
3. 轮询探测（默认重试 100 次、间隔 1 秒，可通过retry_count与retry_timeout调整）直至服务器可访问；
4. 返回绑定该地址的 httpx 客户端；退出上下文时终止子进程。

用法示例（对应仓库示例 docs/examples/testing/test_subprocess_sse.py）：

@pytest.fixture(name="async_client") async def fx_async_client() -> AsyncIterator[httpx.AsyncClient]: async with subprocess_async_client(workdir=ROOT, app="subprocess_sse_app:app", capture_output=True) as client: yield client

参数说明：

• workdir：应用模块所在的工作目录（子进程命令的执行目录）；
• app：可解析的应用路径字符串，如"my_app:application"；
• capture_output：默认True，子进程输出会透传到主进程 stdout/stderr；设为False则丢弃输出，适合测试输出繁杂的场景。

若应用在给定重试次数内未能启动，会抛出StartupError（定义于 litestar/testing/client/subprocess_client.py）。

LifeSpanHandler：驱动 ASGI 生命周期协议

LifeSpanHandler（定义于 litestar/testing/life_span_handler.py）是测试基础设施的底层组件：它构造两个内存对象流（stream_send/stream_receive），在进入上下文时向应用发送lifespan.startup事件并等待lifespan.startup.complete，退出时发送lifespan.shutdown并等待lifespan.shutdown.complete，从而完整驱动 ASGI 的 lifespan 协议。

两个测试客户端在进入/退出上下文时都通过它触发应用的启动与关闭（见 litestar/testing/client/sync_client.py），这意味着测试天然覆盖了on_startup/on_shutdown钩子以及使用lifespan参数的异步上下文管理器。如果你需要自己实现基于 ASGI 协议的测试工具，也可以直接复用它。

补充：BaseTestClient 与测试传输层

参考文档还列出了BaseTestClient。从源码结构看，两个客户端共享的基类逻辑位于 litestar/testing/client/_base.py，包括_get_session_data、_set_session_data以及_prepare_ws_connect_request等内部辅助函数；而SyncTestClientTransport/TestClientTransport则位于 litestar/testing/transport.py，负责把 httpx 请求转换为 ASGI scope 并驱动应用执行。理解这层设计有助于定位"客户端如何把 HTTP 请求翻译成 ASGI 调用"这一核心机制：无需真实 socket，所有通信都在进程内完成，这也是测试速度快的原因之一。

总结与选型建议

Litestar 的测试工具链以 httpx 为底座、以 ASGI 协议为内核对齐，覆盖了从纯单元测试到真实服务器集成测试的完整谱系。建议将 docs/usage/testing.rst 作为入门指南、docs/reference/testing.rst 作为 API 速查，并结合仓库中的示例目录 docs/examples/testing/ 与单元测试 tests/unit/ 中的实际用例进一步深入。

【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestar