如何对线程产生进行单元测试？

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)在断言之前放入一个测试来进行测试,但这显然是hacky和lame - 模拟/单元测试异步内容的可接受方式是什么？

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)

UPDATE不需要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)

UPDATE

在Python 3.x中,补丁threading.Event代替threading._Event.