如何用pytest-qt的waitSignal测试Qt线程池长时计算并获取Worker对象？

问题1：无法从线程池获取Worker对象 #

QThreadPool本身没有提供直接获取Worker实例的接口，你需要在应用代码里主动维护Worker的引用：

• 在主窗口类中添加一个容器（比如列表），专门存储当前正在运行的Worker实例
• 创建Worker时将其加入容器，Worker完成任务发射信号时，再从容器中移除它

主窗口代码示例：

from PyQt5.QtCore import QThreadPool, QRunnable, pyqtSignal, QObject
from PyQt5.QtWidgets import QMainWindow, QPushButton

class WorkerSignals(QObject):
    finished = pyqtSignal()

class Worker(QRunnable):
    def __init__(self):
        super().__init__()
        self.signals = WorkerSignals()

    def run(self):
        # 模拟长耗时计算任务
        import time
        time.sleep(1)
        self.signals.finished.emit()

class MainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.threadpool = QThreadPool.globalInstance()
        self.active_workers = []  # 维护活跃Worker的引用
        self.btn = QPushButton("Start Task", self)
        self.btn.clicked.connect(self.start_task)
        self.setCentralWidget(self.btn)

    def start_task(self):
        worker = Worker()
        # Worker完成时从列表移除
        worker.signals.finished.connect(lambda: self.active_workers.remove(worker))
        self.active_workers.append(worker)
        self.threadpool.start(worker)

这样在测试用例中，直接通过window.active_workers就能获取到Worker实例。

问题2：mouseClick触发按钮后信号未被正确监听 #

常见原因及对应解决方法：

1. 信号绑定时机错误：如果任务执行过快，可能在调用waitSignal之前就已完成，这种情况要先绑定信号监听，再点击按钮
2. 监听的信号不匹配：确保你监听的是Worker实际发射的信号，或者主窗口转发的信号
3. 超时时间不足：如果任务耗时超过默认的5000ms，需要手动设置更长的超时时间

测试用例示例（先绑定监听再触发按钮）：

from PyQt5.QtCore import Qt

def test_worker_finished_signal(qtbot):
    window = MainWindow()
    qtbot.addWidget(window)
    
    # 先设置信号监听，再点击按钮触发任务
    with qtbot.waitSignal(window.task_finished, timeout=2000) as blocker:
        qtbot.mouseClick(window.btn, Qt.LeftButton)
    
    # 验证信号确实被触发
    assert blocker.signal_triggered

更推荐的做法是让主窗口转发Worker信号（解耦测试与Worker内部实现）：

# 主窗口添加窗口级别的完成信号
class MainWindow(QMainWindow):
    task_finished = pyqtSignal()
    
    def start_task(self):
        worker = Worker()
        worker.signals.finished.connect(self.task_finished)  # 转发信号
        worker.signals.finished.connect(lambda: self.active_workers.remove(worker))
        self.active_workers.append(worker)
        self.threadpool.start(worker)

另外要检查Worker的run方法中是否正确调用了self.signals.finished.emit()，避免漏写信号发射逻辑。

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