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中对外暴露其核心成员包括成员类型用途TestClient同步测试客户端在独立线程的新事件循环中运行应用AsyncTestClient异步测试客户端在外部托管的事件循环上运行应用与客户端create_test_client快捷函数先创建Litestar应用再用TestClient包裹create_async_test_client快捷函数先创建Litestar应用再用AsyncTestClient包裹RequestFactory请求工厂直接构造Request实例无需真实网络WebSocketTestSession/AsyncWebSocketTestSessionWebSocket 测试会话模拟 WebSocket 连接的收发subprocess_sync_client/subprocess_async_client子进程客户端在子进程中启动真实服务器并返回 httpx 客户端LifeSpanHandler生命周期处理器驱动 ASGI 应用的lifespan协议从源码入口 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.pyfrom litestar import Litestar, MediaType, get get(path/health-check, media_typeMediaType.TEXT, sync_to_threadFalse) def health_check() - str: return healthy app Litestar(route_handlers[health_check], debugTrue)同步测试使用TestClientfrom litestar.status_codes import HTTP_200_OK from litestar.testing import TestClient from my_app.main import app def test_health_check(): with TestClient(appapp) as client: response client.get(/health-check) assert response.status_code HTTP_200_OK assert response.text healthy异步测试使用AsyncTestClientfrom 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(appapp) 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.pyimport pytest from litestar import Litestar from litestar.testing import TestClient from my_app.main import app pytest.fixture(scopefunction) def test_client() - Iterator[TestClient[Litestar]]: with TestClient(appapp) 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。例如设置debugTrue获取带堆栈的 HTML 错误页或传入pdb_on_exceptionTrue在异常时进入 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、依赖函数或自定义的请求处理逻辑。构造参数参数默认值说明app空Litestar实例设置到request.scope[litestar_app]servertest.org服务器域名port3000服务器端口root_path服务器根路径schemehttp协议 schemehandler_kwargsNone传给为请求创建的路由处理器的 kwargsRequestFactory提供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(appmy_app, servermy_server).get( /person, query_paramsquery_params ) # POST 请求 JSON 数据 create_user_request RequestFactory(appmy_app, servermy_server).post( /person, dataperson ) # 携带自定义头 request_with_header RequestFactory(appmy_app, servermy_server).get( /person, query_paramsquery_params, headers{header1: value1} ) # 指定 multipart 媒体类型 request_with_media_type RequestFactory(appmy_app, servermy_server).post( /person, dataperson, request_media_typeRequestEncodingType.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(requestrequest, route_handlercopied_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(requestrequest, route_handlercopied_endpoint_handler)底层实现要点从源码看RequestFactory的请求构造分为两条路径无请求体的方法get/delete直接调用_create_scope构建 ASGI scope再将 headers 编码后构造Request(scopescope)见 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.pyfrom 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(appapp, session_configsession_config) as client: await client.post(/test) assert await client.get_session_data() {foo: bar}反向用例——先注入再请求async with AsyncTestClient(appapp, session_configsession_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.25blocking_portal.call用于同步等待某个异步调用完成start_task_soon则把任务放入后台并发执行并返回Future适合在测试中模拟并发的后台工作。子进程真实服务器subprocess_sync_client 与 subprocess_async_client测试客户端的默认模式是让 httpx 直接调用 ASGI 应用走内存中的 transport不监听真实端口。这在绝大多数场景下足够但存在局限例如带有无限生成器的 Server-Sent EventsSSE端点httpx 会等待整个响应体读取完毕才返回导致测试客户端卡死。此时应使用subprocess_sync_client/subprocess_async_client实现于 litestar/testing/client/subprocess_client.py。它们的工作流程是_get_available_port通过绑定(localhost, 0)获取一个空闲端口以litestar --app app run --port port命令在子进程中启动真实服务器轮询探测默认重试 100 次、间隔 1 秒可通过retry_count与retry_timeout调整直至服务器可访问返回绑定该地址的 httpx 客户端退出上下文时终止子进程。用法示例对应仓库示例 docs/examples/testing/test_subprocess_sse.pypytest.fixture(nameasync_client) async def fx_async_client() - AsyncIterator[httpx.AsyncClient]: async with subprocess_async_client(workdirROOT, appsubprocess_sse_app:app, capture_outputTrue) as client: yield client参数说明workdir应用模块所在的工作目录子进程命令的执行目录app可解析的应用路径字符串如my_app:applicationcapture_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所有通信都在进程内完成这也是测试速度快的原因之一。总结与选型建议场景推荐工具同步测试、无外部事件循环TestClient异步测试、存在异步资源AsyncTestClient快速隔离验证端点/通用逻辑create_test_client/create_async_test_client单测接收Request的纯逻辑守卫、依赖RequestFactoryWebSocket 端点收发验证client.websocket_connect 会话收发 API会话中间件相关测试set_session_data/get_session_datasession_configSSE、真实端口、子进程集成测试subprocess_sync_client/subprocess_async_clientLitestar 的测试工具链以 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
