Testing Asynchronous Qt Code (part 1)
Posted on Mon 29 June 2015 in programming by Ryan Day • Tagged with python, qt, testing
Asynchronous code is a little different to test than a normal code. If you are testing operations that happen in an asynchronous manner, then the results of those operations may simply not be present when your test case tries to verify them. You must have a better understanding of what is happening behind the scenes in order to test out your asynchronous code.
I've put together a quick repo on Github with examples of a multi threaded GUI written with PySide (a Python Qt library). This GUI will ask for your name. Once you enter it, it will say hello "Name", and add your name to a list. The list is cached using a Mock external service. This service operates outside of Qt on it's own thread.
Qt Connection Types
The first tricky situation is with the Submit Name functionality. You can follow along by executing the program, adding your name to the box, and clicking Submit. Your name pops up in the names list and is added to the NameManager. Here is the code path we want to test:
add_name_signal = Signal(unicode)
def __init__(self):
# ...
self.name_manager = NameManager()
self.name_manager_thread = QThread()
self.name_manager.moveToThread(self.name_manager_thread)
self.name_manager_thread.start()
self.submit_button.clicked.connect(self.say_hello)
self.add_name_signal.connect(self.name_manager.store_name)
self.add_name_signal.connect(self.cache_name)
@Slot()
def say_hello(self):
self.add_name_signal.emit(self.name_text.text())
self.hello_label.setText(u"{} {}".format(GREETING, self.name_text.text()))
self.name_text.setText('')
@Slot()
def cache_name(self, name):
self.names_list.addItem(name)
Take a look at the testNamesList test in the testNames.py file. Run it, it will pass!
def testNamesList(self):
self.main_window.name_text.setText('Ryan')
self.main_window.submit_button.click()
self.main_window.name_text.setText('Meg')
self.main_window.submit_button.click()
assert self.main_window.names_list.count() == 2
assert self.main_window.names_list.item(0).text() == 'Ryan'
assert self.main_window.names_list.item(1).text() == 'Meg'
This test doesn't look tricky, but it is. We aren't testing one condition. The name_manager.store_name connection. Let's add that to the test:
def testNamesList(self):
self.main_window.name_text.setText('Ryan')
self.main_window.submit_button.click()
self.main_window.name_text.setText('Meg')
self.main_window.submit_button.click()
assert self.main_window.names_list.count() == 2
assert self.main_window.names_list.item(0).text() == 'Ryan'
assert self.main_window.names_list.item(1).text() == 'Meg'
assert len(self.main_window.name_manager.names) == 2
We can see that name_manager.store_name is called just the same as the cache_name method is called. So the two should always be equal. Let's run the test now.
$ nosetests tests.testNames:QtTest.testNamesList
.
----------------------------------------------------------------------
Ran 1 test in 0.075s
OK
Great! But wait... what if we try some more times:
$ nosetests tests.testNames:QtTest.testNamesList
.
----------------------------------------------------------------------
Ran 1 test in 0.075s
OK
nosetests tests.testNames:QtTest.testNamesList
F
======================================================================
FAIL: testNamesList (tests.testNames.QtTest)
----------------------------------------------------------------------
Traceback (most recent call last):
File "/home/curious/Development/QtTestMethods/tests/testNames.py", line 38, in testNamesList
assert len(self.main_window.name_manager.names) == 2
AssertionError
----------------------------------------------------------------------
Ran 1 test in 0.075s
FAILED (failures=1)
Ok, so this fails sometimes for the name_manager.names length check, but never for the names_list.count check. What is actually happening here?
The problem lies with connecting the add_name_signal to the handlers.
# This is what the code says
self.add_name_signal.connect(self.name_manager.store_name)
self.add_name_signal.connect(self.cache_name)
# This is what the code is actually doing
self.add_name_signal.connect(self.name_manager.store_name, Qt.AutoConnection)
self.add_name_signal.connect(self.cache_name, Qt.AutoConnection)
# Which really means that the code is doing
self.add_name_signal.connect(self.name_manager.store_name, Qt.QueuedConnection)
self.add_name_signal.connect(self.cache_name, Qt.DirectConnection)
By not specifying the type of connection, we are using a Qt.DirectConnection for the cache_name method, and a Qt.QueuedConnection for the name_manager.store_name method.
To understand this, we turn to the Qt manual for Signal Connection Types. Since our code doesn't specify a connection type, we are using the default Qt.AutoConnection. Since the name_manager runs in its own QThread, that connection becomes a Qt.QueuedConnection. Since cache_name runs in the current thread, it becomes a Qt.DirectConnection.
Qt.DirectConnections* are run synchronously. You can try this out yourself. In Pycharm try stepping through the code. Qt.QueuedConnections are run when the target Qt thread's event loop gets a chance to run them.
The reason our test fails intermittently is because the events for the name_manager thread are sometimes processed before our test checks the result. The rest of the time they are not processed. To avoid having race conditions in our tests, we need to wait for the appropriate result to become available.
def wait_for(cond, to):
"""
Wait for the condition function to return True until the timeout is met.
The timeout is broken into pieces to make sure the application can continue
processing without blocking functionality
:param cond: A function that returns a boolean
:param to: Timeout in seconds
:return: The result of the condition function
"""
watchdog = 0
msecs = (to / 8.) * 1000
while cond() is False and watchdog < 8:
QThread.msleep(msecs)
watchdog += 1
return cond()
def testNamesList(self):
self.done.connect(self.main_window.name_manager_thread.quit)
self.main_window.name_text.setText('Ryan')
self.main_window.submit_button.click()
self.main_window.name_text.setText('Meg')
self.main_window.submit_button.click()
assert self.main_window.names_list.count() == 2
assert self.main_window.names_list.item(0).text() == 'Ryan'
assert self.main_window.names_list.item(1).text() == 'Meg'
assert wait_for(lambda: len(self.main_window.name_manager.names) == 2, 2) is True
This wait_for function will allow us to wait for a condition to become True. The function will timeout after a period of time of course. This lets us update our test. If we run this new test over and over, it will always return True.