testtools-dev team mailing list archive
-
testtools-dev team
-
Mailing list archive
-
Message #00108
[Merge] lp:~jml/testtools/deferred-support into lp:testtools
Jonathan Lange has proposed merging lp:~jml/testtools/deferred-support into lp:testtools.
Requested reviews:
testtools developers (testtools-dev)
This branch adds experimental support for tests that return Deferreds.
Most of the change in the diff is in the new RunTest objects and the tests I've added for those. It's my hope that the code is self-documenting, and that any questionable implementation decisions have explanatory comments. Let me know if it's otherwise.
The changes in the rest of testtools are just to make it a little more friendly to new RunTest objects. I've split off the RunTest code for handling user exceptions so I can re-use it in the async code, and I've made more methods on TestCase return things.
Since RunTest objects are actually a bit of a pain to use (see bug 657780 as an example), I haven't yet tried these new runners in anger. I might try them with some of the Launchpad tests and see what happens. When I do, I'll update the MP.
--
https://code.launchpad.net/~jml/testtools/deferred-support/+merge/38080
Your team testtools developers is requested to review the proposed merge of lp:~jml/testtools/deferred-support into lp:testtools.
=== modified file 'NEWS'
--- NEWS 2010-09-18 02:10:58 +0000
+++ NEWS 2010-10-10 16:49:39 +0000
@@ -16,6 +16,12 @@
* In normal circumstances, a TestCase will no longer share details with clones
of itself. (Andrew Bennetts, bug #637725)
+* Experimental support for running tests that return Deferreds.
+ (Jonathan Lange)
+
+* 'runTest' can now be passed to TestCase to specify a RunTest object to use.
+ (Jonathan Lange, bug #657760)
+
0.9.6
~~~~~
=== added file 'testtools/deferredruntest.py'
--- testtools/deferredruntest.py 1970-01-01 00:00:00 +0000
+++ testtools/deferredruntest.py 2010-10-10 16:49:39 +0000
@@ -0,0 +1,372 @@
+# Copyright (c) 2010 Jonathan M. Lange. See LICENSE for details.
+
+"""Individual test case execution for tests that return Deferreds."""
+
+__all__ = [
+ 'AsynchronousDeferredRunTest',
+ 'SynchronousDeferredRunTest',
+ ]
+
+import signal
+import sys
+
+from testtools.runtest import RunTest
+
+from twisted.internet import defer
+from twisted.internet.interfaces import IReactorThreads
+from twisted.python.failure import Failure
+from twisted.python.util import mergeFunctionMetadata
+
+
+class DeferredNotFired(Exception):
+ """Raised when we extract a result from a Deferred that's not fired yet."""
+
+
+class UnhandledErrorInDeferred(Exception):
+ """Raised when there are unhandlede errors in Deferreds.
+
+ If you are getting this error then you are probably either not returning a
+ Deferred from a function that makes one, or you are not adding an errback
+ to a Deferred. Or both. Use `Deferred.DEBUG` to get more information.
+ """
+
+ def __init__(self, debug_infos):
+ super(UnhandledErrorInDeferred, self).__init__(
+ "Unhandled error in Deferreds: %r" % (
+ [info.failResult for info in debug_infos]))
+
+
+def extract_result(deferred):
+ """Extract the result from a fired deferred.
+
+ It can happen that you have an API that returns Deferreds for
+ compatibility with Twisted code, but is in fact synchronous, i.e. the
+ Deferreds it returns have always fired by the time it returns. In this
+ case, you can use this function to convert the result back into the usual
+ form for a synchronous API, i.e. the result itself or a raised exception.
+
+ It would be very bad form to use this as some way of checking if a
+ Deferred has fired.
+ """
+ failures = []
+ successes = []
+ deferred.addCallbacks(successes.append, failures.append)
+ if len(failures) == 1:
+ failures[0].raiseException()
+ elif len(successes) == 1:
+ return successes[0]
+ else:
+ raise DeferredNotFired("%r has not fired yet." % (deferred,))
+
+
+def trap_unhandled_errors(function, *args, **kwargs):
+ """Run a function, trapping any unhandled errors in Deferreds.
+
+ Assumes that 'function' will have handled any errors in Deferreds by the
+ time it is complete. This is almost never true of any Twisted code, since
+ you can never tell when someone has added an errback to a Deferred.
+
+ If 'function' raises, then don't bother doing any unhandled error
+ jiggery-pokery, since something horrible has probably happened anyway.
+
+ :return: A tuple of '(result, error)', where 'result' is the value returned
+ by 'function' and 'error' is a list of `defer.DebugInfo` objects that
+ have unhandled errors in Deferreds.
+ """
+ real_DebugInfo = defer.DebugInfo
+ debug_infos = []
+ def DebugInfo():
+ info = real_DebugInfo()
+ debug_infos.append(info)
+ return info
+ defer.DebugInfo = DebugInfo
+ try:
+ result = function(*args, **kwargs)
+ finally:
+ defer.DebugInfo = real_DebugInfo
+ errors = []
+ for info in debug_infos:
+ if info.failResult is not None:
+ errors.append(info)
+ # Disable the destructor that logs to error. We are already
+ # catching the error here.
+ info.__del__ = lambda: None
+ return result, errors
+
+
+class SynchronousDeferredRunTest(RunTest):
+ """Runner for tests that return synchronous Deferreds."""
+
+ def _run_user(self, function, *args):
+ d = defer.maybeDeferred(function, *args)
+ def got_exception(failure):
+ return self._got_user_exception(
+ (failure.type, failure.value, failure.tb))
+ d.addErrback(got_exception)
+ result = extract_result(d)
+ return result
+
+
+class AsynchronousDeferredRunTest(RunTest):
+ """Runner for tests that return Deferreds that fire asynchronously.
+
+ That is, this test runner assumes that the Deferreds will only fire if the
+ reactor is left to spin for a while.
+
+ Do not rely too heavily on the nuances of the behaviour of this class.
+ What it does to the reactor is black magic, and if we can find nicer ways
+ of doing it we will gladly break backwards compatibility.
+
+ This is highly experimental code. Use at your own risk.
+ """
+
+ def __init__(self, case, handlers=None, reactor=None, timeout=0.005):
+ super(AsynchronousDeferredRunTest, self).__init__(case, handlers)
+ if reactor is None:
+ from twisted.internet import reactor
+ self._reactor = reactor
+ self._timeout = timeout
+
+ @classmethod
+ def make_factory(cls, reactor, timeout):
+ return lambda case, handlers=None: AsynchronousDeferredRunTest(
+ case, handlers, reactor, timeout)
+
+ @defer.inlineCallbacks
+ def _run_cleanups(self):
+ """Run the cleanups on the test case.
+
+ We expect that the cleanups on the test case can also return
+ asynchronous Deferreds. As such, we take the responsibility for
+ running the cleanups, rather than letting TestCase do it.
+ """
+ while self.case._cleanups:
+ f, args, kwargs = self.case._cleanups.pop()
+ try:
+ yield defer.maybeDeferred(f, *args, **kwargs)
+ except:
+ exc_info = sys.exc_info()
+ self.case._report_traceback(exc_info)
+ last_exception = exc_info[1]
+ defer.returnValue(last_exception)
+
+ def _run_deferred(self):
+ """Run the test, assuming everything in it is Deferred-returning.
+
+ This should return a Deferred that fires with True if the test was
+ successful and False if the test was not successful. It should *not*
+ call addSuccess on the result, because there's reactor clean up that
+ we needs to be done afterwards.
+ """
+ fails = []
+
+ def fail_if_exception_caught(exception_caught):
+ if self.exception_caught == exception_caught:
+ fails.append(None)
+
+ def clean_up(ignored=None):
+ """Run the cleanups."""
+ d = self._run_cleanups()
+ def clean_up_done(result):
+ if result is not None:
+ self._exceptions.append(result)
+ fails.append(None)
+ return d.addCallback(clean_up_done)
+
+ def set_up_done(exception_caught):
+ """Set up is done, either clean up or run the test."""
+ if self.exception_caught == exception_caught:
+ fails.append(None)
+ return clean_up()
+ else:
+ d = self._run_user(self.case._run_test_method, self.result)
+ d.addCallback(fail_if_exception_caught)
+ d.addBoth(tear_down)
+ return d
+
+ def tear_down(ignored):
+ d = self._run_user(self.case._run_teardown, self.result)
+ d.addCallback(fail_if_exception_caught)
+ d.addBoth(clean_up)
+ return d
+
+ d = self._run_user(self.case._run_setup, self.result)
+ d.addCallback(set_up_done)
+ d.addBoth(lambda ignored: len(fails) == 0)
+ return d
+
+ def _run_core(self):
+ spinner = _Spinner(self._reactor)
+ # XXX: This can call addError on result multiple times. Not sure if
+ # this is a good idea.
+ successful, unhandled = trap_unhandled_errors(
+ spinner.run, self._timeout, self._run_deferred)
+ if unhandled:
+ successful = False
+ try:
+ raise UnhandledErrorInDeferred(unhandled)
+ except UnhandledErrorInDeferred:
+ self._got_user_exception(sys.exc_info())
+ junk = spinner.clean()
+ if junk:
+ successful = False
+ try:
+ raise UncleanReactorError(junk)
+ except UncleanReactorError:
+ self._got_user_exception(sys.exc_info())
+ if successful:
+ self.result.addSuccess(self.case, details=self.case.getDetails())
+
+ def _run_user(self, function, *args):
+ # XXX: I think this traps KeyboardInterrupt, and I think this is a bad
+ # thing. Perhaps we should have a maybeDeferred-like thing that
+ # re-raises KeyboardInterrupt. Or, we should have our own exception
+ # handler that stops the test run in the case of KeyboardInterrupt. But
+ # of course, the reactor installs a SIGINT handler anyway.
+ return defer.maybeDeferred(
+ super(AsynchronousDeferredRunTest, self)._run_user,
+ function, *args)
+
+
+class ReentryError(Exception):
+ """Raised when we try to re-enter a function that forbids it."""
+
+ def __init__(self, function):
+ super(ReentryError, self).__init__(
+ "%r in not re-entrant but was called within a call to itself."
+ % (function,))
+
+
+class UncleanReactorError(Exception):
+ """Raised when the reactor has junk in it."""
+
+ def __init__(self, junk):
+ super(UncleanReactorError, self).__init__(
+ "The reactor still thinks it needs to do things. Close all "
+ "connections, kill all processes and make sure all delayed "
+ "calls have either fired or been cancelled. The management "
+ "thanks you: %s"
+ % map(repr, junk))
+
+
+def not_reentrant(function, _calls={}):
+ """Decorates a function as not being re-entrant.
+
+ The decorated function will raise an error if called from within itself.
+ """
+ def decorated(*args, **kwargs):
+ if _calls.get(function, False):
+ raise ReentryError(function)
+ _calls[function] = True
+ try:
+ return function(*args, **kwargs)
+ finally:
+ _calls[function] = False
+ return mergeFunctionMetadata(function, decorated)
+
+
+class TimeoutError(Exception):
+ """Raised when run_in_reactor takes too long to run a function."""
+
+
+class _Spinner(object):
+ """Spin the reactor until a function is done.
+
+ This class emulates the behaviour of twisted.trial in that it grotesquely
+ and horribly spins the Twisted reactor while a function is running, and
+ then kills the reactor when that function is complete and all the
+ callbacks in its chains are done.
+ """
+
+ _UNSET = object()
+
+ # Signals that we save and restore for each spin.
+ _PRESERVED_SIGNALS = [
+ signal.SIGINT,
+ signal.SIGTERM,
+ signal.SIGCHLD,
+ ]
+
+ def __init__(self, reactor):
+ self._reactor = reactor
+ self._timeout_call = None
+ self._success = self._UNSET
+ self._failure = self._UNSET
+ self._saved_signals = []
+
+ def _cancel_timeout(self):
+ if self._timeout_call:
+ self._timeout_call.cancel()
+
+ def _get_result(self):
+ if self._failure is not self._UNSET:
+ self._failure.raiseException()
+ if self._success is not self._UNSET:
+ return self._success
+ raise AssertionError("Tried to get result when no result is available.")
+
+ def _got_failure(self, result):
+ self._cancel_timeout()
+ self._failure = result
+
+ def _got_success(self, result):
+ self._cancel_timeout()
+ self._success = result
+
+ def _stop_reactor(self, ignored=None):
+ """Stop the reactor!"""
+ self._reactor.crash()
+
+ def _timed_out(self, function, timeout):
+ e = TimeoutError(
+ "%r took longer than %s seconds" % (function, timeout))
+ self._failure = Failure(e)
+ self._stop_reactor()
+
+ def clean(self):
+ """Clean up any junk in the reactor."""
+ junk = []
+ for delayed_call in self._reactor.getDelayedCalls():
+ delayed_call.cancel()
+ junk.append(delayed_call)
+ for selectable in self._reactor.removeAll():
+ # Twisted sends a 'KILL' signal to selectables that provide
+ # IProcessTransport. Since only _dumbwin32proc processes do this,
+ # we aren't going to bother.
+ junk.append(selectable)
+ # XXX: Not tested. Not sure that the cost of testing this reliably
+ # outweighs the benefits.
+ if IReactorThreads.providedBy(self._reactor):
+ self._reactor.suggestThreadPoolSize(0)
+ if self._reactor.threadpool is not None:
+ self._reactor._stopThreadPool()
+ return junk
+
+ def _save_signals(self):
+ self._saved_signals = [
+ (sig, signal.getsignal(sig)) for sig in self._PRESERVED_SIGNALS]
+
+ def _restore_signals(self):
+ for sig, hdlr in self._saved_signals:
+ signal.signal(sig, hdlr)
+ self._saved_signals = []
+
+ @not_reentrant
+ def run(self, timeout, function, *args, **kwargs):
+ """Run 'function' in a reactor.
+
+ If 'function' returns a Deferred, the reactor will keep spinning until
+ the Deferred fires and its chain completes or until the timeout is
+ reached -- whichever comes first.
+ """
+ self._save_signals()
+ self._timeout_call = self._reactor.callLater(
+ timeout, self._timed_out, function, timeout)
+ def run_function():
+ d = defer.maybeDeferred(function, *args, **kwargs)
+ d.addCallbacks(self._got_success, self._got_failure)
+ d.addBoth(self._stop_reactor)
+ self._reactor.callWhenRunning(run_function)
+ self._reactor.run()
+ self._restore_signals()
+ return self._get_result()
=== modified file 'testtools/runtest.py'
--- testtools/runtest.py 2010-07-29 18:20:02 +0000
+++ testtools/runtest.py 2010-10-10 16:49:39 +0000
@@ -2,7 +2,6 @@
"""Individual test case execution."""
-__metaclass__ = type
__all__ = [
'RunTest',
]
@@ -145,11 +144,14 @@
except KeyboardInterrupt:
raise
except:
- exc_info = sys.exc_info()
- e = exc_info[1]
- self.case.onException(exc_info)
- for exc_class, handler in self.handlers:
- if isinstance(e, exc_class):
- self._exceptions.append(e)
- return self.exception_caught
- raise e
+ return self._got_user_exception(sys.exc_info())
+
+ def _got_user_exception(self, exc_info):
+ """Called when user code raises an exception."""
+ e = exc_info[1]
+ self.case.onException(exc_info)
+ for exc_class, handler in self.handlers:
+ if isinstance(e, exc_class):
+ self._exceptions.append(e)
+ return self.exception_caught
+ raise e
=== modified file 'testtools/testcase.py'
--- testtools/testcase.py 2010-09-18 02:10:58 +0000
+++ testtools/testcase.py 2010-10-10 16:49:39 +0000
@@ -86,6 +86,7 @@
supplied testtools.runtest.RunTest is used. The instance to be
used is created when run() is invoked, so will be fresh each time.
"""
+ self.__RunTest = kwargs.pop('runTest', RunTest)
unittest.TestCase.__init__(self, *args, **kwargs)
self._cleanups = []
self._unique_id_gen = itertools.count(1)
@@ -95,7 +96,6 @@
# __details is lazy-initialized so that a constructed-but-not-run
# TestCase is safe to use with clone_test_with_new_id.
self.__details = None
- self.__RunTest = kwargs.get('runTest', RunTest)
self.__exception_handlers = []
self.exception_handlers = [
(self.skipException, self._report_skip),
@@ -440,13 +440,14 @@
:raises ValueError: If the base class setUp is not called, a
ValueError is raised.
"""
- self.setUp()
+ ret = self.setUp()
if not self.__setup_called:
raise ValueError(
"TestCase.setUp was not called. Have you upcalled all the "
"way up the hierarchy from your setUp? e.g. Call "
"super(%s, self).setUp() from your setUp()."
% self.__class__.__name__)
+ return ret
def _run_teardown(self, result):
"""Run the tearDown function for this test.
@@ -455,13 +456,14 @@
:raises ValueError: If the base class tearDown is not called, a
ValueError is raised.
"""
- self.tearDown()
+ ret = self.tearDown()
if not self.__teardown_called:
raise ValueError(
"TestCase.tearDown was not called. Have you upcalled all the "
"way up the hierarchy from your tearDown? e.g. Call "
"super(%s, self).tearDown() from your tearDown()."
% self.__class__.__name__)
+ return ret
def _run_test_method(self, result):
"""Run the test method for this test.
@@ -476,7 +478,7 @@
# Python 2.4
method_name = getattr(self, '_TestCase__testMethodName')
testMethod = getattr(self, method_name)
- testMethod()
+ return testMethod()
def setUp(self):
unittest.TestCase.setUp(self)
=== modified file 'testtools/tests/__init__.py'
--- testtools/tests/__init__.py 2010-08-04 12:45:22 +0000
+++ testtools/tests/__init__.py 2010-10-10 16:49:39 +0000
@@ -7,6 +7,7 @@
test_compat,
test_content,
test_content_type,
+ test_deferredruntest,
test_matchers,
test_monkey,
test_runtest,
@@ -22,6 +23,7 @@
test_compat,
test_content,
test_content_type,
+ test_deferredruntest,
test_matchers,
test_monkey,
test_runtest,
=== added file 'testtools/tests/test_deferredruntest.py'
--- testtools/tests/test_deferredruntest.py 1970-01-01 00:00:00 +0000
+++ testtools/tests/test_deferredruntest.py 2010-10-10 16:49:39 +0000
@@ -0,0 +1,567 @@
+# Copyright (c) 2010 Jonathan M. Lange. See LICENSE for details.
+
+"""Tests for the DeferredRunTest single test execution logic."""
+
+import signal
+
+from testtools import (
+ TestCase,
+ )
+from testtools.deferredruntest import (
+ AsynchronousDeferredRunTest,
+ DeferredNotFired,
+ extract_result,
+ not_reentrant,
+ ReentryError,
+ _Spinner,
+ SynchronousDeferredRunTest,
+ TimeoutError,
+ trap_unhandled_errors,
+ )
+from testtools.tests.helpers import ExtendedTestResult
+from testtools.matchers import (
+ Equals,
+ Is,
+ )
+from testtools.runtest import RunTest
+
+from twisted.internet import defer
+from twisted.python.failure import Failure
+
+
+class TestExtractResult(TestCase):
+
+ def test_not_fired(self):
+ # extract_result raises DeferredNotFired if it's given a Deferred that
+ # has not fired.
+ self.assertRaises(DeferredNotFired, extract_result, defer.Deferred())
+
+ def test_success(self):
+ # extract_result returns the value of the Deferred if it has fired
+ # successfully.
+ marker = object()
+ d = defer.succeed(marker)
+ self.assertThat(extract_result(d), Equals(marker))
+
+ def test_failure(self):
+ # extract_result raises the failure's exception if it's given a
+ # Deferred that is failing.
+ try:
+ 1/0
+ except ZeroDivisionError:
+ f = Failure()
+ d = defer.fail(f)
+ self.assertRaises(ZeroDivisionError, extract_result, d)
+
+
+class TestNotReentrant(TestCase):
+
+ def test_not_reentrant(self):
+ # A function decorated as not being re-entrant will raise a
+ # ReentryError if it is called while it is running.
+ calls = []
+ @not_reentrant
+ def log_something():
+ calls.append(None)
+ if len(calls) < 5:
+ log_something()
+ self.assertRaises(ReentryError, log_something)
+ self.assertEqual(1, len(calls))
+
+ def test_deeper_stack(self):
+ calls = []
+ @not_reentrant
+ def g():
+ calls.append(None)
+ if len(calls) < 5:
+ f()
+ @not_reentrant
+ def f():
+ calls.append(None)
+ if len(calls) < 5:
+ g()
+ self.assertRaises(ReentryError, f)
+ self.assertEqual(2, len(calls))
+
+
+class TestTrapUnhandledErrors(TestCase):
+
+ def test_no_deferreds(self):
+ marker = object()
+ result, errors = trap_unhandled_errors(lambda: marker)
+ self.assertEqual([], errors)
+ self.assertIs(marker, result)
+
+ def test_unhandled_error(self):
+ failures = []
+ def make_deferred_but_dont_handle():
+ try:
+ 1/0
+ except ZeroDivisionError:
+ f = Failure()
+ failures.append(f)
+ defer.fail(f)
+ result, errors = trap_unhandled_errors(make_deferred_but_dont_handle)
+ self.assertIs(None, result)
+ self.assertEqual(failures, [error.failResult for error in errors])
+
+
+class X(object):
+ """Tests that we run as part of our tests, nested to avoid discovery."""
+
+ class Base(TestCase):
+ def setUp(self):
+ super(X.Base, self).setUp()
+ self.calls = ['setUp']
+ self.addCleanup(self.calls.append, 'clean-up')
+ def test_something(self):
+ self.calls.append('test')
+ def tearDown(self):
+ self.calls.append('tearDown')
+ super(X.Base, self).tearDown()
+
+ class Success(Base):
+ expected_calls = ['setUp', 'test', 'tearDown', 'clean-up']
+ expected_results = [['addSuccess']]
+
+ class ErrorInSetup(Base):
+ expected_calls = ['setUp', 'clean-up']
+ expected_results = [('addError', RuntimeError)]
+ def setUp(self):
+ super(X.ErrorInSetup, self).setUp()
+ raise RuntimeError("Error in setUp")
+
+ class ErrorInTest(Base):
+ expected_calls = ['setUp', 'tearDown', 'clean-up']
+ expected_results = [('addError', RuntimeError)]
+ def test_something(self):
+ raise RuntimeError("Error in test")
+
+ class FailureInTest(Base):
+ expected_calls = ['setUp', 'tearDown', 'clean-up']
+ expected_results = [('addFailure', AssertionError)]
+ def test_something(self):
+ self.fail("test failed")
+
+ class ErrorInTearDown(Base):
+ expected_calls = ['setUp', 'test', 'clean-up']
+ expected_results = [('addError', RuntimeError)]
+ def tearDown(self):
+ raise RuntimeError("Error in tearDown")
+
+ class ErrorInCleanup(Base):
+ expected_calls = ['setUp', 'test', 'tearDown', 'clean-up']
+ expected_results = [('addError', ZeroDivisionError)]
+ def test_something(self):
+ self.calls.append('test')
+ self.addCleanup(lambda: 1/0)
+
+ class TestIntegration(TestCase):
+
+ def assertResultsMatch(self, test, result):
+ events = list(result._events)
+ self.assertEqual(('startTest', test), events.pop(0))
+ for expected_result in test.expected_results:
+ result = events.pop(0)
+ if len(expected_result) == 1:
+ self.assertEqual((expected_result[0], test), result)
+ else:
+ self.assertEqual((expected_result[0], test), result[:2])
+ error_type = expected_result[1]
+ self.assertIn(error_type.__name__, str(result[2]))
+ self.assertEqual([('stopTest', test)], events)
+
+ def test_runner(self):
+ result = ExtendedTestResult()
+ test = self.test_factory('test_something', runTest=self.runner)
+ test.run(result)
+ self.assertEqual(test.calls, self.test_factory.expected_calls)
+ self.assertResultsMatch(test, result)
+
+
+def make_integration_tests():
+ from unittest import TestSuite
+ from testtools import clone_test_with_new_id
+ runners = [
+ RunTest,
+ SynchronousDeferredRunTest,
+ AsynchronousDeferredRunTest,
+ ]
+
+ tests = [
+ X.Success,
+ X.ErrorInSetup,
+ X.ErrorInTest,
+ X.ErrorInTearDown,
+ X.FailureInTest,
+ X.ErrorInCleanup,
+ ]
+ base_test = X.TestIntegration('test_runner')
+ integration_tests = []
+ for runner in runners:
+ for test in tests:
+ new_test = clone_test_with_new_id(
+ base_test, '%s(%s, %s)' % (
+ base_test.id(),
+ runner.__name__,
+ test.__name__))
+ new_test.test_factory = test
+ new_test.runner = runner
+ integration_tests.append(new_test)
+ return TestSuite(integration_tests)
+
+
+class TestSynchronousDeferredRunTest(TestCase):
+
+ def make_result(self):
+ return ExtendedTestResult()
+
+ def make_runner(self, test):
+ return SynchronousDeferredRunTest(test, test.exception_handlers)
+
+ def test_success(self):
+ class SomeCase(TestCase):
+ def test_success(self):
+ return defer.succeed(None)
+ test = SomeCase('test_success')
+ runner = self.make_runner(test)
+ result = self.make_result()
+ runner.run(result)
+ self.assertThat(
+ result._events, Equals([
+ ('startTest', test),
+ ('addSuccess', test),
+ ('stopTest', test)]))
+
+ def test_failure(self):
+ class SomeCase(TestCase):
+ def test_failure(self):
+ return defer.maybeDeferred(self.fail, "Egads!")
+ test = SomeCase('test_failure')
+ runner = self.make_runner(test)
+ result = self.make_result()
+ runner.run(result)
+ self.assertThat(
+ [event[:2] for event in result._events], Equals([
+ ('startTest', test),
+ ('addFailure', test),
+ ('stopTest', test)]))
+
+ def test_setUp_followed_by_test(self):
+ class SomeCase(TestCase):
+ def setUp(self):
+ super(SomeCase, self).setUp()
+ return defer.succeed(None)
+ def test_failure(self):
+ return defer.maybeDeferred(self.fail, "Egads!")
+ test = SomeCase('test_failure')
+ runner = self.make_runner(test)
+ result = self.make_result()
+ runner.run(result)
+ self.assertThat(
+ [event[:2] for event in result._events], Equals([
+ ('startTest', test),
+ ('addFailure', test),
+ ('stopTest', test)]))
+
+
+class TestAsynchronousDeferredRunTest(TestCase):
+
+ def make_reactor(self):
+ from twisted.internet import reactor
+ return reactor
+
+ def make_result(self):
+ return ExtendedTestResult()
+
+ def make_runner(self, test, timeout=None):
+ if timeout is None:
+ timeout = self.make_timeout()
+ return AsynchronousDeferredRunTest(
+ test, test.exception_handlers, timeout=timeout)
+
+ def make_timeout(self):
+ return 0.005
+
+ def test_setUp_returns_deferred_that_fires_later(self):
+ # setUp can return a Deferred that might fire at any time.
+ # AsynchronousDeferredRunTest will not go on to running the test until
+ # the Deferred returned by setUp actually fires.
+ call_log = []
+ marker = object()
+ d = defer.Deferred().addCallback(call_log.append)
+ class SomeCase(TestCase):
+ def setUp(self):
+ super(SomeCase, self).setUp()
+ call_log.append('setUp')
+ return d
+ def test_something(self):
+ call_log.append('test')
+ def fire_deferred():
+ self.assertThat(call_log, Equals(['setUp']))
+ d.callback(marker)
+ test = SomeCase('test_something')
+ timeout = self.make_timeout()
+ runner = self.make_runner(test, timeout=timeout)
+ result = self.make_result()
+ reactor = self.make_reactor()
+ reactor.callLater(timeout, fire_deferred)
+ runner.run(result)
+ self.assertThat(call_log, Equals(['setUp', marker, 'test']))
+
+ def test_calls_setUp_test_tearDown_in_sequence(self):
+ # setUp, the test method and tearDown can all return
+ # Deferreds. AsynchronousDeferredRunTest will make sure that each of
+ # these are run in turn, only going on to the next stage once the
+ # Deferred from the previous stage has fired.
+ call_log = []
+ a = defer.Deferred()
+ a.addCallback(lambda x: call_log.append('a'))
+ b = defer.Deferred()
+ b.addCallback(lambda x: call_log.append('b'))
+ c = defer.Deferred()
+ c.addCallback(lambda x: call_log.append('c'))
+ class SomeCase(TestCase):
+ def setUp(self):
+ super(SomeCase, self).setUp()
+ call_log.append('setUp')
+ return a
+ def test_success(self):
+ call_log.append('test')
+ return b
+ def tearDown(self):
+ super(SomeCase, self).tearDown()
+ call_log.append('tearDown')
+ return c
+ test = SomeCase('test_success')
+ timeout = self.make_timeout()
+ runner = self.make_runner(test, timeout)
+ result = self.make_result()
+ reactor = self.make_reactor()
+ def fire_a():
+ self.assertThat(call_log, Equals(['setUp']))
+ a.callback(None)
+ def fire_b():
+ self.assertThat(call_log, Equals(['setUp', 'a', 'test']))
+ b.callback(None)
+ def fire_c():
+ self.assertThat(
+ call_log, Equals(['setUp', 'a', 'test', 'b', 'tearDown']))
+ c.callback(None)
+ reactor.callLater(timeout * 0.25, fire_a)
+ reactor.callLater(timeout * 0.5, fire_b)
+ reactor.callLater(timeout * 0.75, fire_c)
+ runner.run(result)
+ self.assertThat(
+ call_log, Equals(['setUp', 'a', 'test', 'b', 'tearDown', 'c']))
+
+ def test_async_cleanups(self):
+ # Cleanups added with addCleanup can return
+ # Deferreds. AsynchronousDeferredRunTest will run each of them in
+ # turn.
+ class SomeCase(TestCase):
+ def test_whatever(self):
+ pass
+ test = SomeCase('test_whatever')
+ log = []
+ a = defer.Deferred().addCallback(lambda x: log.append('a'))
+ b = defer.Deferred().addCallback(lambda x: log.append('b'))
+ c = defer.Deferred().addCallback(lambda x: log.append('c'))
+ test.addCleanup(lambda: a)
+ test.addCleanup(lambda: b)
+ test.addCleanup(lambda: c)
+ def fire_a():
+ self.assertThat(log, Equals([]))
+ a.callback(None)
+ def fire_b():
+ self.assertThat(log, Equals(['a']))
+ b.callback(None)
+ def fire_c():
+ self.assertThat(log, Equals(['a', 'b']))
+ c.callback(None)
+ timeout = self.make_timeout()
+ reactor = self.make_reactor()
+ reactor.callLater(timeout * 0.25, fire_a)
+ reactor.callLater(timeout * 0.5, fire_b)
+ reactor.callLater(timeout * 0.75, fire_c)
+ runner = self.make_runner(test, timeout)
+ result = self.make_result()
+ runner.run(result)
+ self.assertThat(log, Equals(['a', 'b', 'c']))
+
+ def test_clean_reactor(self):
+ # If there's cruft left over in the reactor, the test fails.
+ reactor = self.make_reactor()
+ timeout = self.make_timeout()
+ class SomeCase(TestCase):
+ def test_cruft(self):
+ reactor.callLater(timeout * 2.0, lambda: None)
+ test = SomeCase('test_cruft')
+ runner = self.make_runner(test, timeout)
+ result = self.make_result()
+ runner.run(result)
+ error = result._events[1][2]
+ result._events[1] = ('addError', test, None)
+ self.assertThat(result._events, Equals(
+ [('startTest', test),
+ ('addError', test, None),
+ ('stopTest', test)]))
+ self.assertThat(list(error.keys()), Equals(['traceback']))
+
+ def test_unhandled_error_from_deferred(self):
+ # If there's a Deferred with an unhandled error, the test fails.
+ class SomeCase(TestCase):
+ def test_cruft(self):
+ # Note we aren't returning the Deferred so that the error will
+ # be unhandled.
+ defer.maybeDeferred(lambda: 1/0)
+ test = SomeCase('test_cruft')
+ runner = self.make_runner(test)
+ result = self.make_result()
+ runner.run(result)
+ error = result._events[1][2]
+ result._events[1] = ('addError', test, None)
+ self.assertThat(result._events, Equals(
+ [('startTest', test),
+ ('addError', test, None),
+ ('stopTest', test)]))
+ self.assertThat(list(error.keys()), Equals(['traceback']))
+
+ def test_convenient_construction(self):
+ # As a convenience method, AsynchronousDeferredRunTest has a
+ # classmethod that returns an AsynchronousDeferredRunTest
+ # factory. This factory has the same API as the RunTest constructor.
+ reactor = object()
+ timeout = object()
+ handler = object()
+ factory = AsynchronousDeferredRunTest.make_factory(reactor, timeout)
+ runner = factory(self, [handler])
+ self.assertIs(reactor, runner._reactor)
+ self.assertIs(timeout, runner._timeout)
+ self.assertIs(self, runner.case)
+ self.assertEqual([handler], runner.handlers)
+
+
+class TestRunInReactor(TestCase):
+
+ def make_reactor(self):
+ from twisted.internet import reactor
+ return reactor
+
+ def make_spinner(self, reactor=None):
+ if reactor is None:
+ reactor = self.make_reactor()
+ return _Spinner(reactor)
+
+ def make_timeout(self):
+ return 0.01
+
+ def test_function_called(self):
+ # run_in_reactor actually calls the function given to it.
+ calls = []
+ marker = object()
+ self.make_spinner().run(self.make_timeout(), calls.append, marker)
+ self.assertThat(calls, Equals([marker]))
+
+ def test_return_value_returned(self):
+ # run_in_reactor returns the value returned by the function given to
+ # it.
+ marker = object()
+ result = self.make_spinner().run(self.make_timeout(), lambda: marker)
+ self.assertThat(result, Is(marker))
+
+ def test_exception_reraised(self):
+ # If the given function raises an error, run_in_reactor re-raises that
+ # error.
+ self.assertRaises(
+ ZeroDivisionError,
+ self.make_spinner().run, self.make_timeout(), lambda: 1 / 0)
+
+ def test_keyword_arguments(self):
+ # run_in_reactor passes keyword arguments on.
+ calls = []
+ function = lambda *a, **kw: calls.extend([a, kw])
+ self.make_spinner().run(self.make_timeout(), function, foo=42)
+ self.assertThat(calls, Equals([(), {'foo': 42}]))
+
+ def test_not_reentrant(self):
+ # run_in_reactor raises an error if it is called inside another call
+ # to run_in_reactor.
+ spinner = self.make_spinner()
+ self.assertRaises(
+ ReentryError,
+ spinner.run, self.make_timeout(),
+ spinner.run, self.make_timeout(), lambda: None)
+
+ def test_deferred_value_returned(self):
+ # If the given function returns a Deferred, run_in_reactor returns the
+ # value in the Deferred at the end of the callback chain.
+ marker = object()
+ result = self.make_spinner().run(
+ self.make_timeout(), lambda: defer.succeed(marker))
+ self.assertThat(result, Is(marker))
+
+ def test_preserve_signal_handler(self):
+ signals = [signal.SIGINT, signal.SIGTERM, signal.SIGCHLD]
+ for sig in signals:
+ self.addCleanup(signal.signal, sig, signal.getsignal(sig))
+ new_hdlrs = [lambda *a: None, lambda *a: None, lambda *a: None]
+ for sig, hdlr in zip(signals, new_hdlrs):
+ signal.signal(sig, hdlr)
+ spinner = self.make_spinner()
+ spinner.run(self.make_timeout(), lambda: None)
+ self.assertEqual(new_hdlrs, map(signal.getsignal, signals))
+
+ def test_timeout(self):
+ # If the function takes too long to run, we raise a TimeoutError.
+ timeout = self.make_timeout()
+ self.assertRaises(
+ TimeoutError,
+ self.make_spinner().run, timeout, lambda: defer.Deferred())
+
+ def test_clean_do_nothing(self):
+ # If there's nothing going on in the reactor, then clean does nothing
+ # and returns an empty list.
+ spinner = self.make_spinner()
+ result = spinner.clean()
+ self.assertThat(result, Equals([]))
+
+ def test_clean_delayed_call(self):
+ # If there's a delayed call in the reactor, then clean cancels it and
+ # returns an empty list.
+ reactor = self.make_reactor()
+ spinner = self.make_spinner(reactor)
+ call = reactor.callLater(10, lambda: None)
+ results = spinner.clean()
+ self.assertThat(results, Equals([call]))
+ self.assertThat(call.active(), Equals(False))
+
+ def test_clean_delayed_call_cancelled(self):
+ # If there's a delayed call that's just been cancelled, then it's no
+ # longer there.
+ reactor = self.make_reactor()
+ spinner = self.make_spinner(reactor)
+ call = reactor.callLater(10, lambda: None)
+ call.cancel()
+ results = spinner.clean()
+ self.assertThat(results, Equals([]))
+
+ def test_clean_selectables(self):
+ # If there's still a selectable (e.g. a listening socket), then
+ # clean() removes it from the reactor's registry.
+ #
+ # Note that the socket is left open. This emulates a bug in trial.
+ from twisted.internet.protocol import ServerFactory
+ reactor = self.make_reactor()
+ spinner = self.make_spinner(reactor)
+ port = reactor.listenTCP(0, ServerFactory())
+ spinner.run(self.make_timeout(), lambda: None)
+ results = spinner.clean()
+ self.assertThat(results, Equals([port]))
+
+
+def test_suite():
+ from unittest import TestLoader, TestSuite
+ return TestSuite(
+ [TestLoader().loadTestsFromName(__name__),
+ make_integration_tests()])
=== modified file 'testtools/tests/test_testresult.py'
--- testtools/tests/test_testresult.py 2010-08-04 14:59:32 +0000
+++ testtools/tests/test_testresult.py 2010-10-10 16:49:39 +0000
@@ -396,7 +396,7 @@
File "...testtools...runtest.py", line ..., in _run_user...
return fn(*args)
File "...testtools...testcase.py", line ..., in _run_test_method
- testMethod()
+ return testMethod()
File "...testtools...tests...test_testresult.py", line ..., in error
1/0
ZeroDivisionError:... divi... by zero...
@@ -410,7 +410,7 @@
File "...testtools...runtest.py", line ..., in _run_user...
return fn(*args)
File "...testtools...testcase.py", line ..., in _run_test_method
- testMethod()
+ return testMethod()
File "...testtools...tests...test_testresult.py", line ..., in failed
self.fail("yo!")
AssertionError: yo!
Follow ups
