You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

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代码的修复方法

你的代码执行完无法退出,是三个明显问题导致的:

  1. 类实例方法function缺少第一个必填的self参数
  2. 线程启动逻辑错误:Thread.start()的返回值是None,你原来的写法将self.thread赋值为None,根本拿不到线程引用做后续回收
  3. 关闭逻辑无效: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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.28 11:21:42