Run asynchronous code from test casesΒΆ

The Reactor fixture can be used to drive asynchronous Twisted code from a regular synchronous Python TestCase.

The approach differs from trial or testtools twisted support: instead of starting the reactor in the main thread and letting it spin for a while waiting for the Deferred returned by the test to fire, this fixture will keep the reactor running in a background thread until cleanup.

When used with testresources‘s FixtureResource and OptimisingTestSuite, this fixture makes it possible to have full control and monitoring over long-running processes that should be up for the whole test suite run, and maybe produce output useful for the test itself.

The typical use case is integration testing.

>>> from testtools import TestCase

>>> from twisted.internet import reactor
>>> from twisted.internet.threads import blockingCallFromThread
>>> from twisted.internet.utils import getProcessOutput

>>> from txfixtures import Reactor

>>> class TestUsingAsyncAPIs(TestCase):
...
...     def setUp(self):
...         super().setUp()
...         self.useFixture(Reactor())
...
...     def test_uptime(self):
...         out = blockingCallFromThread(reactor, getProcessOutput, b"uptime")
...         self.assertIn("load average", out.decode("utf-8"))
...
>>> test = TestUsingAsyncAPIs(methodName="test_uptime")
>>> test.run().wasSuccessful()
True