Как разрешить, чтобы поток был порожден?

У меня состояние гонки в unittest, которое я пытаюсь исправить.

Предположим, есть модуль 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. AssertionError: Expected to be called once. Called 0 times. потому что вызов bar() не блокируется, поэтому утверждение происходит слишком рано.

Тест может быть проделан, чтобы пройти, поставив time.sleep(1) перед утверждением, но это, очевидно, взломанно и хромотно - каков приемлемый способ издеваться над асинхронным материалом?

Ответ 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)

ОБНОВИТЬ

В Python 3.x, patch threading.Event вместо threading._Event.