Я фанат Tornado , одной из основных асинхронных веб-фреймворков для Python, но асинхронный код для юнит-тестирования — это большая проблема. Я собираюсь рассмотреть, в чем проблема, взглянуть на некоторые решения klutzy и предложить лучший способ. Если вам все равно, что я скажу, и вы просто хотите украсть мой код, загрузите его на GitHub .
Эта проблема
Допустим, вы работаете над какой-то очень сложной библиотекой, которая выполняет трудоемкие вычисления, и вы хотите проверить ее вывод:
# test_sync.py
import time
import unittest
def calculate():
# Do something profoundly complex
time.sleep(1)
return 42
class SyncTest(unittest.TestCase):
def test_find(self):
result = calculate()
self.assertEqual(42, result)
if __name__ == '__main__':
unittest.main()
Увидеть? Вы делаете операцию, затем вы проверяете, что получили ожидаемый результат. Нет пота.
Но как насчет тестирования асинхронных вычислений? У тебя будут проблемы. Давайте напишем асинхронный калькулятор и протестируем его:
# test_async.py
import time
import unittest
from tornado import ioloop
def async_calculate(callback):
"""
@param callback: A function taking params (result, error)
"""
# Do something profoundly complex requiring non-blocking I/O, which
# will complete in one second
ioloop.IOLoop.instance().add_timeout(
time.time() + 1,
lambda: callback(42, None)
)
class AsyncTest(unittest.TestCase):
def test_find(self):
def callback(result, error):
print 'Got result', result
self.assertEqual(42, result)
async_calculate(callback)
ioloop.IOLoop.instance().start()
if __name__ == '__main__':
unittest.main()
Да. Если вы запустите python test_async.py, вы увидите, что ожидаемый результат выводится на консоль:
Got result 42
… и тогда программа зависает навсегда. Проблема в том, что ioloop.IOLoop.instance (). Start () запускает бесконечный цикл. Вы должны остановить это явно, прежде чем вызов start () вернется.
Решение Клуци
Давайте остановим цикл в обратном вызове:
def callback(result, error):
ioloop.IOLoop.instance().stop()
print 'Got result', result
self.assertEqual(42, result)
Теперь, если вы запустили python test_async.py, то все выглядит так:
$ python test_async.py Got result 42 . ---------------------------------------------------------------------- Ran 1 test in 1.001s OK
Давайте посмотрим, действительно ли наш тест обнаружит ошибку. Измените функцию async_calculate (), чтобы получить число 17 вместо 42:
def async_calculate(callback):
"""
@param callback: A function taking params (result, error)
"""
# Do something profoundly complex requiring non-blocking I/O, which
# will complete in one second
ioloop.IOLoop.instance().add_timeout(
time.time() + 1,
<b>lambda: callback(17, None)</b>
)
И запустить тест:
$ python foo.py
Got result 17
ERROR:root:Exception in callback
Traceback (most recent call last):
File "/Users/emptysquare/.virtualenvs/blog/lib/python2.7/site-packages/tornado/ioloop.py", line 396, in _run_callback
callback()
File "foo.py", line 14, in
lambda: callback(17, None)
File "foo.py", line 22, in callback
self.assertEqual(42, result)
File "/System/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/unittest/case.py", line 494, in assertEqual
assertion_func(first, second, msg=msg)
File "/System/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/unittest/case.py", line 487, in _baseAssertEqual
raise self.failureException(msg)
AssertionError: 42 != 17
.
----------------------------------------------------------------------
Ran 1 test in 1.002s
OK
Ошибка AssertionError, но тест все еще проходит ! Увы, IOLoop Торнадо подавляет все исключения. Исключения выводятся на консоль, но инфраструктура unittest считает, что тест пройден.
Лучший путь
Мы собираемся выполнить небольшую операцию на Tornado, чтобы исправить это, создав и установив наш собственный IOLoop, который повторно вызывает все исключения в обратных вызовах. К счастью, Торнадо делает это легко. Добавьте import sys в начало test_async.py и вставьте следующее:
class PuritanicalIOLoop(ioloop.IOLoop):
"""
A loop that quits when it encounters an Exception.
"""
def handle_callback_exception(self, callback):
exc_type, exc_value, tb = sys.exc_info()
raise exc_value
Теперь добавьте метод setUp () в AsyncTest, который установит наш пуританский цикл:
def setUp(self):
super(AsyncTest, self).setUp()
# So any function that calls IOLoop.instance() gets the
# PuritanicalIOLoop instead of the default loop.
if not ioloop.IOLoop.initialized():
loop = PuritanicalIOLoop()
loop.install()
else:
loop = ioloop.IOLoop.instance()
self.assert_(
isinstance(loop, PuritanicalIOLoop),
"Couldn't install PuritanicalIOLoop"
)
Это немного сложнее для нашего простого случая — достаточно было бы вызова PuritanicalIOLoop (). Install () — но это все пригодится позже. В нашем простом наборе тестов setUp () запускается только один раз, поэтому проверка IOLoop.initialized () не нужна, но она понадобится вам, если вы запустите несколько тестов. Вызов super () будет необходим, если мы унаследуем от TestCase с помощью метода setUp (), что мы и сделаем ниже. А пока просто запустите python test_async.py и обратите внимание, что мы получаем правильный сбой:
$ python foo.py
Got result 17
F
======================================================================
FAIL: test_find (__main__.SyncTest)
----------------------------------------------------------------------
Traceback (most recent call last):
File "foo.py", line 49, in test_find
ioloop.IOLoop.instance().start()
File "/Users/emptysquare/.virtualenvs/blog/lib/python2.7/site-packages/tornado/ioloop.py", line 263, in start
self._run_callback(timeout.callback)
File "/Users/emptysquare/.virtualenvs/blog/lib/python2.7/site-packages/tornado/ioloop.py", line 398, in _run_callback
self.handle_callback_exception(callback)
File "foo.py", line 25, in handle_callback_exception
raise exc_value
AssertionError: 42 != 17
----------------------------------------------------------------------
Ran 1 test in 1.002s
FAILED (failures=1)
Прекрасный. Измените async_calculate () обратно на правильную версию, которая выдает 42.
Еще лучший путь
Итак, мы убедились, что наш тест выявляет ошибки в расчете. Но что, если у нас есть ошибка, которая не позволяет нашему обратному вызову когда-либо вызываться? Добавьте оператор return в верхней части async_calculate (), чтобы мы не выполняли обратный вызов:
def async_calculate(callback):
"""
@param callback: A function taking params (result, error)
"""
# Do something profoundly complex requiring non-blocking I/O, which
# will complete in one second
return
ioloop.IOLoop.instance().add_timeout(
time.time() + 1,
lambda: callback(42, None)
)
Теперь, если мы запустим тест, он зависнет навсегда, потому что IOLoop.stop () никогда не вызывается. Как мы можем написать тест, который утверждает, что обратный вызов в конечном итоге выполняется? Не бойся, я написал код:
class AssertEventuallyTest(unittest.TestCase):
def setUp(self):
super(AssertEventuallyTest, self).setUp()
# Callbacks registered with assertEventuallyEqual()
self.assert_callbacks = set()
def assertEventuallyEqual(
self, expected, fn, msg=None, timeout_sec=None
):
if timeout_sec is None:
timeout_sec = 5
timeout_sec = max(timeout_sec, int(os.environ.get('TIMEOUT_SEC', 0)))
start = time.time()
loop = ioloop.IOLoop.instance()
def callback():
try:
self.assertEqual(expected, fn(), msg)
# Passed
self.assert_callbacks.remove(callback)
if not self.assert_callbacks:
# All asserts have passed
loop.stop()
except AssertionError:
# Failed -- keep waiting?
if time.time() - start < timeout_sec:
# Try again in about 0.1 seconds
loop.add_timeout(time.time() + 0.1, callback)
else:
# Timeout expired without passing test
loop.stop()
raise
self.assert_callbacks.add(callback)
# Run this callback on the next I/O loop iteration
loop.add_callback(callback)
Этот класс позволяет нам регистрировать любое количество функций, которые вызываются периодически, пока они не будут равны ожидаемым значениям или не истечет время ожидания. Последняя функция, которая завершается успешно или останавливается, останавливает IOLoop, поэтому ваш тест определенно заканчивается. Время ожидания настраивается либо в качестве аргумента assertEventuallyEqual (), либо в качестве переменной среды TIMEOUT_SEC. Установка очень большого значения тайм-аута в вашей среде полезна для отладки некорректно работающего юнит-теста — установите его на миллион секунд, чтобы не прерывать время, пока вы выполняете код.
(Мой код вдохновлен тестом «в конечном итоге» мира Scala , который мне показал Брендан В. Макадамс .)
Вставьте AssertEventuallyTest в test_async.py и исправьте ваш тестовый пример, чтобы унаследовать его:
class AsyncTest(AssertEventuallyTest):
def setUp(self):
< ... snip ... >
def test_find(self):
results = []
def callback(result, error):
print 'Got result', result
results.append(result)
async_calculate(callback)
self.assertEventuallyEqual(
42,
lambda: results and results[0]
)
ioloop.IOLoop.instance().start()
Вызов IOLoop.stop () удален из обратного вызова, и мы добавили вызов assertEventuallyEqual () непосредственно перед запуском IOLoop.
Об этом коде следует отметить две детали:
Детализируйте первый: первый аргумент assertEventuallyEqual () является ожидаемым значением, а второй аргумент является функцией, которая должна в конечном итоге равняться ожидаемому значению. Отсюда и лямбда.
Подробно Второе: callback () нужно место для хранения своего результата, чтобы лямбда могла его найти, но здесь мы сталкиваемся с неприятной особенностью Python. Функции Python могут назначать переменные в своей области видимости или в глобальной области видимости (с ключевым словом global), но внутренние функции не могут назначаться значениям в области видимости внешних функций. Python 3 вводит нелокальное ключевое слово для решения этой проблемы, но в то же время мы можем обойти проблему, создав список результатов во внешней функции и добавив к нему внутреннюю функцию. Это распространенная идиома, которую вы будете часто использовать при написании обратных вызовов в асинхронных тестах модулей.
Вывод
Я собрал PuritanicalIOLoop и AssertEventuallyTest на GitHub ; иди возьми код. Ваши тестовые случаи могут выбрать наследование от PuritanicalTornadoTest, AssertEventuallyTest или обоих. Просто убедитесь, что ваши методы setUp вызывают super (MyTestCaseClass, self) .setUp (). Иди и проверь!
Источник: http://emptysquare.net/blog/tornado-unittesting-eventually-correct/