如何在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
相关产品推荐
相关产品推荐

