pytest中启动本地gRPC服务需用threading.Thread启守护线程,配合socket连接检查确保就绪;服务与channel用session级fixture管理,stub用function级fixture复用;错误断言须用grpc.StatusCode枚举而非字符串匹配。

如何在 pytest 中启动并连接本地 gRPC 服务
直接用 pytest 测试 gRPC 接口,关键不是“怎么写断言”,而是“怎么让服务跑起来且不阻塞测试流程”。gRPC 服务默认是长运行的 server.serve(),会阻塞主线程,导致 pytest 卡住不动。
推荐做法:用 threading.Thread 启动服务,并设为守护线程;同时加短时等待逻辑,确保服务端 socket 真正就绪再发起客户端调用。
- 服务启动必须在
pytest.fixture(scope="session")中完成,避免每个 test 都重复启停 - 用
grpc.aio.Channel或grpc.insecure_channel均可,但同步测试建议用grpc.insecure_channel,更轻量、无 event loop 冲突风险 - 启动后需主动检查端口是否可连(例如用
socket.socket().connect_ex()),不能只靠time.sleep(0.5)
如何构造 gRPC 客户端 stub 并复用到多个测试函数
pytest 的 fixture 是复用 client stub 的自然选择。注意 stub 本身不带连接状态,真正建连发生在第一次 RPC 调用时,所以只要 channel 活着,stub 就可用。
常见错误:在每个 test 函数里新建 grpc.insecure_channel("localhost:50051") —— 这会累积大量 idle channel,触发 warning 甚至内存泄漏。
立即学习“Python免费学习笔记(深入)”;
- 把
channel和stub分开定义:channel 用 session 级 fixture,stub 用 function 级 fixture(因为 stub 是无状态的轻量对象) - 务必在 session fixture 的
yield后调用channel.close(),否则 pytest 退出时可能残留连接 - 若服务端用了 TLS,这里要换成
grpc.secure_channel并传入grpc.ssl_channel_credentials()
如何捕获和断言 gRPC 错误(如 NOT_FOUND、INVALID_ARGUMENT)
gRPC Python 默认把错误包装成 grpc.RpcError 异常,它继承自 Exception,但不等于普通异常——它的状态码、详情、metadata 都需通过属性访问,不能用字符串匹配。
典型误操作:写 assert "NOT_FOUND" in str(exc),这不可靠,且无法区分同名不同码的错误。
- 用
grpc.StatusCode.NOT_FOUND等枚举值做判断,而不是字符串 - 捕获异常后,检查
exc.code()和exc.details(),例如:with pytest.raises(grpc.RpcError) as exc_info: stub.GetUser(GetUserRequest(id="missing")) assert exc_info.value.code() == grpc.StatusCode.NOT_FOUND assert "not found" in exc_info.value.details() - 如果服务端返回了 custom metadata(比如 trace-id),可通过
exc_info.value.trailing_metadata()获取
为什么测试中不要用 asyncio + grpc.aio 在 Python 3.10 下混用 pytest
Python 3.10 默认启用新式 asyncio event loop 策略,而 pytest 默认不管理 event loop 生命周期。若你在 fixture 或 test 中直接 await stub.SayHello(),大概率遇到 RuntimeError: no running event loop 或 asyncio.get_event_loop(): no current event loop。
这不是 pytest 不支持异步,而是当前主流 pytest 插件(如 pytest-asyncio)对 grpc.aio 的兼容仍有边界问题:aio stub 内部依赖 loop,但 loop 可能被 pytest 多次创建/关闭,造成状态错乱。
- 除非明确需要测试流式响应(stream-stream)或超大 payload 的异步行为,否则坚持用同步 stub + 同步 channel
- 若真要用
grpc.aio,必须配合@pytest.mark.asyncio,且所有相关 fixture 也得是 async 的,包括 server 启动逻辑——这会让 setup 成本陡增 - 注意
grpc.aio在 Python 3.10+ 对 SSL 和 keepalive 的默认行为与 sync 版本不一致,容易出现连接提前关闭
context.invocation_metadata(),那权限校验永远走不到你期望的分支。


















