Python中I/O事件模拟问题：文件变更监听测试失败排查

场景与代码 #

业务代码 #

类A基于watchdog实现文件夹监听，文件变更时触发类B的on_any_event方法：

class A:
  def __init__(self, b):
    ...
    self.handler = b

  def run(self):
    # 用watchdog监听文件夹变更
    self.observer.schedule(self.handler, self.input_directory, recursive=True)
    self.observer.start()
    try:
        while not self.stopped:
            time.sleep(self.scanning_frequency)
    except:
        self.observer.stop()

    self.observer.join()

class B(FileSystemEventHandler):
  ...
  def on_any_event(self, event):
    # 处理文件变更事件

测试代码 #

需求：验证文件复制到监听文件夹时，on_any_event被触发，但断言始终为False：

def test_file_watcher(self):
    # arrange
    b = B()
    a = A(b)
    a.handler.on_any_event = MagicMock()

    shutil.copy(# 复制文件到监听文件夹)
    p1 = Process(target=a.run)
    p1.start()

    time.sleep(folder scanning frequency + 1 second)
    a.stop() # 停止监听

    assert a.handler.on_any_event.called
    p1.join()

问题分析与解决 #

1. 断言失败的核心原因 #

就是多进程内存隔离：

• 主进程中给a.handler.on_any_event赋值的MagicMock，只存在主进程内存
• 子进程启动时会复制主进程内存快照，但后续实际调用的是子进程自己内存里的handler方法，主进程的Mock从未被触发，所以断言永远不成立

2. 解决方法 #

方案一：用线程替代进程

watchdog的Observer本身基于线程实现，没必要额外套Process。改用Thread后，主线程和子线程共享内存，Mock的调用状态能被正确捕获：

from threading import Thread

def test_file_watcher(self):
    # 准备Mock对象
    b = B()
    a = A(b)
    mock_on_event = MagicMock()
    a.handler.on_any_event = mock_on_event

    # 先启动监听线程，再复制文件（顺序不能反）
    watcher_thread = Thread(target=a.run)
    watcher_thread.start()
    # 给监听线程留初始化时间
    time.sleep(0.5)
    
    # 执行文件复制操作
    shutil.copy(source_file_path, a.input_directory)

    # 等待事件被捕获处理
    time.sleep(a.scanning_frequency + 1)
    a.stop()
    watcher_thread.join()

    # 断言Mock被调用
    assert mock_on_event.called

方案二：进程间通信传递触发信号

如果必须用进程，可通过multiprocessing.Queue传递事件触发的信号：

from multiprocessing import Process, Queue

def test_file_watcher(self):
    event_queue = Queue()
    
    # 自定义Handler，触发时向队列发送信号
    class SignalB(FileSystemEventHandler):
        def on_any_event(self, event):
            event_queue.put(True)
    
    b = SignalB()
    a = A(b)
    watcher_process = Process(target=a.run)
    watcher_process.start()
    time.sleep(0.5)

    shutil.copy(source_file_path, a.input_directory)

    time.sleep(a.scanning_frequency + 1)
    a.stop()
    watcher_process.join()

    # 检查队列是否收到触发信号
    assert not event_queue.empty()

3. 完全Mock类B的测试方案 #

当然可以，完全Mock B后无需依赖真实文件系统，专注测试A的逻辑：

def test_file_watcher_mock_b(self):
    # 完全Mock B实例
    mock_b = MagicMock(spec=B)
    a = A(mock_b)
    
    # 方案1：模拟事件触发，验证Mock方法被调用
    test_event = FileSystemEvent('test_file_path')
    mock_b.on_any_event(test_event)
    mock_b.on_any_event.assert_called_once_with(test_event)

    # 方案2：验证A是否正确注册了handler到Observer
    a.run()
    a.observer.schedule.assert_called_with(mock_b, a.input_directory, recursive=True)

这种单元测试速度更快、更稳定，无需真实文件操作。

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