如何Mock Python Thread子类run方法中调用的函数及替代测试方案

问题根源 #

你当前的patch目标错误：MyThread类里并没有my_function属性，run方法调用的是导入到my_thread模块中的my_function引用，不是my_function模块里的原函数，也不是MyThread类的成员。之前patchMyThread.my_function完全无效，patchmy_function.my_function也不生效，因为my_thread模块已经缓存了原函数的引用。

修正后的测试代码 #

修改patch的目标为my_thread.my_function，同时简化测试逻辑（直接调用run是在当前线程执行，无需多余的sleep和join操作）：

from my_thread import MyThread
import unittest
from unittest.mock import patch


def mock_my_function():
    print("not doing something")
    return False


class TestMyThread(unittest.TestCase):

    def setUp(self):
        self.my_thread = MyThread()

    def test_my_thread_run_method(self):
        # 关键：patch的是my_thread模块中导入的my_function引用
        with patch("my_thread.my_function", side_effect=mock_my_function) as mock_func:
            # 手动设置事件，避免循环直接退出
            self.my_thread.running.set()
            # 执行run方法
            self.my_thread.run()
            # 立即清除事件终止循环
            self.my_thread.running.clear()
            
            # 断言mock函数被调用一次
            mock_func.assert_called_once()
            # 验证返回值符合预期
            self.assertEqual(mock_func.return_value, False)

线程启动场景的测试调整 #

如果是通过self.my_thread.start()启动新线程，测试逻辑需要稍作调整：

def test_my_thread_start(self):
    with patch("my_thread.my_function", side_effect=mock_my_function) as mock_func:
        self.my_thread.start()
        # 等待足够让一次循环执行的时间（无需等满3秒）
        import time
        time.sleep(0.5)
        self.my_thread.running.clear()
        self.my_thread.join()
        
        mock_func.assert_called_once()

通过依赖注入重构MyThread，让它支持传入自定义函数，测试时直接传入测试用函数即可：

重构后的MyThread代码（my_thread.py） #

from my_function import my_function
from threading import Thread, Event
import time


class MyThread(Thread):

    def __init__(self, target_func=None):
        super().__init__(name="MyThread")
        self.running = Event()
        # 允许注入自定义函数，默认使用原my_function
        self.target_func = target_func or my_function

    def run(self):
        self.running.set()

        while self.running.is_set():
            return_value = self.target_func()
            print(return_value)
            time.sleep(3)

对应的测试代码 #

from my_thread import MyThread
import unittest


class TestMyThread(unittest.TestCase):

    def test_my_thread_with_injected_func(self):
        # 用计数器跟踪函数调用次数
        call_count = 0
        def test_func():
            nonlocal call_count
            call_count += 1
            return False
        
        my_thread = MyThread(target_func=test_func)
        my_thread.running.set()
        my_thread.run()
        my_thread.running.clear()
        
        # 断言函数被调用一次
        self.assertEqual(call_count, 1)

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