ExceptionStackContext для тестирования Торнадо асинхронного приложения

Question

наблюдал ajdavis @ pycon2015 о тестировании асинхронных приложениях, как показано ниже: http://pyvideo.org/video/3419/eventually-correct-testing-async-apps

У меня есть вопрос об использовании ExceptionStackContext. В презентации, он используется, как показано ниже:

import unittest 
from tornado.stack_context import ExceptionStackContext 
from tornado.ioloop import IOLoop 

class TestAsync(unittest.TestCase): 
    def test_delay(self): 
     io_loop = IOLoop.current() 
     def handle_exception(typ, val, tb): 
      self.failure = (typ, val, tb) 
      io_loop.stop() 
     def done(): 
      self.assertAlmostEqual(time.time() - start, 2) 
      self.stop() 
     with ExceptionStackContext(handle_exception): 
      delay(3, done) # fail the assert 
     io_loop.start() 

запустить этот тест, io_loop не остановится. потому что handle_exception не вызывается.

мой метод асинхронной delay является:

import threading 
import time 

def delay(delay_seconds, callback): 
    def wrap():  
     time.sleep(delay_seconds) 
     callback() 
    t = threading.Thread(target=wrap) 
    t.daemon = True 
    t.start() 

так я думаю, он должен ExceptionStackContext должен обернуть done(), как показано ниже:

def callback(): 
    with ExceptionStackContext(handle_exception): 
     done() 
delay(2, callback) 
io_loop.start() 

это путь райт использовать ExceptionStackContext?

, кстати, ExceptionStackContext в tornado.testing.AsyncTestCase бесполезно на самом деле:

def run(self, result=None): 
    with ExceptionStackContext(self._handle_exception): 
     super(AsyncTestCase, self).run(result) 

super(AsyncTestCase, self).run(result) не бросает вверх AssertException.

Answer 1

StackContexts является магическим: Есть много мест в Торнадо, который автоматически перехватывать текущий StackContext и возобновить ее позже. Поэтому, когда delay() звонит done() (предполагается, что delay() реализован в терминах IOLoop.add_timeout или иным образом обрабатывает StackContext правильно), восстановлен ExceptionStackContext.

Аналогичным образом, несмотря на то, что ExceptionStackContext в AsyncTestCase.run() никогда не попадает прямо в какие-либо исключения, он устанавливает «текущий» StackContext для захвата в любом месте теста. (Вот почему это with ExceptionStackContext вместо обычных try/except)

Answer 2

Если вы внимательно следите за примерами, то ExceptionStackContext работает, как я утверждаю. It's much easier to follow if you read my article.

Вы назвали self.stop с unittest.TestCase, но TestCase не имеет self.stop, только AsyncTestCase. Чтобы использовать TestCase, попробуйте:

import unittest 
import time 
from tornado.stack_context import ExceptionStackContext 
from tornado.ioloop import IOLoop 


def delay(seconds, callback): 
    io_loop = IOLoop.current() 
    io_loop.add_timeout(io_loop.time() + seconds, callback) 


class TestAsync(unittest.TestCase): 
    def test_delay(self): 
     io_loop = IOLoop.current() 
     start = time.time() 

     def handle_exception(typ, val, tb): 
      self.failure = (typ, val, tb) 
      io_loop.stop() 

     def done(): 
      self.assertAlmostEqual(time.time() - start, 2, places=1) 
      io_loop.stop() 

     with ExceptionStackContext(handle_exception): 
      delay(3, done) # fail the assert 
     io_loop.start() 

Примечание я должен был объявить «старт», который вы не в вашем коде, и установить «места = 1». Это работает правильно. Или с AsyncTestCase:

import time 

from tornado import testing 
from tornado.ioloop import IOLoop 


def delay(seconds, callback): 
    io_loop = IOLoop.current() 
    io_loop.add_timeout(io_loop.time() + seconds, callback) 


class TestAsync(testing.AsyncTestCase): 
    def test_delay(self): 
     start = time.time() 
     delay(3, self.stop) 
     self.wait() 
     self.assertAlmostEqual(time.time() - start, 2, places=1)