0

我正在尝试在 PyQt5 中实现长时间运行的任务。我的灵感是这个页面。

使用该页面，我已经设法让事情在预测试阶段运行无错误（到目前为止）。现在我想使用测试来构建它。

在之前的一些测试中，我设法QFileDialog通过调用emit并查看连接的插槽是否触发来测试信号（因为它会发生来自类的预先存在的信号）。

但是，当我尝试根据 Maya Posch 的公式为“任务管理器”类设计测试时，我失败了。这是一个 MRE：（取消注释sys.exit以作为应用程序运行的行，或者运行pytest- 必须安装 NB pytest-qt）：

import sys, time
from unittest import mock 

from PyQt5 import QtWidgets, QtCore, QtGui

class LongRunningTask(QtCore.QObject):
    finished_signal = QtCore.pyqtSignal()
    progress_signal = QtCore.pyqtSignal(int)
    
    def run(self):
        print(f'this {self} runs...')
        time.sleep(1)
        self.progress_signal.emit(50)
        time.sleep(1)
        print('...run ends')
        self.progress_signal.emit(100)
        self.finished_signal.emit()
        
    def report_task_progress(self, value):
        print(f'new progress value: {value}')
 
class TaskManager():
    def __init__(self, task_class):
        assert issubclass(task_class, LongRunningTask) 
        self.thread = QtCore.QThread()
        self.task = task_class()
        self.task.moveToThread(self.thread)
        self.thread.started.connect(self.task.run)
        self.task.progress_signal.connect(self.task.report_task_progress)        
        self.task.finished_signal.connect(self.task.deleteLater)
        self.task.finished_signal.connect(self.thread.quit)
        self.thread.finished.connect(self.thread.deleteLater)
        self.thread.start() # can be commented out - see below
    
class MainWindow(QtWidgets.QMainWindow):
    def __init__(self):
        super().__init__()
        self.show()
        # NB horrible errors happen if you make this a local variable
        self.tm = TaskManager(LongRunningTask)

app = QtWidgets.QApplication([])
main_window = MainWindow()
# uncomment to run as app:
# sys.exit(app.exec())

def test_constructor_should_set_started_connect(request, qtbot):
    print(f'\n>>>>>> test: {request.node.nodeid}')
    tm = TaskManager(LongRunningTask)
    with mock.patch.object(tm.task, 'run') as mock_run:
        # NB as stated in the docs, user code can't trigger the "started" signal directly
        # NB this fails even if you don't start the thread in the app code (i.e. comment out self.thread.start())
        tm.thread.start()
        def check_called():
            mock_run.assert_called_once()
        qtbot.waitUntil(check_called, timeout=1000)

def test_constructor_should_set_progress_connect(request, qtbot):
    print(f'\n>>>>>> test: {request.node.nodeid}')
    tm = TaskManager(LongRunningTask)
    with mock.patch.object(tm.task, 'report_task_progress') as mock_report:
        tm.task.progress_signal.emit(33)
        def check_called():
            mock_report.assert_called_once()
        qtbot.waitUntil(check_called, timeout=1000)

之后

...发现qtbot.waitSignal，并取得了一些成功，并带有一些信号...所以我稍微修改了一下：

TaskManager现在给出了两种新方法，目的是查看我是否可以检查是否正确运行QtCore.QTimer.singleShot：

def run_something_async(self):
    def call_async():
        self.delayed_method()
    QtCore.QTimer.singleShot(0, call_async)

def delayed_method(self):
    print('... delayed')

...不幸的是，在合并后waitSignal，以下所有测试都失败了，各不相同，如下所述：

def test_constructor_should_set_started_connect(request, qtbot):
    print(f'\n>>>>>> test: {request.node.nodeid}')
    tm = TaskManager(LongRunningTask)
    with mock.patch.object(tm.task, 'run') as mock_run:
        with qtbot.waitSignal(tm.thread.start, timeout=1000):
        # this fails with 'builtin_function_or_method' object has no attribute 'connect'
            
            # NB as stated in the docs, user code can't trigger the "started" signal directly
            tm.thread.start()
        def check_called():
            mock_run.assert_called_once()
        qtbot.waitUntil(check_called, timeout=1000)

def test_constructor_should_set_progress_connect(request, qtbot):
    print(f'\n>>>>>> test: {request.node.nodeid}')
    tm = TaskManager(LongRunningTask)
    with mock.patch.object(tm.task, 'report_task_progress') as mock_report:
        with qtbot.waitSignal(tm.task.progress_signal, timeout=1000):
            tm.task.progress_signal.emit(33)
        def check_called():
            mock_report.assert_called_once()
        qtbot.waitUntil(check_called, timeout=1000)
         
        # fails with "AssertionError: Expected 'report_task_progress' to have been called once. Called 0 times."
        
 
def test_single_shot_fires(request, qtbot):
    print(f'\n>>>>>> test: {request.node.nodeid}')
    tm = TaskManager(LongRunningTask)
    with mock.patch.object(tm, 'delayed_method') as mock_delayed:
        with qtbot.waitSignal(QtCore.QTimer.timeout, timeout=1000):
        # this fails with "AttributeError: 'PyQt5.QtCore.pyqtSignal' object has no attribute 'connect'"
        
            tm.run_something_async()
         
        def check_called():
            mock_delayed.assert_called_once()
        qtbot.waitUntil(check_called, timeout=1000)
4

0 回答 0