Twisted support¶
testtools provides support for testing Twisted code. Install the
testtools[twisted] extra to use this.
Matching Deferreds¶
testtools provides support for making assertions about synchronous
Deferreds.
A “synchronous” Deferred is one that does
not need the reactor or any other asynchronous process in order to fire.
Normal application code can’t know when a
Deferred is going to fire, because that is
generally left up to the reactor. Well-written unit tests provide fake
reactors, or don’t use the reactor at all, so that
Deferreds fire synchronously.
These matchers allow you to make assertions about when and how
Deferreds fire, and about what values
they fire with.
See also Testing Deferreds without the reactor and the Deferred howto.
- testtools.twistedsupport.succeeded(matcher)
Match a Deferred that has fired successfully.
For example:
fires_with_the_answer = succeeded(Equals(42)) deferred = defer.succeed(42) assert_that(deferred, fires_with_the_answer)
This assertion will pass. However, if
deferredhad fired with a different value, or had failed, or had not fired at all, then it would fail.Use this instead of
twisted.trial.unittest.SynchronousTestCase.successResultOf().- Parameters:
matcher – A matcher to match against the result of a
Deferred.- Returns:
A matcher that can be applied to a synchronous
Deferred.
- testtools.twistedsupport.failed(matcher)
Match a Deferred that has failed.
For example:
error = RuntimeError('foo') fails_at_runtime = failed( AfterPreprocessing(lambda f: f.value, Equals(error))) deferred = defer.fail(error) assert_that(deferred, fails_at_runtime)
This assertion will pass. However, if
deferredhad fired successfully, had failed with a different error, or had not fired at all, then it would fail.Use this instead of
twisted.trial.unittest.SynchronousTestCase.failureResultOf().- Parameters:
matcher – A matcher to match against the result of a failing
Deferred.- Returns:
A matcher that can be applied to a synchronous
Deferred.
- testtools.twistedsupport.has_no_result()
Match a Deferred that has not yet fired.
For example, this will pass:
assert_that(defer.Deferred(), has_no_result())
But this will fail:
>>> assert_that(defer.succeed(None), has_no_result()) Traceback (most recent call last): ... File "testtools/assertions.py", line 22, in assert_that raise MismatchError(matchee, matcher, mismatch, verbose) testtools.matchers._impl.MismatchError: No result expected on <Deferred at ... current result: None>, found None instead
As will this:
>>> assert_that(defer.fail(RuntimeError('foo')), has_no_result()) Traceback (most recent call last): ... File "testtools/assertions.py", line 22, in assert_that raise MismatchError(matchee, matcher, mismatch, verbose) testtools.matchers._impl.MismatchError: No result expected on <Deferred at ... current result: <twisted.python.failure.Failure <type 'exceptions.RuntimeError'>>>, found <twisted.python.failure.Failure <type 'exceptions.RuntimeError'>> instead
Extracting results from Deferreds¶
When testing code that returns synchronous
Deferreds for compatibility reasons, you
may want to extract the result directly instead of using matchers.
- testtools.twistedsupport.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.
As a rule, this function should not be used when operating with asynchronous Deferreds (i.e. for normal use of Deferreds in application code). In those cases, it is better to add callbacks and errbacks as needed.
- exception testtools.twistedsupport.DeferredNotFired(deferred)
Raised when we extract a result from a Deferred that’s not fired yet.
Running tests in the reactor¶
testtools provides support for running asynchronous Twisted tests: tests that
return a Deferred and run the reactor
until it fires and its callback chain is completed.
Here’s how to use it:
from testtools import TestCase
from testtools.twistedsupport import AsynchronousDeferredRunTest
class MyTwistedTests(TestCase):
run_tests_with = AsynchronousDeferredRunTest
def test_foo(self):
# ...
return d
Note that you do not have to use a special base TestCase in order to run
Twisted tests, you should just use the regular testtools.TestCase
base class.
You can also run individual tests within a test case class using the Twisted test runner:
class MyTestsSomeOfWhichAreTwisted(TestCase):
def test_normal(self):
pass
@run_test_with(AsynchronousDeferredRunTest)
def test_twisted(self):
# ...
return d
See AsynchronousDeferredRunTest and
AsynchronousDeferredRunTestForBrokenTwisted
for more information.
Controlling the Twisted logs¶
Users of Twisted Trial will be accustomed to all tests logging to
_trial_temp/test.log. By default,
AsynchronousDeferredRunTest will not
do this, but will instead:
suppress all messages logged during the test run
attach them as the
twisted-logdetail (see Details) which is shown if the test fails
The first behavior is controlled by the suppress_twisted_logging parameter
to AsynchronousDeferredRunTest, which is
set to True by default. The second is controlled by the
store_twisted_logs parameter, which is also True by default.
If store_twisted_logs is set to False, you can still get the logs
attached as a detail by using the
CaptureTwistedLogs fixture. Using the
CaptureTwistedLogs fixture is equivalent
to setting store_twisted_logs to True.
For example:
class DoNotCaptureLogsTests(TestCase):
run_tests_with = partial(AsynchronousDeferredRunTest,
store_twisted_logs=False)
def test_foo(self):
log.msg('logs from this test are not attached')
def test_bar(self):
self.useFixture(CaptureTwistedLogs())
log.msg('logs from this test *are* attached')
Converting Trial tests to testtools tests¶
Use the
AsynchronousDeferredRunTestrunnerMake sure to upcall to
TestCase.setUp()andTestCase.tearDown()Don’t use
setUpClassortearDownClassDon’t expect setting
.todo,.timeoutor.skipattributes to do anythingReplace
twisted.trial.unittest.SynchronousTestCase.flushLoggedErrors()withflush_logged_errors()Replace
twisted.trial.unittest.TestCase.assertFailure()withassert_fails_with()Trial spins the reactor a couple of times before cleaning it up,
AsynchronousDeferredRunTestdoes not. If you rely on this behavior, useAsynchronousDeferredRunTestForBrokenTwisted.