Asynchronous spiesΒΆ

New in version 1.5.1.

Sometimes interaction among your SUT class and their collaborators does not meet a synchronous behavior. That may happen when the SUT performs collaborator invocations in a different thread, or when the invocation pass across a message queue, publish/subscribe service, etc.

Something like that:

class Collaborator(object):
    def write(self, data):
        print "your code here"

class SUT(object):
    def __init__(self, collaborator):
        self.collaborator = collaborator

    def some_method(self):
        thread.start_new_thread(self.collaborator.write, ("something",))

If you try to test your collaborator is called using a Spy, you will get a wrong behavior:

# THE WRONG WAY
class AsyncTests(unittest.TestCase):
    def test_wrong_try_to_test_an_async_invocation(self):
        # given
        spy = Spy(Collaborator)
        sut = SUT(spy)

        # when
        sut.some_method()

        # then
        assert_that(spy.write, called())

due to the called() assertion may happen before the write() invocation, although not always...

You may be tempted to put a sleep before the assertion, but this is a bad solution. A right way to solve that issue is to use something like a barrier. The threading.Event may be used as a barrier. See this new test version:

# THE DIRTY WAY
class AsyncTests(unittest.TestCase):
    def test_an_async_invocation_with_barrier(self):
        # given
        barrier = threading.Event()
        with Spy(Collaborator) as spy:
            spy.write.attach(lambda *args: barrier.set)

        sut = SUT(spy)

        # when
        sut.some_method()
        barrier.wait(1)

        # then
        assert_that(spy.write, called())

The spy.write.attach() is part of the doublex stub-observer mechanism, a way to run arbitrary code when stubbed methods are called.

That works because the called() assertion is performed only when the spy releases the barrier. If the write() invocation never happens, the barrier.wait() continues after 1 second but the test fails, as must do. When all is right, the barrier waits just the required time.

Well, this mechanism is a doublex builtin (the async matcher) since release 1.5.1 providing the same behavior in a clearer way. The next is functionally equivalent to the listing just above:

# THE DOUBLEX WAY
class AsyncTests(unittest.TestCase):
    def test_test_an_async_invocation_with_doublex_async(self):
        # given
        spy = Spy(Collaborator)
        sut = SUT(spy)

        # when
        sut.some_method()

        # then
        assert_that(spy.write, called().async(timeout=1))