如何正确退出 Queue 和 Qthread 以使用 pytest 进行测试？
How to exit properly a Queue and Qthread for tests with pytest?
我用 PyQt5 创建了一个 GUI，我想通过 pytest 来测试它。
我的 GUI 需要重定向标准输出，所以我使用 Qthread 创建一个监听器。
该侦听器将 stdout 放入 Queue 并发送一个被 GUI 利用的信号。
到这里，没有问题。当我退出什么时，我的问题出现了；当我退出使用 python 解释器时我没有问题，但是当我使用 pytest 时我得到 EOFError 或一条消息说我杀死了一个 运行ning 线程。
我试过正确退出但是问题依旧，所以才来求助
这里是 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)
然后pytest文件名为test_GUI.py:
#!/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 才能停止 运行ning，所以 Qthread 不能终止。
不幸的是，我在互联网上没有找到任何关于此类问题的信息。
如有任何建议或帮助，我们将不胜感激。
问题是，当您关闭队列时，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')
我用 PyQt5 创建了一个 GUI，我想通过 pytest 来测试它。
我的 GUI 需要重定向标准输出，所以我使用 Qthread 创建一个监听器。
该侦听器将 stdout 放入 Queue 并发送一个被 GUI 利用的信号。
到这里，没有问题。当我退出什么时，我的问题出现了；当我退出使用 python 解释器时我没有问题，但是当我使用 pytest 时我得到 EOFError 或一条消息说我杀死了一个 运行ning 线程。
我试过正确退出但是问题依旧，所以才来求助
这里是 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)
然后pytest文件名为test_GUI.py:
#!/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 才能停止 运行ning，所以 Qthread 不能终止。
不幸的是，我在互联网上没有找到任何关于此类问题的信息。
如有任何建议或帮助，我们将不胜感激。
问题是，当您关闭队列时，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')