python - How to cover PyQt5 QThreadPool?

问题描述

How to cover PyQt5 QThreadPool ? Is it possible ?

Following this post https://www.learnpyqt.com/tutorials/multithreading-pyqt-applications-qthreadpool/ one can use PyQt5 with thread pool. Here is an slightly adapted code (to enable coverage) :

~> more .\pyqt_multithread.py
from PyQt5.QtGui import *
from PyQt5.QtWidgets import *
from PyQt5.QtCore import *

import time
import traceback, sys


class WorkerSignals(QObject):
    '''
    Defines the signals available from a running worker thread.

    Supported signals are:

    finished
        No data

    error
        tuple (exctype, value, traceback.format_exc() )

    result
        object data returned from processing, anything

    progress
        int indicating % progress

    '''
    finished = pyqtSignal()
    error = pyqtSignal(tuple)
    result = pyqtSignal(object)
    progress = pyqtSignal(int)


class Worker(QRunnable):
    '''
    Worker thread

    Inherits from QRunnable to handler worker thread setup, signals and wrap-up.

    :param callback: The function callback to run on this worker thread. Supplied args and
                     kwargs will be passed through to the runner.
    :type callback: function
    :param args: Arguments to pass to the callback function
    :param kwargs: Keywords to pass to the callback function

    '''

    def __init__(self, fn, *args, **kwargs):
        super(Worker, self).__init__()

        # Store constructor arguments (re-used for processing)
        self.fn = fn
        self.args = args
        self.kwargs = kwargs
        self.signals = WorkerSignals()

        # Add the callback to our kwargs
        self.kwargs['progress_callback'] = self.signals.progress

    @pyqtSlot()
    def run(self):
        '''
        Initialise the runner function with passed args, kwargs.
        '''

        # Retrieve args/kwargs here; and fire processing using them
        try:
            result = self.fn(*self.args, **self.kwargs)
        except:
            traceback.print_exc()
            exctype, value = sys.exc_info()[:2]
            self.signals.error.emit((exctype, value, traceback.format_exc()))
        else:
            self.signals.result.emit(result)  # Return the result of the processing
        finally:
            self.signals.finished.emit()  # Done



class MainWindow(QMainWindow):


    def __init__(self, *args, **kwargs):
        super(MainWindow, self).__init__(*args, **kwargs)

        self.counter = 0

        layout = QVBoxLayout()

        self.l = QLabel("Start")
        self.b = QPushButton("DANGER!")
        self.b.pressed.connect(self.oh_no)

        layout.addWidget(self.l)
        layout.addWidget(self.b)

        w = QWidget()
        w.setLayout(layout)

        self.setCentralWidget(w)

        self.show()

        self.threadpool = QThreadPool()
        print("Multithreading with maximum %d threads" % self.threadpool.maxThreadCount())

        self.timer = QTimer()
        self.timer.setInterval(1000)
        self.timer.timeout.connect(self.recurring_timer)
        self.timer.start()

    def progress_fn(self, n):
        print("%d%% done" % n)

    def execute_this_fn(self, progress_callback):
        print("THREAD CREATED")
        for n in range(0, 5):
            time.sleep(1)
            progress_callback.emit(int(n*100/4))

        return "Done."

    def print_output(self, s):
        print(s)

    def thread_complete(self):
        print("THREAD COMPLETE!")

    def oh_no(self):
        # Pass the function to execute
        worker = Worker(self.execute_this_fn) # Any other args, kwargs are passed to the run function
        worker.signals.result.connect(self.print_output)
        worker.signals.finished.connect(self.thread_complete)
        worker.signals.progress.connect(self.progress_fn)

        # Execute
        self.threadpool.start(worker)


    def recurring_timer(self):
        self.counter +=1
        self.l.setText("Counter: %d" % self.counter)


if __name__ == '__main__':
    app = QApplication([])
    window = MainWindow()
    app.exec_()

Now to cover this code, I created a unit test :

~> more .\tst.py
import unittest

from PyQt5.QtGui import *
from PyQt5.QtWidgets import *
from PyQt5.QtCore import *
from PyQt5.QtTest import QTest
from pyqt_multithread import MainWindow

import functools

def stopTest(window):
    window.close()

class TestStringMethods(unittest.TestCase):

    def test_upper(self):
        app = QApplication([])
        window = MainWindow()
        timerCallback = functools.partial(stopTest, window)
        timer = QTimer()
        timer.timeout.connect(timerCallback)
        timer.start(10000)
        QTest.mouseClick(window.b, Qt.LeftButton)
        QTest.mouseClick(window.b, Qt.LeftButton)
        QTest.mouseClick(window.b, Qt.LeftButton)
        QTest.mouseClick(window.b, Qt.LeftButton)
        app.exec_()

if __name__ == '__main__':
    unittest.main()

But coverage is not relevant : some part of the code are seen (166-121 is the thread callback) as not covered unless they are


~> python -m coverage run --concurrency=thread tst.py
Multithreading with maximum 4 threads
THREAD CREATED
THREAD CREATED
THREAD CREATED
THREAD CREATED
0% done
0% done
0% done
0% done
25% done
25% done
25% done
25% done
50% done
50% done
50% done
50% done
75% done
75% done
75% done
75% done
100% done
Done.
THREAD COMPLETE!
100% done
Done.
THREAD COMPLETE!
100% done
Done.
THREAD COMPLETE!
100% done
Done.
THREAD COMPLETE!
.
----------------------------------------------------------------------
Ran 1 test in 10.916s

OK


~> python -m coverage report -m
Name                  Stmts   Miss  Cover   Missing
---------------------------------------------------
pyqt_multithread.py      73     16    78%   67-76, 116-121, 146-148
tst.py                   24      0   100%
---------------------------------------------------
TOTAL                    97     16    84%

How to get relevant coverage when covering PyQt5 code ?

标签： pythonmultithreadingpyqt5code-coverage

python - How to cover PyQt5 QThreadPool?

问题描述

解决方案

推荐阅读