如何编写 python 龙卷风测试来断言 ioloop 已关闭？

Question

我有一些代码会拦截 SIGTERM/SIGINT 并指示 tornado ioloop（它是 asyncio 循环的包装器）在关闭 tornado ioloop 之前等待飞行中的请求完成（见下文）。

"""signals module provides helper functions for tornado graceful shutdown."""

import asyncio
import functools
import signal
import time

from tornado.httpserver import HTTPServer
from tornado.ioloop import IOLoop

SHUTDOWN_TIMEOUT = 30


def sig_handler(server: HTTPServer, timeout: int, sig, frame):
    """Schedules ioloop shutdown after specified timeout when TERM/INT signals received.

    In-flights tasks running on the asyncio event loop will be given the
    opportunity to finish before the loop is shutdown after specified timeout.

    Expects to be initiated using partial application:
        functools.partial(sig_handler, HTTPServer())
    This partial application is typically handled by signals.sig_listener.
    """

    io_loop = IOLoop.current()

    def stop_loop(deadline):
        now = time.time()
        tasks = asyncio.all_tasks()

        if now < deadline and len(tasks) > 0:
            # defer shutdown until all tasks have a chance to complete
            io_loop.add_timeout(now + 1, stop_loop, deadline)
        else:
            io_loop.stop()

    async def shutdown():
        # stop listening for new connections
        server.stop()

        # schedule ioloop shutdown
        stop_loop(time.time() + timeout)

    # execute callback on next event loop tick
    io_loop.add_callback_from_signal(shutdown)


def sig_listener(server: HTTPServer, timeout: int = 0):
    """Configures listeners for TERM/INT signals.

    Timeout should be a positive integer, otherwise a default will be provided.
    """

    if not timeout:
        timeout = SHUTDOWN_TIMEOUT

    p = functools.partial(sig_handler, server, timeout)
    signal.signal(signal.SIGTERM, p)
    signal.signal(signal.SIGINT, p)

这段代码（见上文）工作正常，我已经手动测试了它是否可以满足我们的需要，但我不知道如何在使用 AsyncHTTPTestCase 的自动化测试中实现相同的目标，因为这会使异步请求同步 +我们试图关闭的 ioloop 与运行测试本身的 ioloop 相同。

以下代码是我目前所拥有的...

import asyncio
import os
import signal
import threading
import time

import bf_metrics

import bf_tornado.handlers
import bf_tornado.signals

import tornado.gen
import tornado.ioloop
import tornado.testing
import tornado.web


class TestGracefulShutdown(tornado.testing.AsyncHTTPTestCase):
    def get_app(self):
        class FooHandler(bf_tornado.handlers.BaseHandler):
            metrics = bf_metrics.Metrics(namespace='foo', host='localhost')

            def get(self):
                asyncio.sleep(5)
                self.finish('OK')

        return tornado.web.Application([
            (r'/', FooHandler)
        ])

    def test_graceful_shutdown(self):
        # override AsyncHTTPTestCase default timeout of 5s
        os.environ["ASYNC_TEST_TIMEOUT"] = "10"

        shutdown_timeout = 8
        bf_tornado.signals.sig_listener(self.http_server, shutdown_timeout)

        pid = os.getpid()

        def trigger_signal():
            # defer SIGNINT long enough to allow HTTP request to tornado server
            time.sleep(2)
            os.kill(pid, signal.SIGINT)

        thread = threading.Thread(target=trigger_signal)
        thread.daemon = True
        thread.start()

        resp = self.fetch('/')
        assert resp.code == 200

        # ???
        #
        # WHAT ASSERTION DO WE USE HERE?
        #
        # WE CAN'T CHECK tornado.ioloop.IOLoop.current()
        # BECAUSE IT'LL STILL BE RUNNING AS PART OF THIS TEST.
        #
        # MAYBE WE COULD TEST asyncio.all_tasks() ? BUT WE WANT TO BE ABLE TO
        # SAY CONCLUSIVELY THAT THE IOLOOP WAS SHUTDOWN.
        #
        # ???

Answer 1

我不认为这是你可以用 AsyncHTTPTestCase 测试的东西，因为事件循环的开始和停止对你是隐藏的（如果你不使用@gen_test，IOLoop 大部分时间已经停止. 如果你确实使用@gen_test，有一个 IOLoop 一直在运行，但如果你停止它，你的测试就会挂起）。为了测试 IOLoop 是否停止，您的代码需要调用IOLoop.start()。

测试此类启动和关闭问题的最全面方法是将服务器作为子进程运行。通过这种方式，您可以向它发送 HTTP 请求和信号，并查看进程是否退出。

在当前流程中也有一些方法可以做到这一点，在现实主义和性能之间进行各种权衡。例如，像这样：

def test_graceful_shutdown(self):
    def trigger_signal():
        time.sleep(2)
        os.kill(os.getpid(), signal.SIGINT)
    thread = threading.Thread(target=trigger_signal)
    thread.start()

    io_loop = IOLoop()
    install_signal_handler(io_loop)
    def failure():
        self.fail("did not shut down within 5 seconds")
        io_loop.stop()
    io_loop.call_after(5, failure)
    io_loop.start()