summaryrefslogtreecommitdiff
path: root/src/leap
diff options
context:
space:
mode:
authorkali <kali@leap.se>2012-11-30 04:46:55 +0900
committerkali <kali@leap.se>2012-12-12 04:27:50 +0900
commitff6d4b8633edc763f22489030766a6c7a9377693 (patch)
tree6f35a3d1114874abc331a79c73bc927097d30156 /src/leap
parentb262ac8bae66c391aa249e93268db9935f1c475f (diff)
progress initial tests
Diffstat (limited to 'src/leap')
-rw-r--r--src/leap/base/pluggableconfig.py3
-rw-r--r--src/leap/gui/progress.py95
-rw-r--r--src/leap/gui/tests/__init__.py0
-rw-r--r--src/leap/gui/tests/test_mainwindow_rc.py (renamed from src/leap/gui/test_mainwindow_rc.py)3
-rw-r--r--src/leap/gui/tests/test_progress.py284
-rw-r--r--src/leap/gui/tests/test_threads.py27
-rw-r--r--src/leap/testing/pyqt.py52
-rw-r--r--src/leap/testing/qunittest.py302
8 files changed, 736 insertions, 30 deletions
diff --git a/src/leap/base/pluggableconfig.py b/src/leap/base/pluggableconfig.py
index 34c1e060..0ca985ea 100644
--- a/src/leap/base/pluggableconfig.py
+++ b/src/leap/base/pluggableconfig.py
@@ -419,7 +419,8 @@ class PluggableConfig(object):
return True
-def testmain():
+def testmain(): # pragma: no cover
+
from tests import test_validation as t
import pprint
diff --git a/src/leap/gui/progress.py b/src/leap/gui/progress.py
index 64b87b2c..e68c35d2 100644
--- a/src/leap/gui/progress.py
+++ b/src/leap/gui/progress.py
@@ -4,7 +4,7 @@ from first run wizard
"""
try:
from collections import OrderedDict
-except ImportError:
+except ImportError: # pragma: no cover
# We must be in 2.6
from leap.util.dicts import OrderedDict
@@ -73,15 +73,16 @@ class ProgressStepContainer(object):
self.steps = {}
def step(self, identity):
- return self.step.get(identity)
+ return self.steps.get(identity, None)
def addStep(self, step):
self.steps[step.index] = step
def removeStep(self, step):
- del self.steps[step.index]
- del step
- self.dirty = True
+ if step and self.steps.get(step.index, None):
+ del self.steps[step.index]
+ del step
+ self.dirty = True
def removeAllSteps(self):
for item in iter(self):
@@ -107,7 +108,7 @@ class StepsTableWidget(QtGui.QTableWidget):
"""
def __init__(self, parent=None):
- super(StepsTableWidget, self).__init__(parent)
+ super(StepsTableWidget, self).__init__(parent=parent)
# remove headers and all edit/select behavior
self.horizontalHeader().hide()
@@ -149,18 +150,39 @@ class StepsTableWidget(QtGui.QTableWidget):
class WithStepsMixIn(object):
+ """
+ This Class is a mixin that can be inherited
+ by InlineValidation pages (which will display
+ a progress steps widget in the same page as the form)
+ or by Validation Pages (which will only display
+ the progress steps in the page, below a progress bar widget)
+ """
+ STEPS_TIMER_MS = 100
- # worker threads for checks
+ #
+ # methods related to worker threads
+ # launched for individual checks
+ #
def setupStepsProcessingQueue(self):
+ """
+ should be called from the init method
+ of the derived classes
+ """
self.steps_queue = Queue.Queue()
self.stepscheck_timer = QtCore.QTimer()
self.stepscheck_timer.timeout.connect(self.processStepsQueue)
- self.stepscheck_timer.start(100)
+ self.stepscheck_timer.start(self.STEPS_TIMER_MS)
# we need to keep a reference to child threads
self.threads = []
def do_checks(self):
+ """
+ main entry point for checks.
+ it calls _do_checks in derived classes,
+ and it expects it to be a generator
+ yielding a tuple in the form (("message", progress_int), checkfunction)
+ """
# yo dawg, I heard you like checks
# so I put a __do_checks in your do_checks
@@ -168,7 +190,7 @@ class WithStepsMixIn(object):
def __do_checks(fun=None, queue=None):
- for checkcase in fun():
+ for checkcase in fun(): # pragma: no cover
checkmsg, checkfun = checkcase
queue.put(checkmsg)
@@ -180,15 +202,34 @@ class WithStepsMixIn(object):
__do_checks,
fun=self._do_checks,
queue=self.steps_queue))
- t.finished.connect(self.on_checks_validation_ready)
+ if hasattr(self, 'on_checks_validation_ready'):
+ t.finished.connect(self.on_checks_validation_ready)
t.begin()
self.threads.append(t)
+ def processStepsQueue(self):
+ """
+ consume steps queue
+ and pass messages
+ to the ui updater functions
+ """
+ while self.steps_queue.qsize():
+ try:
+ status = self.steps_queue.get(0)
+ if status == "failed":
+ self.set_failed_icon()
+ else:
+ self.onStepStatusChanged(*status)
+ except Queue.Empty: # pragma: no cover
+ pass
+
def fail(self, err=None):
"""
return failed state
and send error notification as
- a nice side effect
+ a nice side effect. this function is called from
+ the _do_checks check functions returned in the
+ generator.
"""
wizard = self.wizard()
senderr = lambda err: wizard.set_validation_error(
@@ -202,38 +243,29 @@ class WithStepsMixIn(object):
def launch_checks(self):
self.do_checks()
+ # (gui) presentation stuff begins #####################
+
# slot
#@QtCore.pyqtSlot(str, int)
def onStepStatusChanged(self, status, progress=None):
if status not in ("head_sentinel", "end_sentinel"):
self.add_status_line(status)
if status in ("end_sentinel"):
- self.checks_finished = True
+ #self.checks_finished = True
self.set_checked_icon()
if progress and hasattr(self, 'progress'):
self.progress.setValue(progress)
self.progress.update()
- def processStepsQueue(self):
- """
- consume steps queue
- and pass messages
- to the ui updater functions
- """
- while self.steps_queue.qsize():
- try:
- status = self.steps_queue.get(0)
- if status == "failed":
- self.set_failed_icon()
- else:
- self.onStepStatusChanged(*status)
- except Queue.Empty:
- pass
-
def setupSteps(self):
self.steps = ProgressStepContainer()
# steps table widget
- self.stepsTableWidget = StepsTableWidget(self)
+ if isinstance(self, QtCore.QObject):
+ parent = self
+ else:
+ parent = None
+ import ipdb;ipdb.set_trace()
+ self.stepsTableWidget = StepsTableWidget(parent=parent)
zeros = (0, 0, 0, 0)
self.stepsTableWidget.setContentsMargins(*zeros)
self.errors = OrderedDict()
@@ -295,6 +327,8 @@ class WithStepsMixIn(object):
# setting cell widget.
# see note on StepsTableWidget about plans to
# change this for a better solution.
+ if not hasattr(self, 'steps'):
+ return
index = len(self.steps)
table = self.stepsTableWidget
_index = index - 1 if current else index - 2
@@ -340,6 +374,9 @@ class WithStepsMixIn(object):
def is_done(self):
return self.done
+ # convenience for going back and forth
+ # in the wizard pages.
+
def go_back(self):
self.wizard().back()
diff --git a/src/leap/gui/tests/__init__.py b/src/leap/gui/tests/__init__.py
new file mode 100644
index 00000000..e69de29b
--- /dev/null
+++ b/src/leap/gui/tests/__init__.py
diff --git a/src/leap/gui/test_mainwindow_rc.py b/src/leap/gui/tests/test_mainwindow_rc.py
index c5abb4aa..67b9fae0 100644
--- a/src/leap/gui/test_mainwindow_rc.py
+++ b/src/leap/gui/tests/test_mainwindow_rc.py
@@ -27,3 +27,6 @@ class MainWindowResourcesTest(unittest.TestCase):
self.assertEqual(
hashlib.md5(mainwindow_rc.qt_resource_data).hexdigest(),
'53e196f29061d8f08f112e5a2e64eb53')
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/src/leap/gui/tests/test_progress.py b/src/leap/gui/tests/test_progress.py
new file mode 100644
index 00000000..ff6a0bf1
--- /dev/null
+++ b/src/leap/gui/tests/test_progress.py
@@ -0,0 +1,284 @@
+import sys
+import unittest
+import Queue
+
+import mock
+
+from leap.testing import qunittest
+from leap.testing import pyqt
+
+from PyQt4 import QtGui
+from PyQt4 import QtCore
+from PyQt4.QtTest import QTest
+from PyQt4.QtCore import Qt
+
+from leap.gui import progress
+
+
+class ProgressStepTestCase(unittest.TestCase):
+
+ def test_step_attrs(self):
+ ps = progress.ProgressStep
+ step = ps('test', False, 1)
+ # instance
+ self.assertEqual(step.index, 1)
+ self.assertEqual(step.name, "test")
+ self.assertEqual(step.done, False)
+ step = ps('test2', True, 2)
+ self.assertEqual(step.index, 2)
+ self.assertEqual(step.name, "test2")
+ self.assertEqual(step.done, True)
+
+ # class methods and attrs
+ self.assertEqual(ps.columns(), ('name', 'done'))
+ self.assertEqual(ps.NAME, 0)
+ self.assertEqual(ps.DONE, 1)
+
+
+class ProgressStepContainerTestCase(unittest.TestCase):
+ def setUp(self):
+ self.psc = progress.ProgressStepContainer()
+
+ def addSteps(self, number):
+ Step = progress.ProgressStep
+ for n in range(number):
+ self.psc.addStep(Step("%s" % n, False, n))
+
+ def test_attrs(self):
+ self.assertEqual(self.psc.columns,
+ ('name', 'done'))
+
+ def test_add_steps(self):
+ Step = progress.ProgressStep
+ self.assertTrue(len(self.psc) == 0)
+ self.psc.addStep(Step('one', False, 0))
+ self.assertTrue(len(self.psc) == 1)
+ self.psc.addStep(Step('two', False, 1))
+ self.assertTrue(len(self.psc) == 2)
+
+ def test_del_all_steps(self):
+ self.assertTrue(len(self.psc) == 0)
+ self.addSteps(5)
+ self.assertTrue(len(self.psc) == 5)
+ self.psc.removeAllSteps()
+ self.assertTrue(len(self.psc) == 0)
+
+ def test_del_step(self):
+ Step = progress.ProgressStep
+ self.addSteps(5)
+ self.assertTrue(len(self.psc) == 5)
+ self.psc.removeStep(self.psc.step(4))
+ self.assertTrue(len(self.psc) == 4)
+ self.psc.removeStep(self.psc.step(4))
+ self.psc.removeStep(Step('none', False, 5))
+ self.psc.removeStep(self.psc.step(4))
+
+ def test_iter(self):
+ self.addSteps(10)
+ self.assertEqual(
+ [x.index for x in self.psc],
+ [x for x in range(10)])
+
+
+class StepsTableWidgetTestCase(unittest.TestCase):
+
+ def setUp(self):
+ self.app = QtGui.QApplication(sys.argv)
+ QtGui.qApp = self.app
+ self.stw = progress.StepsTableWidget()
+
+ def tearDown(self):
+ QtGui.qApp = None
+ self.app = None
+
+ def test_defaults(self):
+ self.assertTrue(isinstance(self.stw, QtGui.QTableWidget))
+ self.assertEqual(self.stw.focusPolicy(), 0)
+
+
+class TestWithStepsClass(QtGui.QWidget, progress.WithStepsMixIn):
+
+ def __init__(self):
+ self.setupStepsProcessingQueue()
+ self.statuses = []
+ self.current_page = "testpage"
+
+ def onStepStatusChanged(self, *args):
+ """
+ blank out this gui method
+ that will add status lines
+ """
+ self.statuses.append(args)
+
+
+class WithStepsMixInTestCase(qunittest.TestCase):
+
+ TIMER_WAIT = 2 * progress.WithStepsMixIn.STEPS_TIMER_MS / 1000.0
+
+ # XXX can spy on signal connections
+
+ def setUp(self):
+ self.app = QtGui.QApplication(sys.argv)
+ QtGui.qApp = self.app
+ self.stepy = TestWithStepsClass()
+ #self.connects = []
+ #pyqt.enableSignalDebugging(
+ #connectCall=lambda *args: self.connects.append(args))
+ #self.assertEqual(self.connects, [])
+ #self.stepy.stepscheck_timer.timeout.disconnect(
+ #self.stepy.processStepsQueue)
+
+ def tearDown(self):
+ QtGui.qApp = None
+ self.app = None
+
+ def test_has_queue(self):
+ s = self.stepy
+ self.assertTrue(hasattr(s, 'steps_queue'))
+ self.assertTrue(isinstance(s.steps_queue, Queue.Queue))
+ self.assertTrue(isinstance(s.stepscheck_timer, QtCore.QTimer))
+
+ def test_do_checks_delegation(self):
+ s = self.stepy
+
+ _do_checks = mock.Mock()
+ _do_checks.return_value = (
+ (("test", 0), lambda: None),
+ (("test", 0), lambda: None))
+ s._do_checks = _do_checks
+ s.do_checks()
+ self.waitFor(seconds=self.TIMER_WAIT)
+ _do_checks.assert_called_with()
+ self.assertEqual(len(s.statuses), 2)
+
+ # test that a failed test interrupts the run
+
+ s.statuses = []
+ _do_checks = mock.Mock()
+ _do_checks.return_value = (
+ (("test", 0), lambda: None),
+ (("test", 0), lambda: False),
+ (("test", 0), lambda: None))
+ s._do_checks = _do_checks
+ s.do_checks()
+ self.waitFor(seconds=self.TIMER_WAIT)
+ _do_checks.assert_called_with()
+ self.assertEqual(len(s.statuses), 2)
+
+ def test_process_queue(self):
+ s = self.stepy
+ q = s.steps_queue
+ s.set_failed_icon = mock.MagicMock()
+ with self.assertRaises(AssertionError):
+ q.put('foo')
+ self.waitFor(seconds=self.TIMER_WAIT)
+ s.set_failed_icon.assert_called_with()
+ q.put("failed")
+ self.waitFor(seconds=self.TIMER_WAIT)
+ s.set_failed_icon.assert_called_with()
+
+ def test_on_checks_validation_ready_called(self):
+ s = self.stepy
+ s.on_checks_validation_ready = mock.MagicMock()
+
+ _do_checks = mock.Mock()
+ _do_checks.return_value = (
+ (("test", 0), lambda: None),)
+ s._do_checks = _do_checks
+ s.do_checks()
+
+ self.waitFor(seconds=self.TIMER_WAIT)
+ s.on_checks_validation_ready.assert_called_with()
+
+ def test_fail(self):
+ s = self.stepy
+
+ s.wizard = mock.Mock()
+ wizard = s.wizard.return_value
+ wizard.set_validation_error.return_value = True
+ s.completeChanged = mock.Mock()
+ s.completeChanged.emit.return_value = True
+
+ self.assertFalse(s.fail(err="foo"))
+ self.waitFor(seconds=self.TIMER_WAIT)
+ wizard.set_validation_error.assert_called_with('testpage', 'foo')
+ s.completeChanged.emit.assert_called_with()
+
+ # with no args
+ s.wizard = mock.Mock()
+ wizard = s.wizard.return_value
+ wizard.set_validation_error.return_value = True
+ s.completeChanged = mock.Mock()
+ s.completeChanged.emit.return_value = True
+
+ self.assertFalse(s.fail())
+ self.waitFor(seconds=self.TIMER_WAIT)
+ with self.assertRaises(AssertionError):
+ wizard.set_validation_error.assert_called_with()
+ s.completeChanged.emit.assert_called_with()
+
+ def test_done(self):
+ s = self.stepy
+ s.done = False
+
+ s.completeChanged = mock.Mock()
+ s.completeChanged.emit.return_value = True
+
+ self.assertFalse(s.is_done())
+ s.set_done()
+ self.assertTrue(s.is_done())
+ s.completeChanged.emit.assert_called_with()
+
+ s.completeChanged = mock.Mock()
+ s.completeChanged.emit.return_value = True
+ s.set_undone()
+ self.assertFalse(s.is_done())
+
+ def test_back_and_next(self):
+ s = self.stepy
+ s.wizard = mock.Mock()
+ wizard = s.wizard.return_value
+ wizard.back.return_value = True
+ wizard.next.return_value = True
+ s.go_back()
+ wizard.back.assert_called_with()
+ s.go_next()
+ wizard.next.assert_called_with()
+
+ def test_on_step_statuschanged_slot(self):
+ s = self.stepy
+ s.onStepStatusChanged = progress.WithStepsMixIn.onStepStatusChanged
+ s.add_status_line = mock.Mock()
+ s.set_checked_icon = mock.Mock()
+ s.progress = mock.Mock()
+ s.progress.setValue.return_value = True
+ s.progress.update.return_value = True
+
+ s.onStepStatusChanged(s, "end_sentinel")
+ s.set_checked_icon.assert_called_with()
+
+ s.onStepStatusChanged(s, "foo")
+ s.add_status_line.assert_called_with("foo")
+
+ s.onStepStatusChanged(s, "bar", 42)
+ s.progress.setValue.assert_called_with(42)
+ s.progress.update.assert_called_with()
+
+ def test_steps_and_errors(self):
+ s = self.stepy
+ s.setupSteps()
+ self.assertTrue(isinstance(s.steps, progress.ProgressStepContainer))
+ self.assertEqual(s.errors, {})
+
+
+
+class InlineValidationPageTestCase(unittest.TestCase):
+ pass
+
+
+class ValidationPage(unittest.TestCase):
+ pass
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/src/leap/gui/tests/test_threads.py b/src/leap/gui/tests/test_threads.py
new file mode 100644
index 00000000..06c19606
--- /dev/null
+++ b/src/leap/gui/tests/test_threads.py
@@ -0,0 +1,27 @@
+import unittest
+
+import mock
+from leap.gui import threads
+
+
+class FunThreadTestCase(unittest.TestCase):
+
+ def setUp(self):
+ self.fun = mock.MagicMock()
+ self.fun.return_value = "foo"
+ self.t = threads.FunThread(fun=self.fun)
+
+ def test_thread(self):
+ self.t.begin()
+ self.t.wait()
+ self.fun.assert_called()
+ del self.t
+
+ def test_run(self):
+ # this is called by PyQt
+ self.t.run()
+ del self.t
+ self.fun.assert_called()
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/src/leap/testing/pyqt.py b/src/leap/testing/pyqt.py
new file mode 100644
index 00000000..6edaf059
--- /dev/null
+++ b/src/leap/testing/pyqt.py
@@ -0,0 +1,52 @@
+from PyQt4 import QtCore
+
+_oldConnect = QtCore.QObject.connect
+_oldDisconnect = QtCore.QObject.disconnect
+_oldEmit = QtCore.QObject.emit
+
+
+def _wrapConnect(callableObject):
+ """
+ Returns a wrapped call to the old version of QtCore.QObject.connect
+ """
+ @staticmethod
+ def call(*args):
+ callableObject(*args)
+ _oldConnect(*args)
+ return call
+
+
+def _wrapDisconnect(callableObject):
+ """
+ Returns a wrapped call to the old version of QtCore.QObject.disconnect
+ """
+ @staticmethod
+ def call(*args):
+ callableObject(*args)
+ _oldDisconnect(*args)
+ return call
+
+
+def enableSignalDebugging(**kwargs):
+ """
+ Call this to enable Qt Signal debugging. This will trap all
+ connect, and disconnect calls.
+ """
+
+ f = lambda *args: None
+ connectCall = kwargs.get('connectCall', f)
+ disconnectCall = kwargs.get('disconnectCall', f)
+ emitCall = kwargs.get('emitCall', f)
+
+ def printIt(msg):
+ def call(*args):
+ print msg, args
+ return call
+ QtCore.QObject.connect = _wrapConnect(connectCall)
+ QtCore.QObject.disconnect = _wrapDisconnect(disconnectCall)
+
+ def new_emit(self, *args):
+ emitCall(self, *args)
+ _oldEmit(self, *args)
+
+ QtCore.QObject.emit = new_emit
diff --git a/src/leap/testing/qunittest.py b/src/leap/testing/qunittest.py
new file mode 100644
index 00000000..b89ccec3
--- /dev/null
+++ b/src/leap/testing/qunittest.py
@@ -0,0 +1,302 @@
+# -*- coding: utf-8 -*-
+
+# **qunittest** is an standard Python `unittest` enhancement for PyQt4,
+# allowing
+# you to test asynchronous code using standard synchronous testing facility.
+#
+# The source for `qunittest` is available on [GitHub][gh], and released under
+# the MIT license.
+#
+# Slightly modified by The Leap Project.
+
+### Prerequisites
+
+# Import unittest2 or unittest
+try:
+ import unittest2 as unittest
+except ImportError:
+ import unittest
+
+# ... and some standard Python libraries
+import sys
+import functools
+import contextlib
+import re
+
+# ... and several PyQt classes
+from PyQt4.QtCore import QTimer
+from PyQt4.QtTest import QTest
+from PyQt4 import QtGui
+
+### The code
+
+
+# Override standard main method, by invoking it inside PyQt event loop
+
+def main(*args, **kwargs):
+ qapplication = QtGui.QApplication(sys.argv)
+
+ QTimer.singleShot(0, unittest.main(*args, **kwargs))
+ qapplication.exec_()
+
+"""
+This main substitute does not integrate with unittest.
+
+Note about mixing the event loop and unittests:
+
+Unittest will fail if we keep more than one reference to a QApplication.
+(pyqt expects to be and only one).
+So, for the things that need a QApplication to exist, do something like:
+
+ self.app = QApplication()
+ QtGui.qApp = self.app
+
+in the class setUp, and::
+
+ QtGui.qApp = None
+ self.app = None
+
+in the class tearDown.
+
+For some explanation about this, see
+ http://stuvel.eu/blog/127/multiple-instances-of-qapplication-in-one-process
+and
+ http://www.riverbankcomputing.com/pipermail/pyqt/2010-September/027705.html
+"""
+
+
+# Helper returning the name of a given signal
+
+def _signal_name(signal):
+ s = repr(signal)
+ name_re = "signal (\w+) of (\w+)"
+ match = re.search(name_re, s, re.I)
+ if not match:
+ return "??"
+ return "%s#%s" % (match.group(2), match.group(1))
+
+
+class _SignalConnector(object):
+ """ Encapsulates signal assertion testing """
+ def __init__(self, test, signal, callable_):
+ self.test = test
+ self.callable_ = callable_
+ self.called_with = None
+ self.emited = False
+ self.signal = signal
+ self._asserted = False
+
+ signal.connect(self.on_signal_emited)
+
+ # Store given parameters and mark signal as `emited`
+ def on_signal_emited(self, *args, **kwargs):
+ self.called_with = (args, kwargs)
+ self.emited = True
+
+ def assertEmission(self):
+ # Assert once wheter signal was emited or not
+ was_asserted = self._asserted
+ self._asserted = True
+
+ if not was_asserted:
+ if not self.emited:
+ self.test.fail(
+ "signal %s not emited" % (_signal_name(self.signal)))
+
+ # Call given callable is necessary
+ if self.callable_:
+ args, kwargs = self.called_with
+ self.callable_(*args, **kwargs)
+
+ def __enter__(self):
+ # Assert emission when context is entered
+ self.assertEmission()
+ return self.called_with
+
+ def __exit__(self, *_):
+ return False
+
+### Unit Testing
+
+# `qunittest` does not force much abould how test should look - it just adds
+# several helpers for asynchronous code testing.
+#
+# Common test case may look like this:
+#
+# import qunittest
+# from calculator import Calculator
+#
+# class TestCalculator(qunittest.TestCase):
+# def setUp(self):
+# self.calc = Calculator()
+#
+# def test_should_add_two_numbers_synchronously(self):
+# # given
+# a, b = 2, 3
+#
+# # when
+# r = self.calc.add(a, b)
+#
+# # then
+# self.assertEqual(5, r)
+#
+# def test_should_calculate_factorial_in_background(self):
+# # given
+#
+# # when
+# self.calc.factorial(20)
+#
+# # then
+# self.assertEmited(self.calc.done) with (args, kwargs):
+# self.assertEqual([2432902008176640000], args)
+#
+# if __name__ == "__main__":
+# main()
+#
+# Test can be run by typing:
+#
+# python test_calculator.py
+#
+# Automatic test discovery is not supported now, because testing PyQt needs
+# an instance of `QApplication` and its `exec_` method is blocking.
+#
+
+
+### TestCase class
+
+class TestCase(unittest.TestCase):
+ """
+ Extends standard `unittest.TestCase` with several PyQt4 testing features
+ useful for asynchronous testing.
+ """
+ def __init__(self, *args, **kwargs):
+ super(TestCase, self).__init__(*args, **kwargs)
+
+ self._clearSignalConnectors()
+ self._succeeded = False
+ self.addCleanup(self._clearSignalConnectors)
+ self.tearDown = self._decorateTearDown(self.tearDown)
+
+ ### Protected methods
+
+ def _clearSignalConnectors(self):
+ self._connectedSignals = []
+
+ def _decorateTearDown(self, tearDown):
+ @functools.wraps(tearDown)
+ def decorator():
+ self._ensureEmitedSignals()
+ return tearDown()
+ return decorator
+
+ def _ensureEmitedSignals(self):
+ """
+ Checks if signals were acually emited. Raises AssertionError if no.
+ """
+ # TODO: add information about line
+ for signal in self._connectedSignals:
+ signal.assertEmission()
+
+ ### Assertions
+
+ def assertEmited(self, signal, callable_=None, timeout=1):
+ """
+ Asserts if given `signal` was emited. Waits 1 second by default,
+ before asserts signal emission.
+
+ If `callable_` is given, it should be a function which takes two
+ arguments: `args` and `kwargs`. It will be called after blocking
+ operation or when assertion about signal emission is made and
+ signal was emited.
+
+ When timeout is not `False`, method call is blocking, and ends
+ after `timeout` seconds. After that time, it validates wether
+ signal was emited.
+
+ When timeout is `False`, method is non blocking, and test should wait
+ for signals afterwards. Otherwise, at the end of the test, all
+ signal emissions are checked if appeared.
+
+ Function returns context, which yields to list of parameters given
+ to signal. It can be useful for testing given parameters. Following
+ code:
+
+ with self.assertEmited(widget.signal) as (args, kwargs):
+ self.assertEqual(1, len(args))
+ self.assertEqual("Hello World!", args[0])
+
+ will wait 1 second and test for correct parameters, is signal was
+ emtied.
+
+ Note that code:
+
+ with self.assertEmited(widget.signal, timeout=False) as (a, k):
+ # Will not be invoked
+
+ will always fail since signal cannot be emited in the time of its
+ connection - code inside the context will not be invoked at all.
+ """
+
+ connector = _SignalConnector(self, signal, callable_)
+ self._connectedSignals.append(connector)
+ if timeout:
+ self.waitFor(timeout)
+ connector.assertEmission()
+
+ return connector
+
+ ### Helper methods
+
+ @contextlib.contextmanager
+ def invokeAfter(self, seconds, callable_=None):
+ """
+ Waits given amount of time and executes the context.
+
+ If `callable_` is given, executes it, instead of context.
+ """
+ self.waitFor(seconds)
+ if callable_:
+ callable_()
+ else:
+ yield
+
+ def waitFor(self, seconds):
+ """
+ Waits given amount of time.
+
+ self.widget.loadImage(url)
+ self.waitFor(seconds=10)
+ """
+ QTest.qWait(seconds * 1000)
+
+ def succeed(self, bool_=True):
+ """ Marks test as suceeded for next `failAfter()` invocation. """
+ self._succeeded = self._succeeded or bool_
+
+ def failAfter(self, seconds, message=None):
+ """
+ Waits given amount of time, and fails the test if `succeed(bool)`
+ is not called - in most common case, `succeed(bool)` should be called
+ asynchronously (in signal handler):
+
+ self.widget.signal.connect(lambda: self.succeed())
+ self.failAfter(1, "signal not emited?")
+
+ After invocation, test is no longer consider as succeeded.
+ """
+ self.waitFor(seconds)
+ if not self._succeeded:
+ self.fail(message)
+
+ self._succeeded = False
+
+### Credits
+#
+# * **Who is responsible:** [Dawid Fatyga][df]
+# * **Source:** [GitHub][gh]
+# * **Doc. generator:** [rocco][ro]
+#
+# [gh]: https://www.github.com/dejw/qunittest
+# [df]: https://github.com/dejw
+# [ro]: http://rtomayko.github.com/rocco/
+#