Testing Tornado Applications: From Unit Tests to Integration
Tornado is a powerful Python web framework and asynchronous networking library known for its non-blocking I/O and ability to handle thousands of simultaneous connections. However, building scalable applications is only half the battle — ensuring they work correctly under various conditions is equally important. This tutorial walks you through testing Tornado applications, starting from isolated unit tests and progressing to full integration tests that exercise your entire stack.
Why Testing Tornado Applications Matters
Asynchronous code introduces complexities that synchronous frameworks don't face. Race conditions, callback ordering, and event loop management can all lead to subtle bugs that only manifest under specific conditions. A robust testing strategy helps you:
- Catch regressions before they reach production
- Document expected behavior through executable specifications
- Refactor with confidence
- Validate asynchronous flows that are hard to verify manually
- Ensure your application handles edge cases like timeouts and connection drops
Understanding Tornado's Testing Utilities
Tornado ships with a dedicated testing module called tornado.testing that provides utilities specifically designed for asynchronous test cases. The two most important components are:
AsyncTestCase— A subclass ofunittest.TestCasethat manages an IOLoop for each testAsyncHTTPTestCase— ExtendsAsyncTestCaseand starts a test HTTP server, providing a client to make requestsgen_test— A decorator that runs a coroutine test on the IOLoop, similar toasyncio.run
These utilities handle the boilerplate of setting up and tearing down the event loop, so you can focus on writing meaningful assertions.
Setting Up Your Test Environment
Before writing tests, make sure you have the necessary dependencies installed. Create a requirements-test.txt file:
tornado>=6.2
pytest>=7.0
pytest-tornasync>=0.6.0.post2
coverage>=6.0
Install them with:
pip install -r requirements-test.txt
While Tornado's built-in testing utilities work with the standard unittest framework, many developers prefer pytest for its simpler syntax and powerful fixtures. This tutorial covers both approaches.
Building a Sample Tornado Application
To make the examples concrete, let's build a small application that we can test. Create a file called app.py:
import json
import tornado.ioloop
import tornado.web
from tornado.httpclient import AsyncHTTPClient
class MainHandler(tornado.web.RequestHandler):
def get(self):
self.write({"message": "Welcome to the Tornado app"})
class UserHandler(tornado.web.RequestHandler):
def initialize(self, db):
self.db = db
def get(self, user_id):
user = self.db.get(int(user_id))
if user is None:
self.set_status(404)
self.write({"error": "User not found"})
return
self.write(user)
def post(self):
data = json.loads(self.request.body)
if "name" not in data:
self.set_status(400)
self.write({"error": "Name is required"})
return
user_id = max(self.db.keys()) + 1 if self.db else 1
self.db[user_id] = {"id": user_id, "name": data["name"]}
self.set_status(201)
self.write(self.db[user_id])
class WeatherHandler(tornado.web.RequestHandler):
async def get(self, city):
client = AsyncHTTPClient()
try:
response = await client.fetch(
f"https://api.weather.example.com/{city}"
)
self.write(response.body)
except Exception as e:
self.set_status(502)
self.write({"error": f"Weather service unavailable: {str(e)}"})
def make_app(db=None):
if db is None:
db = {}
return tornado.web.Application([
(r"/", MainHandler),
(r"/users/(\d+)", UserHandler, {"db": db}),
(r"/users", UserHandler, {"db": db}),
(r"/weather/([a-zA-Z]+)", WeatherHandler),
])
if __name__ == "__main__":
app = make_app()
app.listen(8888)
tornado.ioloop.IOLoop.current().start()
This application has three handlers: a simple welcome endpoint, a user management endpoint backed by an in-memory dictionary, and an asynchronous weather endpoint that calls an external API. Each of these presents different testing challenges.
Writing Unit Tests
Unit tests focus on testing individual components in isolation. For Tornado applications, this means testing handlers and business logic without spinning up the full HTTP server or making real network calls.
Testing Handler Logic Directly
You can test handler methods by instantiating them with mock request objects. However, a cleaner approach is to test the business logic separately from the handler. Let's refactor the UserHandler to extract its logic:
# user_service.py
class UserService:
def __init__(self, db):
self.db = db
def get_user(self, user_id):
return self.db.get(user_id)
def create_user(self, name):
if not name or not isinstance(name, str):
raise ValueError("Name is required and must be a string")
user_id = max(self.db.keys()) + 1 if self.db else 1
user = {"id": user_id, "name": name}
self.db[user_id] = user
return user
Now write a unit test for this service:
# test_user_service.py
import unittest
from user_service import UserService
class TestUserService(unittest.TestCase):
def setUp(self):
self.db = {}
self.service = UserService(self.db)
def test_create_user_assigns_incrementing_id(self):
user1 = self.service.create_user("Alice")
user2 = self.service.create_user("Bob")
self.assertEqual(user1["id"], 1)
self.assertEqual(user2["id"], 2)
self.assertEqual(user1["name"], "Alice")
self.assertEqual(user2["name"], "Bob")
def test_create_user_stores_in_db(self):
user = self.service.create_user("Charlie")
self.assertEqual(self.db[1], user)
def test_create_user_rejects_empty_name(self):
with self.assertRaises(ValueError):
self.service.create_user("")
def test_create_user_rejects_non_string_name(self):
with self.assertRaises(ValueError):
self.service.create_user(123)
def test_get_user_returns_none_for_missing(self):
self.assertIsNone(self.service.get_user(999))
def test_get_user_returns_user_when_found(self):
self.service.create_user("Dana")
user = self.service.get_user(1)
self.assertEqual(user["name"], "Dana")
if __name__ == "__main__":
unittest.main()
These tests run fast because they don't involve any I/O or network operations. They verify the core business logic in complete isolation.
Testing Asynchronous Functions
For testing async functions without a full HTTP server, use Tornado's AsyncTestCase and gen_test decorator. Let's test a utility function that fetches data asynchronously:
# async_utils.py
from tornado.httpclient import AsyncHTTPClient
from tornado import gen
async def fetch_json(url):
client = AsyncHTTPClient()
response = await client.fetch(url)
import json
return json.loads(response.body)
# test_async_utils.py
from tornado.testing import AsyncTestCase, gen_test
from tornado.httpclient import HTTPResponse
from unittest.mock import patch, AsyncMock
from io import BytesIO
import json
from async_utils import fetch_json
class TestFetchJson(AsyncTestCase):
@gen_test
async def test_fetch_json_parses_response(self):
mock_body = json.dumps({"temp": 72, "city": "Seattle"}).encode()
mock_response = HTTPResponse(
request=None,
code=200,
buffer=BytesIO(mock_body),
)
with patch("async_utils.AsyncHTTPClient") as mock_client_class:
mock_client = mock_client_class.return_value
mock_client.fetch = AsyncMock(return_value=mock_response)
result = await fetch_json("https://example.com/weather")
self.assertEqual(result["temp"], 72)
self.assertEqual(result["city"], "Seattle")
@gen_test
async def test_fetch_json_raises_on_invalid_json(self):
mock_body = b"not json"
mock_response = HTTPResponse(
request=None,
code=200,
buffer=BytesIO(mock_body),
)
with patch("async_utils.AsyncHTTPClient") as mock_client_class:
mock_client = mock_client_class.return_value
mock_client.fetch = AsyncMock(return_value=mock_response)
with self.assertRaises(json.JSONDecodeError):
await fetch_json("https://example.com/bad")
Here we mock the AsyncHTTPClient to avoid making real network calls. The gen_test decorator ensures the coroutine runs on the test's IOLoop and properly waits for completion.
Writing Integration Tests
Integration tests exercise multiple components working together. In Tornado, this typically means starting the actual HTTP server and making real requests against it. The AsyncHTTPTestCase class makes this straightforward.
Basic Integration Test with AsyncHTTPTestCase
# test_app_integration.py
import json
from tornado.testing import AsyncHTTPTestCase
from app import make_app
class TestAppIntegration(AsyncHTTPTestCase):
def setUp(self):
self.db = {}
super().setUp()
def get_app(self):
return make_app(db=self.db)
def test_main_handler_returns_welcome(self):
response = self.fetch("/")
self.assertEqual(response.code, 200)
body = json.loads(response.body)
self.assertEqual(body["message"], "Welcome to the Tornado app")
def test_get_nonexistent_user_returns_404(self):
response = self.fetch("/users/999")
self.assertEqual(response.code, 404)
body = json.loads(response.body)
self.assertEqual(body["error"], "User not found")
def test_create_and_retrieve_user(self):
# Create a user
response = self.fetch(
"/users",
method="POST",
body=json.dumps({"name": "Eve"}),
)
self.assertEqual(response.code, 201)
created = json.loads(response.body)
self.assertEqual(created["name"], "Eve")
self.assertEqual(created["id"], 1)
# Retrieve the user
response = self.fetch("/users/1")
self.assertEqual(response.code, 200)
retrieved = json.loads(response.body)
self.assertEqual(retrieved["name"], "Eve")
def test_create_user_without_name_returns_400(self):
response = self.fetch(
"/users",
method="POST",
body=json.dumps({}),
)
self.assertEqual(response.code, 400)
body = json.loads(response.body)
self.assertEqual(body["error"], "Name is required")
The get_app method returns the application instance to test. The self.fetch method is a convenience wrapper around AsyncHTTPClient that makes synchronous-looking requests against the test server. Despite the synchronous appearance, everything runs on the IOLoop under the hood.
Testing Asynchronous Handlers with External APIs
The WeatherHandler calls an external API, which we need to mock in integration tests. We can patch the AsyncHTTPClient at the handler level:
# test_weather_integration.py
import json
from unittest.mock import patch, AsyncMock
from tornado.testing import AsyncHTTPTestCase
from tornado.httpclient import HTTPResponse
from io import BytesIO
from app import make_app
class TestWeatherHandler(AsyncHTTPTestCase):
def get_app(self):
return make_app()
@patch("app.AsyncHTTPClient")
def test_weather_returns_data_on_success(self, mock_client_class):
mock_body = json.dumps({
"city": "Seattle",
"temperature": 58,
"condition": "Rainy"
}).encode()
mock_response = HTTPResponse(
request=None,
code=200,
buffer=BytesIO(mock_body),
)
mock_client = mock_client_class.return_value
mock_client.fetch = AsyncMock(return_value=mock_response)
response = self.fetch("/weather/Seattle")
self.assertEqual(response.code, 200)
body = json.loads(response.body)
self.assertEqual(body["city"], "Seattle")
self.assertEqual(body["temperature"], 58)
@patch("app.AsyncHTTPClient")
def test_weather_returns_502_on_failure(self, mock_client_class):
mock_client = mock_client_class.return_value
mock_client.fetch = AsyncMock(
side_effect=ConnectionError("DNS resolution failed")
)
response = self.fetch("/weather/Nowhere")
self.assertEqual(response.code, 502)
body = json.loads(response.body)
self.assertIn("Weather service unavailable", body["error"])
By patching AsyncHTTPClient at the module level where it's imported, we intercept all calls made by the handler. This keeps tests fast and deterministic while still exercising the full request-response cycle through Tornado's routing and handler logic.
Using pytest with Tornado
Many teams prefer pytest over unittest for its concise syntax and powerful fixture system. The pytest-tornasync plugin provides fixtures for testing Tornado applications with pytest.
Configuring pytest for Tornado
Create a conftest.py file with shared fixtures:
# conftest.py
import pytest
from tornado.httpclient import AsyncHTTPClient
from app import make_app
@pytest.fixture
def app():
db = {}
return make_app(db=db)
@pytest.fixture
def db(app):
# Access the db passed to the application
for handler in app.default_router.rules:
if hasattr(handler.target, 'db'):
return handler.target.db
return {}
@pytest.fixture
async def http_client(http_server, http_client):
return http_client
Writing pytest-style Tests
# test_app_pytest.py
import json
import pytest
@pytest.mark.gen_test
async def test_main_handler(http_client, base_url):
response = await http_client.fetch(f"{base_url}/")
assert response.code == 200
body = json.loads(response.body)
assert body["message"] == "Welcome to the Tornado app"
@pytest.mark.gen_test
async def test_create_user_flow(http_client, base_url, db):
# Create
response = await http_client.fetch(
f"{base_url}/users",
method="POST",
body=json.dumps({"name": "Frank"}),
)
assert response.code == 201
created = json.loads(response.body)
assert created["name"] == "Frank"
# Verify it was stored
assert 1 in db
assert db[1]["name"] == "Frank"
# Retrieve
response = await http_client.fetch(f"{base_url}/users/1")
assert response.code == 200
retrieved = json.loads(response.body)
assert retrieved["name"] == "Frank"
@pytest.mark.gen_test
async def test_missing_user_returns_404(http_client, base_url):
response = await http_client.fetch(
f"{base_url}/users/999",
raise_error=False
)
assert response.code == 404
body = json.loads(response.body)
assert body["error"] == "User not found"
Note the raise_error=False parameter — by default, AsyncHTTPClient raises an exception for non-200 responses. In tests, you often want to inspect the response instead, so disabling this behavior is common.
Testing Error Handling and Edge Cases
Robust applications handle errors gracefully. Your tests should verify that error paths work as expected.
Testing Invalid Input
# test_edge_cases.py
import json
from tornado.testing import AsyncHTTPTestCase
from app import make_app
class TestEdgeCases(AsyncHTTPTestCase):
def get_app(self):
return make_app(db={})
def test_post_invalid_json_returns_error(self):
response = self.fetch(
"/users",
method="POST",
body="not valid json",
raise_error=False,
)
self.assertEqual(response.code, 500)
def test_post_with_extra_fields_ignores_them(self):
response = self.fetch(
"/users",
method="POST",
body=json.dumps({"name": "Grace", "extra": "field"}),
)
self.assertEqual(response.code, 201)
body = json.loads(response.body)
self.assertEqual(body["name"], "Grace")
self.assertNotIn("extra", body)
def test_get_user_with_non_numeric_id_returns_404(self):
# The regex r"/users/(\d+)" won't match non-numeric IDs
response = self.fetch("/users/abc", raise_error=False)
self.assertEqual(response.code, 404)
def test_weather_with_numbers_in_city_returns_404(self):
response = self.fetch("/weather/123", raise_error=False)
self.assertEqual(response.code, 404)
Testing Concurrent Requests
One of Tornado's strengths is handling concurrent requests. You can test this behavior using async tests:
# test_concurrency.py
import json
from tornado.testing import AsyncHTTPTestCase, gen_test
from tornado import gen
from app import make_app
class TestConcurrency(AsyncHTTPTestCase):
def get_app(self):
return make_app(db={})
@gen_test
async def test_multiple_concurrent_creates_get_unique_ids(self):
# Fire off multiple POST requests concurrently
futures = [
self.fetch("/users", method="POST", body=json.dumps({"name": f"User{i}"}))
for i in range(5)
]
responses = await gen.multi(futures)
ids = []
for response in responses:
self.assertEqual(response.code, 201)
body = json.loads(response.body)
ids.append(body["id"])
# All IDs should be unique
self.assertEqual(len(ids), len(set(ids)))
self.assertEqual(sorted(ids), [1, 2, 3, 4, 5])
This test reveals an important point: our simple in-memory database is not thread-safe or concurrency-safe. In a real application with concurrent requests, you'd need proper locking or an async-safe data store. This is exactly the kind of issue that integration tests can surface.
Measuring Test Coverage
Coverage measurement helps you identify untested code paths. Use the coverage package to track which lines your tests exercise:
coverage run -m pytest test_app_pytest.py
coverage report -m
coverage html
This generates an HTML report showing line-by-line coverage. Aim for high coverage on business logic, but don't obsess over 100% — some code (like error handling for impossible states) may not be worth testing.
Best Practices for Testing Tornado Applications
Structure Your Tests in Layers
- Unit tests for pure business logic — fast, isolated, numerous
- Handler tests for request parsing and response formatting
- Integration tests for end-to-end flows through the HTTP stack
- Smoke tests for deployment verification
Mock External Dependencies
Never make real network calls in tests. Mock HTTP clients, database connections, and any external services. This keeps tests fast, deterministic, and independent of network conditions.
Use Meaningful Test Names
Test names should describe the scenario and expected outcome. test_create_user_assigns_incrementing_id is far more informative than test_create_user_1.
Test One Thing Per Test
Each test should verify a single behavior. If a test fails, you should immediately know what broke. Avoid testing multiple unrelated assertions in a single test method.
Leverage Fixtures for Setup and Teardown
Use setUp and tearDown methods (or pytest fixtures) to ensure each test starts with a clean state. Shared mutable state between tests leads to flaky, order-dependent failures.
Test Both Happy and Unhappy Paths
It's easy to test that things work when everything goes right. It's more valuable to test that your application fails gracefully when inputs are invalid, services are unavailable, or resources are exhausted.
Keep Tests Fast
Slow tests discourage developers from running them frequently. Mock external dependencies, use in-memory databases, and avoid unnecessary delays. A full test suite should complete in seconds, not minutes.
Use raise_error=False for Expected Errors
When testing endpoints that return error status codes, always pass raise_error=False to fetch. Otherwise, Tornado will raise an HTTPError before you can inspect the response.
Running Tests in CI/CD
Integrate your tests into a CI pipeline to catch regressions automatically. Here's a sample GitHub Actions configuration:
# .github/workflows/tests.yml
name: Tests
on: [push, pull_request]
jobs:
test:
runs-on: ubuntu-latest
strategy:
matrix:
python-version: ["3.9", "3.10", "3.11", "3.12"]
steps:
- uses: actions/checkout@v4
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v4
with:
python-version: ${{ matrix.python-version }}
- name: Install dependencies
run: |
pip install -r requirements.txt
pip install -r requirements-test.txt
- name: Run tests with coverage
run: |
coverage run -m pytest -v
coverage report
- name: Upload coverage
uses: codecov/codecov-action@v3
Conclusion
Testing Tornado applications requires understanding both standard testing practices and the unique challenges of asynchronous code. By starting with focused unit tests for your business logic, adding integration tests that exercise the full HTTP stack, and mocking external dependencies to keep tests fast and reliable, you can build a comprehensive test suite that gives you confidence in your application's correctness. Tornado's built-in testing utilities like AsyncTestCase and AsyncHTTPTestCase handle the event loop plumbing, letting you focus on writing meaningful assertions. Whether you choose unittest or pytest, the key is to test in layers, cover both happy and error paths, and run your tests frequently in CI. With these practices in place, you can refactor aggressively, add features confidently, and ship your Tornado applications knowing they behave as expected under real-world conditions.