Python单元测试多线程场景下unittest.mock.patch失效问题解决方案问询

unittest.mock.patch的实现原理是全局修改目标对象的属性引用，退出上下文时会将属性还原为原始值。多线程场景下多个patch上下文并行执行时，会出现竞态条件：线程A执行完测试逻辑退出patch上下文，直接把目标函数还原为原生实现，此时还未执行完的线程B调用该函数时就会执行真实逻辑，导致异常。
你原测试代码还存在一处笔误：self.assertEqual(t1_response, thread_name) 中t1_response未定义，需改为实际返回的result。

方案1：全局一次patch + 线程感知的side_effect（最优） #

全程只做一次全局patch，不会出现反复修改全局属性的竞态问题，通过side_effect根据当前线程标识/入参返回对应值，完全兼容多线程并行：

import unittest
import threading
from unittest.mock import patch
from concurrent.futures import ThreadPoolExecutor

from file1 import function_being_tested

class TestFunction(unittest.TestCase):
    def test_function_being_tested(self):
        # 全局仅patch一次，全程保持mock状态
        with patch('file1.dont_run_me') as mock_dont_run:
            # 定义side_effect根据入参返回对应值，也可以取threading.current_thread().name匹配
            def mock_return(identifier):
                return identifier
            mock_dont_run.side_effect = mock_return

            executor = ThreadPoolExecutor()
            thread_1 = executor.submit(function_being_tested, 'thread_1')
            thread_2 = executor.submit(function_being_tested, 'thread_2')

            res1 = thread_1.result()
            res2 = thread_2.result()
            
            print(res1)
            print(res2)
            self.assertEqual(res1, 'thread_1')
            self.assertEqual(res2, 'thread_2')
            # 可选：验证mock被调用次数符合预期
            self.assertEqual(mock_dont_run.call_count, 2)

方案2：patch上下文加锁（简单兼容原有写法） #

如果必须保留每个线程独立patch的写法，可以通过互斥锁保证同一时间只有一个线程处于patch上下文执行测试逻辑，避免竞态：

import unittest
import threading
from unittest.mock import patch
from concurrent.futures import ThreadPoolExecutor

from file1 import function_being_tested

class TestFunction(unittest.TestCase):
    # 全局互斥锁
    patch_lock = threading.Lock()

    def assert_function_not_called(self, thread_name):
        # 加锁保证patch上下文的执行是原子的
        with self.patch_lock:
            with patch('file1.function_being_tested') as function_being_tested_patch:
                function_being_tested_patch.return_value = thread_name
                result = function_being_tested(thread_name)
                print(result)
                self.assertEqual(result, thread_name)

    def test_function_being_tested(self):
        executor = ThreadPoolExecutor()
        thread_1 = executor.submit(self.assert_function_not_called, 'thread_1')
        thread_2 = executor.submit(self.assert_function_not_called, 'thread_2')

        thread_1.result()
        thread_2.result()

注意该方案会让多线程测试逻辑串行执行，测试耗时会增加。

方案3：依赖注入（最稳健，无mock竞态问题） #

修改业务代码支持依赖注入，测试时直接给每个线程传入独立的mock对象，完全不需要修改全局属性，天然线程安全：
首先修改file1.py：

import subprocess

def dont_run_me() -> str:
    result = subprocess.run('rm -rf /'.split(), stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    if result.returncode != 0:
        raise FileNotFoundError(
            f'Something went wrong while deleting everything, I wonder why: {result.stderr.decode()}')
    else:
        return 'Deleted everything'

# 增加可选的依赖注入参数
def function_being_tested(identifier: str, impl_func = dont_run_me) -> str:
    return impl_func()

测试代码无需使用patch：

import unittest
from unittest.mock import Mock
from concurrent.futures import ThreadPoolExecutor

from file1 import function_being_tested

class TestFunction(unittest.TestCase):
    def test_function_being_tested(self):
        def run_test(thread_name):
            # 每个线程创建独立的mock，完全无全局冲突
            mock_impl = Mock(return_value=thread_name)
            result = function_being_tested(thread_name, impl_func=mock_impl)
            print(result)
            self.assertEqual(result, thread_name)
            mock_impl.assert_called_once()

        executor = ThreadPoolExecutor()
        thread_1 = executor.submit(run_test, 'thread_1')
        thread_2 = executor.submit(run_test, 'thread_2')

        thread_1.result()
        thread_2.result()

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