Python多线程函数用pytest/unittest mock时线程仍执行如何解决

• 导入路径匹配错误：你的业务代码通过from threading import Thread将Thread导入到当前模块命名空间，mock时必须patch业务代码所在模块的Thread，而非全局threading.Thread，否则patch不生效，真实线程仍会启动。
• Mock对象参数顺序错误：unittest的mock.patch装饰器参数传递顺序是从下到上，你之前的装饰器顺序和测试函数参数顺序不匹配，导致mock对象注入错误。
• Mock实现逻辑缺失：你自定义的MockThread没有实现参数接收、也没有模拟目标函数返回值写入队列的逻辑，就算mock生效也得不到预期结果。

不需要mock线程，直接mock耗时的目标函数threaded_function即可，既保留多线程执行的真实逻辑，又能跳过100秒的sleep耗时：

# 注意：如果业务代码和测试代码在同一文件，将路径替换为 __main__.threaded_function
@mock.patch("你的业务模块名.threaded_function")
def test_run_threads(mock_func):
    # 自定义mock的返回值规则
    mock_func.side_effect = lambda name: f"res_{name}"
    result = run_threads()
    # 校验返回结果
    assert sorted(result) == ["res_A", "res_B", "res_C"]
    # 校验函数被正确调用3次，参数符合预期
    mock_func.assert_has_calls([mock.call("A"), mock.call("B"), mock.call("C")], any_order=True)

如果你确实需要完全隔离线程执行逻辑，可以用如下正确的mock实现：

# 注意：路径替换为你业务代码所在模块的对应导入路径
@mock.patch("你的业务模块名.Thread")
@mock.patch("你的业务模块名.queue.Queue")
def test_run_threads_mock_thread(mock_queue_cls, mock_thread_cls):
    # 模拟队列返回
    mock_queue = mock.Mock()
    mock_queue_cls.return_value = mock_queue
    # 模拟队列判空和取值逻辑
    mock_queue.empty.side_effect = [False, False, False, True]
    mock_queue.get.side_effect = ["X", "Y", "Z"]

    # 模拟线程实例的start和join方法
    mock_thread = mock.Mock()
    mock_thread_cls.return_value = mock_thread

    result = run_threads()
    # 结果校验
    assert result == ["X", "Y", "Z"]
    # 线程逻辑校验：3个线程都被正确初始化、启动、等待
    assert mock_thread_cls.call_count == 3
    assert mock_thread.start.call_count == 3
    assert mock_thread.join.call_count == 3

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