如何单元测试线程是否被创建？

Question

我在一个单元测试中遇到了竞争条件，正在尝试修复它。

假设有一个模块叫做 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()

这里是它的测试代码：

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)

当然，这个测试会失败，错误信息是 AssertionError: Expected to be called once. Called 0 times.，因为调用 bar() 是非阻塞的，所以断言（检查）发生得太早了。

为了让测试通过，可以在断言之前加上 time.sleep(1)，但是这显然是个不太好的解决办法——那么，处理异步代码的测试和模拟的正确方法是什么呢？

Answer 1

怎么样把 bar 修改成返回一个 thead 对象呢：

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

然后，在测试代码中加入这个线程：

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)

更新 另一种方法，不需要改动 bar。

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)

更新

在 Python 3.x 中，使用 threading.Event 来替代 threading._Event。