|
17 | 17 | from qgis.core import QgsTask, QgsApplication
|
18 | 18 |
|
19 | 19 | from qgis.PyQt.QtCore import QCoreApplication
|
| 20 | +from qgis.PyQt.QtTest import QSignalSpy |
20 | 21 |
|
21 | 22 | from qgis.testing import start_app, unittest
|
22 | 23 | from time import sleep
|
@@ -261,6 +262,36 @@ def finished_multiple_value_result(e, results):
|
261 | 262 | self.assertEqual(result_value, 5)
|
262 | 263 | self.assertEqual(result_statement, 'whoo')
|
263 | 264 |
|
| 265 | + def testTaskFromFunctionWithSubTaskCompletedIsCalledOnce(self): |
| 266 | + """ test that when a parent task has subtasks it does emit taskCompleted only once""" |
| 267 | + |
| 268 | + self.finished = 0 |
| 269 | + self.completed = 0 |
| 270 | + |
| 271 | + def _on_finished(): |
| 272 | + self.finished += 1 |
| 273 | + |
| 274 | + def _on_completed(): |
| 275 | + self.completed += 1 |
| 276 | + |
| 277 | + task = QgsTask.fromFunction('test task', run_no_result, on_finished=_on_finished) |
| 278 | + task.taskCompleted.connect(_on_completed) |
| 279 | + spy = QSignalSpy(task.taskCompleted) |
| 280 | + sub_task_1 = QgsTask.fromFunction('test subtask 1', run_no_result, on_finished=_on_finished) |
| 281 | + sub_task_2 = QgsTask.fromFunction('test subtask 2', run_no_result, on_finished=_on_finished) |
| 282 | + task.addSubTask(sub_task_1, [], QgsTask.ParentDependsOnSubTask) |
| 283 | + task.addSubTask(sub_task_2, [], QgsTask.ParentDependsOnSubTask) |
| 284 | + |
| 285 | + QgsApplication.taskManager().addTask(task) |
| 286 | + while task.status() not in [QgsTask.Complete, QgsTask.Terminated]: |
| 287 | + pass |
| 288 | + while QgsApplication.taskManager().countActiveTasks() > 0: |
| 289 | + QCoreApplication.processEvents() |
| 290 | + |
| 291 | + self.assertEqual(self.completed, 1) |
| 292 | + self.assertEqual(self.finished, 3) |
| 293 | + self.assertEqual(len(spy), 1) |
| 294 | + |
264 | 295 |
|
265 | 296 | if __name__ == '__main__':
|
266 | 297 | unittest.main()
|
0 commit comments