[我用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上对这种问题一无所获。
对此案的任何建议或帮助，将不胜感激。
问题是，当您关闭队列时，self.queue.get()仍在运行，从而阻止了线程在阻止执行while self.runCondition:时完成执行。考虑到上述情况，一种可能的解决方案是发送None，然后关闭队列：
class MyReceiver(QObject):
mysignal = pyqtSignal(str)
def __init__(self, queue, *args, **kwargs):
super(MyReceiver, self).__init__(*args, **kwargs)
self.queue = queue
self.runCondition = True
def run(self):
while self.runCondition:
text = self.queue.get()
if isinstance(text, str):
self.mysignal.emit(text)
def finish(queue, qthread, my_receiver):
print('Finish')
my_receiver.runCondition = False
queue.put(None)
qthread.quit()
qthread.wait()
qthread.exit()
queue.close()
queue.join_thread()
print('Finish Done')