System Tests using Trial (Part 1)

You are tasked with writing or maintaining a collection of system tests for a large, Enterprise-scale, customer-facing, software-as-a-service product. The tests that already exist, however, take far too long for other project engineers to care to run them regularly. Your mission: improve system testing throughput by 50% or better. What do you do?

Most projects will have system tests written in the same (or similar) framework as their unit tests. Most SUnit frameworks do not make parallelism and/or concurrency a first-class consideration for programmers, even if they do support some manner of running multiple tests concurrently. Inevitably, engineers will resort to writing sequential tests, making simultaneous execution of tests more difficult (and if left long enough in this state, utterly impossible as the base of tests grows). Using a framework which emphasizes concurrency as a first-class concern will help prevent writing tests which cannot be run concurrently.

In this article, I'll use Twisted's Trial and dreid's treq module to illustrate how I avoid premature sequentialization in the test cases themselves. I'll assume, but will not describe, how to write re-usable components for testing; instead, I'll defer that subject for the next article.

Always Return Deferred Instances

Use Trial's version of unittest; avoid Python's unittest. This will let you express your tests using Twisted's Deferred instances. However, it's not as easy as changing a single import in your code.

For example, and if you'll permit me to elide a fair amount of boilerplate, take a look at the test_identity method below:

from .... import IdentityV2, TestResources

from twisted.trial import unittest

def find_end_point(rcs):
	... look through service catalog here ...

class TestIdentity(unittest.TestCase):
	def setUp(self):
		self.identity = IdentityV2(...)

    def test_identify(self):
        rcs = TestResources()
    	d = (
	        self.identity.authenticate_user(rcs)
            .addCallback(find_end_point)
    	)
	    return d

Notice how we make a call to self.identity.authenticate_user, which would return a Deferred instance. When that instance fires, we know that authentication either succeeded or failed; thus, we use addCallback above to tell Trial what to do with the results of that authentication attempt (in this case, attempt to find some relevant endpoint from the returned service catalog).

It's important to remember, though, that you're not actually authenticating at the time this point in the program is reached. Rather, you're creating a kind of to-do list for later execution. Trial will take it upon itself to execute this to-do list when it's convenient for it.

Keep your Reactor Clean

If we were to run test class as it's written above, and of course assuming we had suitable IdentityV2, et. al. definitions imported, we'd find that the test would run to completion, but would produce a pretty scary-looking error:

# trial test_wtf.TestWtf
test_wtf
  TestWtf
    test_scaling_up ...                                                 [ERROR]
                                                [ERROR]

===============================================================================
[ERROR]
Traceback (most recent call last):
Failure: twisted.trial.util.DirtyReactorAggregateError: Reactor was unclean.
DelayedCalls: (set twisted.internet.base.DelayedCall.debug = True to debug)
<DelayedCall 0x20ee5a8 [239.994215012s] called=0 cancelled=0 HTTPConnectionPool._removeConnection(('https', 'identity.api.rackspacecloud.com', 443), <twisted.web._newclient.HTTP11ClientProtocol instance at 0x20d2f38>)>

test_wtf.TestWtf.test_scaling_up
===============================================================================
[ERROR]
Traceback (most recent call last):
Failure: twisted.trial.util.DirtyReactorAggregateError: Reactor was unclean.
Selectables:
<<class 'twisted.internet.tcp.Client'> to ('identity.api.rackspacecloud.com', 443) at 20d6a50>

test_wtf.TestWtf.test_scaling_up
-------------------------------------------------------------------------------
Ran 1 tests in 0.738s

Egads!! But, have no fear — the solution turns out to be quite simple. We simply need to manually manage our HTTPConnectionPool at test setup and teardown, like so:

from twisted.internet import reactor
from twisted.web.client import HTTPConnectionPool
from twisted.internet.task import deferLater
from twisted.internet.tcp import Client

class TestIdentity(unittest.TestCase):
	def setUp(self):
		self.pool = HTTPConnectionPool(reactor, False)
		self.identity = IdentityV2(pool=self.pool, ...)   # Note!!

	def tearDown(self):
        def _check_fds(_):
	        fds = set(reactor.getReaders() + reactor.getReaders())
            if not [fd for fd in fds if isinstance(fd, Client)]:
    	        return
	        return deferLater(reactor, 0, _check_fds, None)
        return self.pool.closeCachedConnections().addBoth(_check_fds)

	# ... rest of class as it is ...

It's vitally important that you pass your explicitly created HTTPConnectionPool to all libraries you intend on using, so that they make requests using a consistent pool. If you don't do this, you'll find your requests do not terminate, and your tests will appear to deadlock. This is why we pass self.pool to the IdentityV2 constructor.

Always Rely on Reusable Component Libraries

Earlier, I stated that I'd show examples which rely only upon Twisted and treq modules. I kind of lied, as you probably already guessed; the reality is, I rely on Twisted and a set of reusable components. Those components, in turn, rely on Twisted and treq.

One reason why you want to structure your code this way is rapidity of writing tests. Having a catalog of test-related methods in useful classes allows you to express your tests in a kind of domain-specific language unique for your needs. For example, here's an excerpt of our Autoscaling API tests:

    d = (
        self.identity.authenticate_user(rcs)
        .addCallback(find_end_point)
        .addCallback(print_token_and_ep)
        .addCallback(self.scaling_group.start, self)
        .addCallback(self.scaling_policy_up_2.start, self)
        .addCallback(self.scaling_policy_up_2.execute)
        .addCallback(self.scaling_group.wait_for_N_servers,
                     2, timeout=1800)
        .addCallback(
            lambda _: self.scaling_group.get_scaling_group_state(rcs))
        .addCallback(dump_state)
        .addCallback(lambda _: rcs)
        .addCallback(self.scaling_policy_down_1.start, self)
        .addCallback(self.scaling_policy_down_1.execute)
        .addCallback(self.scaling_group.wait_for_N_servers,
                     1, timeout=900)
        .addCallback(
            lambda _: self.scaling_group.get_scaling_group_state(rcs)
        ).addCallback(dump_state)
    )
    return d

If you ignore the repeated calls to addCallback, you can read vertically down the list of callbacks added, and kind of make sense of what the test behavior intends to exercise, and what the expected results should be. If you're interested in looking at the real test file, click here.

In the next article, I'll show how you actually write these reusable components. Stay tuned!