pytest中mock http.get函数失效问题求助

问题核心原因 #

你的代码使用multiprocessing.Pool创建子进程处理API请求，但pytest的patch mock仅在主进程生效。子进程会重新导入目标模块，导致主进程的mock无法被继承，最终还是调用了真实的API接口。

可行解决方案 #

方案1：替换多进程为线程池（推荐） #

你的任务属于IO密集型（API请求），线程池更适合这类场景，且线程会共享主进程的mock环境。修改sync_mdl_status.py中的多进程代码：

# 导入线程池模块
from concurrent.futures import ThreadPoolExecutor

def update_mdl_status(mdl_con_url: str, schema: str, mdl_threads: int, mdl_auth_tokens: dict):
    # ... 保留原有代码逻辑 ...
    
    # 替换multiprocessing.Pool为ThreadPoolExecutor
    with ThreadPoolExecutor(max_workers=mdl_threads) as executor:
        res = pd.concat(executor.map(
            partial(process_mdl_batch, 
                    mdl_endpoint=mdl_endpoint_for_current_batch,
                    mdl_auth_tokens=mdl_auth_tokens), 
            batch
        ))

修改后，原测试代码中的mock就能正常生效，无需额外调整。

方案2：重构代码，抽离可注入的HTTP客户端 #

将HTTP请求逻辑封装为可替换的依赖，从外部传入客户端实例，避免直接依赖全局的http对象：

# sync_mdl_status.py
def fetch_mdl_status(mdl_auth_tokens: dict, mdl_endpoint: str, row: dict, http_client=None):
    # 使用传入的客户端，未传入则用全局实例
    http_client = http_client or http
    mdl_id = row['mdl_id']
    mdl_auth_token_to_use = mdl_auth_tokens[row['obfuscate_info']['db_name']]
    mdl_headers = {'Content-Type': 'application/json', 'Authorization': 'MDL-AUTH ' + mdl_auth_token_to_use}
    res = http_client.get(f"{mdl_endpoint}/{mdl_id}", headers=mdl_headers)

# 同步修改process_mdl_batch，传递http_client参数
def process_mdl_batch(batch: pandas.DataFrame, mdl_endpoint: str, mdl_auth_tokens: dict, http_client=None):
    batch['status'] = batch['status'].apply(lambda x: x.strip())
    batch['new_status'] = batch['status']
    return batch.apply(lambda row: fetch_mdl_status(mdl_auth_tokens, mdl_endpoint, row, http_client), axis=1)

测试时直接传入mock的HTTP客户端：

def test_status_updater_mdl(self):
    # 创建mock客户端
    from unittest.mock import Mock
    mock_http = Mock()
    mock_http.get.return_value.status_code = 200

    # 调用时传入mock客户端
    batch = update_mdl_status(self.postgresql.url(), "test", 1, aws_auth_token)
    # 验证mock被调用
    mock_http.get.assert_called()

方案3：子进程内重新应用mock（不推荐） #

如果必须保留多进程，需要在子进程执行的函数内重新初始化mock，但这种方式繁琐且容易出错：

def process_mdl_batch(batch: pandas.DataFrame, mdl_endpoint: str, mdl_auth_tokens: dict):
    # 子进程内重新patch
    from unittest.mock import patch
    p = patch('lib.sync_mdl_status.http.get')
    mock_get = p.start()
    mock_get.return_value.status_code = 200
    
    # ... 原有函数逻辑 ...
    
    p.stop()

该方案需要处理进程间的配置传递，且mock逻辑与业务代码耦合度高，仅作为临时替代方案。

测试代码优化示例（配合方案1） #

from lib.sync_mdl_status import update_mdl_status
from unittest.mock import patch


class TestStatusUpdaterMdl(unittest.TestCase):
    def setUp(self):
        self.postgresql = testing.postgresql.Postgresql()
        db_init(self.postgresql)

    def tearDown(self):
        db_ops.close_session()
        self.postgresql.stop()

    def test_status_updater_mdl(self):
        with patch('lib.sync_mdl_status.http.get') as mock_mdl_http_get:
            mock_mdl_http_get.return_value.status_code = 200

            batch = update_mdl_status(self.postgresql.url(), "test", 1, aws_auth_token)
            # 添加断言验证mock被正确调用
            mock_mdl_http_get.assert_called()

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