使用AsyncHTTPTestCase测试基于asyncio的Tornado应用遇AssertionError问题

I’ve run into this exact issue before when testing Tornado apps tied to asyncio’s event loop. The problem boils down to persistent global IOLoop state between test runs: when you call AsyncIOMainLoop().install(), you bind asyncio’s main loop as Tornado’s global IOLoop, but the default AsyncHTTPTestCase doesn’t automatically reset this state after each test. The first test works because the loop is fresh, but subsequent tests hit the assertion because the old loop is still registered.

Here’s how to fix it properly, with step-by-step solutions:

1. Understand the Root Cause #

Tornado’s AsyncHTTPTestCase normally creates a new IOLoop for each test, but manually installing AsyncIOMainLoop overrides this global state. Once installed, the loop sticks around between tests—so when the next test tries to initialize a new loop, it hits the assertion because a global loop already exists.

2. Solution 1: Custom Test Base Class (Recommended) #

Create a reusable base class that manages the asyncio loop lifecycle for each test, ensuring full isolation between test runs:

import asyncio
from tornado.testing import AsyncHTTPTestCase
from tornado.platform.asyncio import AsyncIOMainLoop
from tornado.ioloop import IOLoop

class AsyncioIsolatedTestCase(AsyncHTTPTestCase):
    def setUp(self):
        # Reset any existing global IOLoop state
        AsyncIOMainLoop.clear_current()
        # Create a fresh asyncio loop for this test
        self.test_loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self.test_loop)
        # Install it as Tornado's IOLoop
        AsyncIOMainLoop(self.test_loop).install()
        # Proceed with standard test setup
        super().setUp()

    def tearDown(self):
        # Clean up the test loop first
        super().tearDown()
        self.test_loop.close()
        # Wipe global Tornado and asyncio loop state
        AsyncIOMainLoop.clear_current()
        asyncio.set_event_loop(None)
        # Ensure no leftover IOLoop references
        if IOLoop.initialized():
            IOLoop.current().close()

    def get_new_ioloop(self):
        # Tell the test case to use our custom asyncio loop
        return self.test_loop

Now inherit this base class in your tests instead of AsyncHTTPTestCase:

from app import make_app

class TestMyApp(AsyncioIsolatedTestCase):
    def get_app(self):
        # Just return your app—no loop installation here!
        return make_app()

    def test_first_request(self):
        resp = self.fetch("/")
        self.assertEqual(resp.code, 200)

    def test_second_request(self):
        resp = self.fetch("/")
        self.assertEqual(resp.code, 200)

3. Solution 2: Quick Fix (Per-Test Cleanup) #

If you don’t want to create a base class, add cleanup logic directly to your test’s tearDown method:

from tornado.testing import AsyncHTTPTestCase
from tornado.platform.asyncio import AsyncIOMainLoop
from tornado.ioloop import IOLoop
import asyncio

class TestMyApp(AsyncHTTPTestCase):
    def setUp(self):
        AsyncIOMainLoop.clear_current()
        self.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self.loop)
        AsyncIOMainLoop(self.loop).install()
        super().setUp()

    def tearDown(self):
        super().tearDown()
        self.loop.close()
        AsyncIOMainLoop.clear_current()
        asyncio.set_event_loop(None)
        if IOLoop.initialized():
            IOLoop.current().close()

    def get_new_ioloop(self):
        return self.loop

    def get_app(self):
        return make_app()

    # ... your test methods ...

4. Bonus: Separate App Initialization from Loop Setup #

To avoid future conflicts, move your loop installation code out of your app creation logic and into your startup script (e.g., start.py):

app.py (no loop code here):

from tornado.web import Application, RequestHandler

class MainHandler(RequestHandler):
    async def get(self):
        self.write("Hello from Tornado + AsyncIO!")

def make_app():
    return Application([(r"/", MainHandler)])

start.py (handles loop setup):

from tornado.platform.asyncio import AsyncIOMainLoop
from app import make_app
from tornado.ioloop import IOLoop

if __name__ == "__main__":
    AsyncIOMainLoop().install()
    app = make_app()
    app.listen(8888)
    IOLoop.current().start()

This way, your tests can fully control the loop lifecycle without conflicting with production startup code.

The key takeaway is ensuring full isolation of event loop state between tests—each test gets a fresh loop, and all global state is cleaned up afterward.

内容的提问来源于stack exchange，提问作者Alexander Kamyanskiy