Skip to content
Permalink
Browse files
Add python test for task manager
  • Loading branch information
nyalldawson committed Dec 5, 2016
1 parent d270b4f commit b64025df5c7215b0037de3839df5c3a4276fe61f
Showing with 132 additions and 6 deletions.
  1. +6 −5 python/core/__init__.py
  2. +1 −0 src/core/qgstaskmanager.cpp
  3. +1 −0 tests/src/python/CMakeLists.txt
  4. +2 −1 tests/src/python/test_qgsfloatingwidget.py
  5. +122 −0 tests/src/python/test_qgstaskmanager.py
@@ -187,16 +187,17 @@ def __exit__(self, ex_type, ex_value, traceback):

class QgsTaskWrapper(QgsTask):

def __init__(self, description, function, *extraArgs):
def __init__(self, description, function, *args, **kwargs):
QgsTask.__init__(self, description)
self.extraArgs = extraArgs
self.args = args
self.kwargs = kwargs
self.function = function
self.result = None
self.exception = None

def run(self):
try:
self.function(self, *self.extraArgs)
self.result = self.function(self, *self.args, **self.kwargs)
except Exception as ex:
# report error
self.exception = ex
@@ -206,7 +207,7 @@ def run(self):
self.completed()


def fromFunction(cls, description, function, extraArgs):
return QgsTaskWrapper(description, function, extraArgs)
def fromFunction(cls, description, function, *args, **kwargs):
return QgsTaskWrapper(description, function, *args, **kwargs)

QgsTask.fromFunction = classmethod(fromFunction)
@@ -71,6 +71,7 @@ void QgsTask::unhold()

void QgsTask::setProgress( double progress )
{

mProgress = progress;
emit progressChanged( progress );
}
@@ -108,6 +108,7 @@ ADD_PYTHON_TEST(PyQgsSyntacticSugar test_syntactic_sugar.py)
ADD_PYTHON_TEST(PyQgsStringUtils test_qgsstringutils.py)
ADD_PYTHON_TEST(PyQgsSymbol test_qgssymbol.py)
ADD_PYTHON_TEST(PyQgsSymbolLayerUtils test_qgssymbollayerutils.py)
ADD_PYTHON_TEST(PyQgsTaskManager test_qgstaskmanager.py)
ADD_PYTHON_TEST(PyQgsTextFormatWidget test_qgstextformatwidget.py)
ADD_PYTHON_TEST(PyQgsTreeWidgetItem test_qgstreewidgetitem.py)
ADD_PYTHON_TEST(PyQgsUnitTypes test_qgsunittypes.py)
@@ -13,7 +13,8 @@
__revision__ = '$Format:%H$'

import qgis # NOQA
from PyQt.QtGui import QWidget, QGridLayout
from PyQt.QtGui import QGridLayout
from PyQt.QtWidgets import QWidget

import os

@@ -0,0 +1,122 @@
# -*- coding: utf-8 -*-
"""QGIS Unit tests for QgsTaskManager.
.. note:: This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; either version 2 of the License, or
(at your option) any later version.
"""
__author__ = 'Nyall Dawson'
__date__ = '26/04/2016'
__copyright__ = 'Copyright 2016, The QGIS Project'
# This will get replaced with a git SHA1 when you do a git archive
__revision__ = '$Format:%H$'

import qgis # NOQA
import os

from qgis.core import (
QgsTask,
QgsTaskManager
)

from qgis.testing import start_app, unittest
from time import sleep

start_app()


def run(task, result):
if not result:
raise Exception('cancelled')
else:
return result


def run_with_kwargs(task, password, result):
if not password == 1:
raise Exception('bad password value')
else:
return result


def cancellable(task):
while not task.isCancelled():
pass
if task.isCancelled():
raise Exception('cancelled')


def progress_function(task):
task.setProgress(50)

while not task.isCancelled():
pass
if task.isCancelled():
raise Exception('cancelled')


class TestQgsTaskManager(unittest.TestCase):

def testTaskFromFunction(self):
""" test creating task from function """

task = QgsTask.fromFunction('test task', run, 20)
QgsTaskManager.instance().addTask(task)
while task.status() not in [QgsTask.Complete, QgsTask.Terminated]:
pass

self.assertEqual(task.result, 20)
self.assertEqual(task.status(), QgsTask.Complete)

# try a task which cancels itself
bad_task = QgsTask.fromFunction('test task', run)
QgsTaskManager.instance().addTask(bad_task)
while bad_task.status() not in [QgsTask.Complete, QgsTask.Terminated]:
pass

self.assertFalse(bad_task.result)
self.assertEqual(bad_task.status(), QgsTask.Terminated)

def testTaskFromFunctionWithKwargs(self):
""" test creating task from function using kwargs """

task = QgsTask.fromFunction('test task', run_with_kwargs, result=5, password=1)
QgsTaskManager.instance().addTask(task)
while task.status() not in [QgsTask.Complete, QgsTask.Terminated]:
pass

self.assertEqual(task.result, 5)
self.assertEqual(task.status(), QgsTask.Complete)

def testTaskFromFunctionIsCancellable(self):
""" test that task from function can check cancelled status """
bad_task = QgsTask.fromFunction('test task', cancellable)
QgsTaskManager.instance().addTask(bad_task)
while bad_task.status() != QgsTask.Running:
pass

bad_task.cancel()
while bad_task.status() == QgsTask.Running:
pass

self.assertEqual(bad_task.status(), QgsTask.Terminated)

def testTaskFromFunctionCanSetProgress(self):
""" test that task from function can set progress """
task = QgsTask.fromFunction('test task', progress_function)
QgsTaskManager.instance().addTask(task)
while task.status() != QgsTask.Running:
pass

#wait a fraction so that setProgress gets a chance to be called
sleep(0.001)
self.assertEqual(task.progress(), 50)

task.cancel()
while task.status() == QgsTask.Running:
pass


if __name__ == '__main__':
unittest.main()

0 comments on commit b64025d

Please sign in to comment.