龙卷风测试异步应用程序的 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 的正确方式吗？

顺便说一句，tornado.testing.AsyncTestCase中的ExceptionStackContext其实是没用的：

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

super(AsyncTestCase, self).run(result) 不会抛出 AssertException。

Answer 1

如果您仔细按照示例进行操作，那么 ExceptionStackContext 会像我声称的那样工作。 It's much easier to follow if you read my article.

您用 unittest.TestCase 调用了 "self.stop"，但 TestCase 没有 self.stop，只有 AsyncTestCase 有。要使用测试用例，请尝试：

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()

请注意，我必须声明 "start"，而您没有在您的代码中声明，并设置 "places=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)

Answer 2

StackContexts 很神奇：Tornado 中有很多地方可以自动捕获当前的 StackContext 并稍后恢复。因此，当 delay() 调用 done() 时（假设 delay() 是根据 IOLoop.add_timeout 实现的，或者以其他方式正确处理 StackContext），ExceptionStackContext 已重新建立.

类似地，即使 AsyncTestCase.run() 中的 ExceptionStackContext 从未直接捕获任何异常，它建立了 "current" StackContext 以在测试中的任何地方捕获。（这就是为什么它是 with ExceptionStackContext 而不是普通的 try/except）

龙卷风测试异步应用程序的 ExceptionStackContext

ExceptionStackContext for tornado testing async app

tornado