如何使用pytest正确退出队列和Qthread进行测试？

Question

[我用PyQt5创建了一个GUI，然后通过pytest进行了测试。

我的GUI需要重定向标准输出，因此我使用Qthread创建了一个侦听器。该侦听器将stdout放在Queue中，并发送由GUI利用的信号。

直到这里，没有问题。我的问题在我退出时出现。当我退出使用python解释器时，我没有问题，但是当我使用pytest时，我得到了EOFError或一条消息，说我杀死了一个正在运行的线程。我试图正确退出，但问题仍然存在，所以我寻求帮助。

这里是GUI.py的示例：

#!/usr/bin/python3
# -*- coding: utf-8 -*-
import sys

from functools import partial
import multiprocessing
from PyQt5 import QtGui, QtCore, QtWidgets, QtTest

from PyQt5.QtWidgets import *
from PyQt5.QtCore import QCoreApplication, Qt, QObject, pyqtSignal, pyqtSlot, QThread
from PyQt5.QtGui import QIcon, QTextCursor


class MyReceiver(QObject):
    mysignal = pyqtSignal(str)

    def __init__(self,queue,*args,**kwargs):
        QObject.__init__(self,*args,**kwargs)
        self.queue = queue
        self.runCondition=True

    @pyqtSlot(str)
    def run(self):
        while self.runCondition:
            text = self.queue.get()
            self.mysignal.emit(text)

def QueueStreamSetup():

    queue = multiprocessing.Queue(-1)

    sys.stdout = WriteStream(queue)
    #sys.stderr = WriteStream(queue)

    return queue

class WriteStream(object):
    def __init__(self,queue):
        self.queue = queue

    def write(self, text):
        self.queue.put(text)

    def flush(self):
        self.queue.put('FLUSH ')
        QtTest.QTest.qWait(2 * 1000)
        pass

def threadConnect(view, queue):
    qthread = QThread()
    my_receiver = MyReceiver(queue)

    my_receiver.mysignal.connect(view.append_text)
    my_receiver.moveToThread(qthread)
    #
    qthread.started.connect(partial(my_receiver.run,))

    qthread.start()
    return(qthread, my_receiver)


class Example(QMainWindow):
    def __init__(self):
        super().__init__()
        self.initUI(self)

    def restore(self):
        # Restore sys.stdout
        sys.stdout = sys.__stdout__
        sys.stderr = sys.__stderr__

    @pyqtSlot(str)
    def append_text(self,text):
        self.textEdit.moveCursor(QTextCursor.End)
        self.textEdit.insertPlainText( text )
        self.textEdit.moveCursor(QTextCursor.End)

    def initUI(self, MainWindow):
        # centralwidget
        MainWindow.resize(346, 193)
        self.centralwidget = QtWidgets.QWidget(MainWindow)
        # The Action to quit
        self.exitAction = QAction(QIcon('exit24.png'), 'Exit', self)
        self.exitAction.setShortcut('Ctrl+Q')
        self.exitAction.triggered.connect(self.close)

        # The bar
        self.statusBar()
        self.menubar = self.menuBar()
        self.fileMenu = self.menubar.addMenu('&File')
        self.exitMenu=self.fileMenu.addAction(self.exitAction)

        # tThe Button
        self.btn_quit = QtWidgets.QPushButton(self.centralwidget)
        self.btn_quit.setGeometry(QtCore.QRect(120, 20, 89, 25))
        self.btn_quit.clicked.connect(lambda: self.doPrint() )

        # The textEdit
        self.textEdit = QtWidgets.QTextEdit(self.centralwidget)
        self.textEdit.setGeometry(QtCore.QRect(10, 60, 321, 81))

        # Show the frame
        MainWindow.setCentralWidget(self.centralwidget)
        self.show()

    def doPrint(self):
        # Test to print something.
        print('TEST doPrint')

    def closeEvent(self, event):
        # Ask a question before to quit.
        reply = QMessageBox.question(self, 'Message',
            "Are you sure to quit?", QMessageBox.Yes |
            QMessageBox.No, QMessageBox.No)

        # Treat the answer.
        if reply == QMessageBox.Yes:
            self.restore()
            event.accept()
        else:
            event.ignore()

def main():
    queue = QueueStreamSetup()
    app = QApplication(sys.argv)
    ex = Example()
    qthread, my_receiver = threadConnect(ex, queue)

    return app, ex, queue, qthread, my_receiver

def finish(queue, qthread, my_receiver):
    print('Finish')
    my_receiver.runCondition=False
    queue.close()
    queue.join_thread()
    qthread.terminate()
    qthread.wait()
    qthread.exit()
    print('Finish Done')

if __name__ == '__main__':

    app, ex, queue, qthread, my_receiver =main()
    rc= app.exec_()
    finish(queue, qthread, my_receiver)

    print('the application ends with exit code '.format(rc))
    sys.exit(rc)

然后是名为test_GUI.py的pytest文件：

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import os, sys
import pytest

from PyQt5 import QtGui, QtCore, QtWidgets, QtTest
from PyQt5.QtWidgets import *
from PyQt5.QtCore import QCoreApplication, Qt, QObject

import GUI

@pytest.fixture(scope="module")
def Viewer(request):
    print("  SETUP GUI")
    yield GUI.main()
    print("  TEARDOWN GUI")

class Test_GUI_CXS() :

    def test_launching(self, Viewer, qtbot, mocker, caplog):
        # open the window and add the qtbot.
        print("  SETUP Window")
        app, window, queue, qthread, my_receiver = Viewer
        qtbot.addWidget(window)
        qtbot.wait_for_window_shown(window)
        QtTest.QTest.qWait(0.5 *1000)

        # Test
        qtbot.mouseClick( window.btn_quit, QtCore.Qt.LeftButton )
        QtTest.QTest.qWait(0.5 *1000)

        # EXIT
        mocker.patch.object(QMessageBox, 'question', return_value=QMessageBox.Yes)
        window.exitAction.trigger()
        QtTest.QTest.qWait(1 *1000)
        assert window.close()

        # Finish the processes.
        print( "App END")
        GUI.finish(queue, qthread, my_receiver)
        print( "END TEST.")

因此，如果我运行命令：pytest -v -s ./test_GUI.py，则会在消息中得到以下元素：

Qt exceptions in virtual methods:
________________________________________________________________________________
Traceback (most recent call last):
  File "xxx/GUI.py", line 25, in run
   text = self.queue.get()

File "xxx/lib/python3.7/multiprocessing/connection.py", line 383, in _recv
    raise EOFError
EOFError

我不明白为什么会出现文件结尾错误，但是我认为它与Queue和Qthread的终止有关。因为在Queue不为空之前，my_receiver无法停止运行，因此Qthread无法终止。不幸的是，我在Internet上对这种问题一无所获。

对此案的任何建议或帮助，将不胜感激。