summaryrefslogtreecommitdiff
path: root/src/leap/testing/qunittest.py
diff options
context:
space:
mode:
authorkali <kali@leap.se>2013-02-15 09:31:51 +0900
committerkali <kali@leap.se>2013-02-15 09:31:51 +0900
commit9cea9c8a34343f8792d65b96f93ae22bd8685878 (patch)
tree9f512367b1d47ced5614702a00f3ff0a8fe746d7 /src/leap/testing/qunittest.py
parent7159734ec6c0b76fc7f3737134cd22fdaaaa7d58 (diff)
parent1032e07a50c8bb265ff9bd31b3bb00e83ddb451e (diff)
Merge branch 'release/v0.2.0'
Conflicts: README.txt
Diffstat (limited to 'src/leap/testing/qunittest.py')
-rw-r--r--src/leap/testing/qunittest.py302
1 files changed, 302 insertions, 0 deletions
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/
+#