Как написать тест торнадо Python, чтобы утверждать, что ioloop выключился?

Question

У меня есть некоторый код, который будет перехватывать SIGTERM / SIGINT и инструктировать ioloop торнадо (который является оберткой вокруг цикла asyncio) ждать завершения запросов на борту перед тем, как отключить 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, который мы пытаемся отключить, тот же, на котором будет выполняться сам тест.

Следующий код - это то, что у меня сейчас ...

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.
        #
        # ???