使用Pytest Mock类中watchdog调用遇问题，测试代码失效求解答

我的类代码 #

from watchdog.observers.polling import PollingObserver as Observer
import time  # 原代码遗漏该导入，需补充

class MyObserver:  # 原类依赖但未展示的类，测试时需处理
    pass

class FolderObserver:
    def __init__(self):
        self.__event_handler = MyObserver()
        self.__event_observer = Observer()

    def watch(self, path):
        self.start(path)
        try:
            while True:
                time.sleep(1)
        except:
            self.stop()

    def start(self, path):
        self.__schedule(path)
        self.__event_observer.start()

    def stop(self):
        self.__event_observer.stop()
        self.__event_observer.join()

    def __schedule(self, path):
        self.__event_observer.schedule(self.__event_handler, path, recursive=True)

我尝试的测试代码（无法运行） #

from unittest import TestCase
from unittest.mock import patch
import watchdog.observers.polling

class TestWatch(TestCase):
    """Unit test for loadConfig function"""

    @patch.object(watchdog.observers.polling.PollingObserver,'schedule')    
    @patch.object(watchdog.observers.polling.PollingObserver,'start')
    def test_start_calls_schedule(self, start, schedule):
        src_path = "random_path"
        observer = FolderObserver()
        observer.start(src_path)
        start.assert_called_with(mockObserver, src_path, recursive=True)
        schedule.assert_called_with(mockObserver, src_path, recursive=True)

错误原因分析 #

• 未定义mockObserver变量：断言中使用的mockObserver从未声明，直接触发NameError。
• 断言参数完全错误：
  • Observer.start()方法不需要任何参数，你错误地传入了路径等多余参数。
  • Observer.schedule()的第一个参数是事件处理器（即FolderObserver中的__event_handler），而非不存在的mockObserver，且参数顺序不匹配。
• 缺失MyObserver依赖处理：原类中用到了MyObserver，但测试代码未定义或补丁这个类，运行时会抛出导入/定义错误。
• 潜在阻塞问题：stop()方法中的join()会等待线程结束，若测试中触发stop可能导致测试卡住，需要补丁join方法避免阻塞。

修复后的测试代码 #

from unittest import TestCase
from unittest.mock import patch, MagicMock
from your_module import FolderObserver  # 替换为你的类所在模块名

class TestWatch(TestCase):
    def test_start_calls_schedule(self):
        src_path = "random_path"
        # 补丁MyObserver，避免依赖问题
        with patch('your_module.MyObserver', return_value=MagicMock()) as mock_handler_cls:
            # 补丁PollingObserver的实例方法
            with patch('your_module.Observer') as mock_observer_cls:
                mock_observer = mock_observer_cls.return_value
                
                observer = FolderObserver()
                observer.start(src_path)
                
                # 验证schedule调用参数正确
                mock_observer.schedule.assert_called_once_with(
                    mock_handler_cls.return_value, src_path, recursive=True
                )
                # 验证start被无参调用
                mock_observer.start.assert_called_once_with()

扩展测试（验证watch方法） #

如果需要测试watch方法的异常退出逻辑，可以补充以下代码：

def test_watch_stops_on_exception(self):
    src_path = "random_path"
    with patch('your_module.MyObserver'), patch('your_module.Observer') as mock_observer_cls:
        mock_observer = mock_observer_cls.return_value
        # 补丁sleep并触发异常，模拟watch循环退出
        with patch('time.sleep', side_effect=KeyboardInterrupt):
            observer = FolderObserver()
            observer.watch(src_path)
            
            mock_observer.stop.assert_called_once()
            mock_observer.join.assert_called_once()

内容的提问来源于stack exchange，提问作者Dor Lugasi-Gal