Предположим, у меня есть следующая функция, которая работает асинхронно:
import threading
def api1(cb, *args, **kwargs):
def _target():
ret = do_some_work_that_takes_a_lot_of_time(*args, **kwargs)
cb(ret)
thrd1 = threading.Thread(target=_target)
thrd1.start()
return thrd1
Ответ в этом SO вопросе предполагает, что объект Mock
можно использовать для проверки возможности обратного вызовабудет вызван в определенном поведении. Однако, когда я попытался использовать mock в моем тестовом примере, как показано ниже, все просто не работает:
def test_api1(mocker):
mock = mocker.Mock()
t = api1(mock, "foo", bar="bar")
mock.assert_called_with("hoge")
t.join()
И вывод команды pytest выглядит следующим образом:
==================================================================================================== FAILURES ====================================================================================================
____________________________________________________________________________________________________ test_api ____________________________________________________________________________________________________
mocker = <pytest_mock.MockFixture object at 0x7fc0ccc47f10>
def test_api1(mocker):
mock = mocker.Mock()
t = api1(mock, "foo", bar="bar")
> mock.assert_called_with("hoge")
E AssertionError: Expected call: mock("hoge")
E Not called
test_api.py:17: AssertionError
=============================================================================================== 1 failed in 0.03s ================================================================================================
Итак, как правильно написать тест, который утверждает, что обратный вызов должен вызываться определенным образом? Обратите внимание, что сколько времени потребуется, чтобы вызвать обратный вызов, и трудоемкая работа может даже не быть вызвана в главном потоке.