Flask测试中路由嵌套调用报ConnectionError问题解决方案问询

你的问题核心是业务逻辑中直接通过requests等HTTP客户端调用同应用的其他路由，测试阶段应用未启动端口监听，因此HTTP请求无法正常发送。以下是三种可落地的解决方案：

方案一：重构业务逻辑（推荐） #

这是最合理的长期方案，从架构层面降低路由层耦合：

• 将两个路由中复用的业务逻辑抽离为独立的公共函数，路由层仅负责接收请求参数、调用公共函数、封装响应结果
• 测试时直接针对路由入口或公共函数进行测试，无需跨路由发送HTTP请求

示例代码：

# 抽离公共业务逻辑
def common_biz_logic(param1, param2):
    # 原有嵌套路由中的业务实现
    return result

# 路由1
@app.route('/test/url')
def route1():
    param = request.args.get('param')
    res = common_biz_logic(param, 'xxx')
    return jsonify(res)

# 路由2：原有直接调用/test/url的逻辑改为调用公共函数
@app.route('/test/nested_route_call')
def route2():
    res = common_biz_logic('a', 'b')
    return jsonify(res)

方案二：Mock HTTP请求适配测试环境 #

如果暂时无法重构业务代码，可以通过Mock拦截指向本地服务的HTTP请求，直接用test_client处理：

• 在测试fixture中添加requests方法的Mock逻辑，识别host为本地服务的请求，直接转发给当前的test_client执行

示例代码：

import requests
from unittest.mock import patch

@pytest.fixture(scope='module')
def test_client():
    flask_app = create_app("ENV_FILE_LOCATION")    
    with flask_app.test_client() as testing_client:
        with flask_app.app_context():
            # Mock requests.request方法
            def mock_request(method, url, **kwargs):
                from urllib.parse import urlparse
                parsed = urlparse(url)
                # 判断是否是本地服务的请求
                if parsed.netloc in ['0.0.0.0:5500', 'localhost:5500']:
                    # 调用test_client对应方法处理请求
                    client_method = getattr(testing_client, method.lower())
                    path = parsed.path + (f'?{parsed.query}' if parsed.query else '')
                    return client_method(path, **kwargs)
                # 非本地请求走原requests逻辑
                return requests.request(method, url, **kwargs)
            
            with patch('requests.request', side_effect=mock_request):
                yield testing_client

方案三：测试阶段启动临时服务 #

不需要手动干预，通过fixture自动启停临时服务，可直接集成到CI预构建流程：

import threading
import pytest
import time

@pytest.fixture(scope='module')
def test_server():
    app = create_app("ENV_FILE_LOCATION")
    # 启动守护线程运行服务
    server_thread = threading.Thread(target=app.run, kwargs={"host": "0.0.0.0", "port": 5500, "debug": False})
    server_thread.daemon = True
    server_thread.start()
    # 等待服务启动完成
    time.sleep(0.5)
    yield
    # 测试结束后守护线程会随主进程自动销毁

测试用例中直接依赖这个fixture即可，不需要修改原有业务代码。

内容的提问来源于stack exchange，提问作者Amirov