How to unittest that a thread is spawned?

Question

I have a race condition in a unittest I'm trying to fix.

Suppose there's a module spam.py:

import threading

def foo(*args, **kwargs):
    pass

def bar():
    t = threading.Timer(0.5, foo, args=('potato',), kwargs={'x': 69, 'y':'spam'})
    t.start()

And here is the test for it:

from mock import patch
from spam import bar
from unittest import TestCase

class SpamTest(TestCase):
    def test_bar(self):
        with patch('spam.foo') as mock:
            bar()
            mock.assert_called_once_with('potato', y='spam', x=69)

Of course this test fails with AssertionError: Expected to be called once. Called 0 times. because the call to bar() is non-blocking, so the assertion happens too early.

The test can be made to pass by putting a time.sleep(1) in before the assert, but this is obviously hacky and lame - what is the accepted way to mock / unittest asynchronous stuff?

Answer 1

How about modifying the bar to return a thead object:

def bar():
    t = threading.Timer(0.5, foo, args=('potato',), kwargs={'x': 69, 'y':'spam'})
    t.start()
    return t # <----

Then, join the thread in the test code:

class SpamTest(TestCase):
    def test_bar(self):
        with patch('spam.foo') as mock:
            t = bar()
            t.join() # <----
            mock.assert_called_once_with('potato', y='spam', x=69)

UPDATE alternative that does not require bar change.

import threading
import time

...

class SpamTest(TestCase):
    def test_bar(self):
        foo = threading.Event()
        with patch('spam.foo', side_effect=lambda *args, **kwargs: foo.set()) as mock:
            # Make the callback `foo` to be called immediately
            with patch.object(threading._Event, 'wait', time.sleep(0.000001)):
                bar()
            foo.wait() # Wait until `spam.foo` is called. (instead of time.sleep)
            mock.assert_called_once_with('potato', y='spam', x=69)

UPDATE

In Python 3.x, patch threading.Event instead of threading._Event.