Python组件/功能测试Mock HTTP服务器选型及Flask运行关闭问题
适用方案推荐
你需要的是能监听真实本地端口、支持自定义响应逻辑、可以稳定模拟超时等异常、对接pytest的Mock HTTP服务,以下是适配场景的可选方案,按推荐优先级排序:
pytest-httpserver:首选方案,和Flask基于同一个Werkzeug WSGI库实现,专门为测试场景打造。不需要自己写线程管理、启停逻辑,服务默认在后台启动,测试执行完成后自动回收端口、清理进程,完全不会出现服务挂起导致脚本无法退出的问题。原生支持配置任意路由的响应状态码、响应头、响应体,只需给对应路由加固定延迟(设置延迟时长大于你C客户端配置的超时阈值)就能稳定模拟超时场景,同时自带请求校验能力,可以直接断言C端发出的请求内容是否符合预期。官方原生提供pytest fixture,不需要自己写封装代码,直接引入就能用,还支持自动分配空闲端口,避免多用例并行时的端口冲突问题。- 自行封装的Flask服务:如果你已经写了较多Flask版的Mock逻辑不想重构,可以修复现有代码的启停逻辑继续使用,不需要更换框架。
- FastAPI + Uvicorn:如果需要模拟HTTP/2、分块传输等更复杂的HTTP特性,可以选这个组合,只是需要自行实现后台启动、停止的逻辑,样板代码比pytest-httpserver多。
注意:不要选用
responses、requests-mock这类Python进程内的请求拦截库,这类工具只能拦截Python程序内部发出的HTTP请求,无法接收独立C++进程发送的真实网络请求,完全不适用你的场景。
现有Flask代码的修复方法
你的代码执行完无法退出,是三个明显问题导致的:
- 类实例方法
function缺少第一个必填的self参数 - 线程启动逻辑错误:
Thread.start()的返回值是None,你原来的写法将self.thread赋值为None,根本拿不到线程引用做后续回收 - 关闭逻辑无效:
do_teardown_appcontext()只会清理Flask的应用上下文,不会终止底层监听端口的WSGI服务,自然会导致进程挂住。
下面是修复后可正常启停、不会残留进程的Mock类代码:
import time from threading import Thread from flask import Flask, request import requests class MyServerMock: def __init__(self, port=5000): self.port = port self.app = Flask(__name__) self.url = f"http://localhost:{self.port}" self.thread = None # 注册业务路由 self.app.add_url_rule("/path/to/function", view_func=self.function, methods=['POST']) # 注册内部停机路由 self.app.add_url_rule("/_shutdown", view_func=self._shutdown, methods=['GET']) def function(self): # 模拟超时示例:取消下一行注释,将sleep时间设为大于客户端超时阈值即可 # time.sleep(10) return {"code": 0, "msg": "success"} def _shutdown(self): shutdown_func = request.environ.get("werkzeug.server.shutdown") if shutdown_func: shutdown_func() return "server stopped" def run(self): # 设置daemon=True,即使停机逻辑异常,主线程退出时也会强制回收该线程 self.thread = Thread( target=lambda: self.app.run( port=self.port, debug=False, use_reloader=False, threaded=True ), daemon=True ) self.thread.start() # 短暂等待服务完全启动,避免测试代码发起请求时服务还没就绪 time.sleep(0.5) def shutdown(self): # 调用内部停机接口触发服务退出 try: requests.get(f"{self.url}/_shutdown", timeout=1) except Exception: pass # 等待线程回收 if self.thread: self.thread.join(timeout=2)
pytest fixture封装参考
你可以把Mock服务封装为不同作用域的fixture,自动完成启停,不需要在每个用例里重复写启动、关闭逻辑:
import pytest @pytest.fixture(scope="function") def mock_server(): server = MyServerMock(port=6500) server.run() yield server server.shutdown() # 测试用例直接调用即可 def test_example(mock_server): # 这里执行C++组件调用逻辑,请求地址为mock_server.url # 测试断言逻辑 assert 1 == 1
如果不想维护这些启停样板代码,直接替换为pytest-httpserver即可,能省去大量服务管理的工作量。
内容的提问来源于stack exchange,提问作者Kaaml
相关产品推荐
相关产品推荐

