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

如何在multiprocessing.Process启动的进程中mock函数?

问题根源

你遇到的核心问题是:用multiprocessing启动的服务器是独立子进程,主进程里的@patch无法作用于子进程。子进程启动时会重新加载所有Python模块,主进程中做的Mock完全不会被继承,所以真实的validate_jwt函数依然会被执行。

解决方案

方案1:在子进程内部完成Mock(推荐,无需修改业务代码)

直接在启动服务器的子进程逻辑里,先完成Mock再导入并启动服务器。这样Mock会在子进程的模块加载阶段生效。

修改测试代码中的run_server函数:

def run_server(port):
    from unittest.mock import patch
    # 注意:这里要patch的是main模块中已经导入的validate_jwt引用,也就是foo.bar.main.validate_jwt
    with patch('foo.bar.main.validate_jwt') as mock_val:
        mock_val.return_value = {"user": "test_user"}
        # 必须在patch之后再导入start,确保模块加载时Mock已生效
        from foo.bar.main import start
        start(is_prod=False, port=port)

方案2:通过环境变量切换测试模式(需修改业务代码)

在业务代码中增加测试模式判断,当检测到测试环境时,直接使用Mock逻辑跳过真实认证。

步骤1:修改foo/bar/main.py

from .baz import validate_jwt
import os

def mock_validate_jwt(token: str) -> dict:
    return {"user": "test_user"}

def server(token):
    # 检测是否处于测试模式
    if os.getenv("TEST_MODE") == "1":
        valid = mock_validate_jwt(token)
    else:
        valid = validate_jwt(token)

步骤2:修改测试代码的start_server Fixture

@pytest.fixture(scope="module")
def start_server():
    port = 8002
    # 设置测试环境变量
    os.environ["TEST_MODE"] = "1"
    server_process = multiprocessing.Process(target=run_server, args=(port,))
    server_process.start()

    max_retries = 10
    for i in range(max_retries):
        try:
            response = requests.get(f"http://127.0.0.1:{port}/echo?message=test")
            if response.status_code == 200:
                break
        except requests.ConnectionError:
            time.sleep(1)
    else:
        pytest.fail("Server did not start in time")
    
    # 测试结束后清理环境变量并终止进程
    yield
    del os.environ["TEST_MODE"]
    server_process.terminate()
为什么之前的Patch无效?

你之前在主进程的test_function中用@patch,但start_server Fixture启动的是独立子进程——两个进程拥有完全独立的内存空间,主进程的Mock不会被子进程继承。子进程会重新加载foo.bar.main模块,此时validate_jwt还是指向真实函数引用。

内容的提问来源于stack exchange,提问作者Norman Robins

相关产品推荐
方舟 Agent Plan

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

最近更新时间:2026.06.19 06:46:16