FastAPI Async Testing

Introduction

FastAPI is built on top of Starlette and leverages Python's asynchronous capabilities, making it an excellent choice for high-performance applications. However, testing asynchronous code comes with its own set of challenges. In this guide, we'll explore how to properly test asynchronous endpoints and functions in your FastAPI applications.

Testing asynchronous code requires a different approach compared to testing synchronous code. We need to ensure that:

Asynchronous functions are properly awaited
Test clients support asynchronous operations
Mocks and fixtures work correctly with async functions

Let's dive into how to effectively test your FastAPI async applications!

Prerequisites

Before we begin, make sure you have the following installed:

bash
pip install fastapi pytest pytest-asyncio httpx

pytest-asyncio: Provides support for testing asynchronous code with pytest
httpx: Modern HTTP client with support for both sync and async APIs

Setting Up Async Tests

To test asynchronous code, we need to configure pytest to handle async functions. This is where pytest-asyncio comes in handy.

Basic Setup

Create a test file (e.g., test_async_endpoints.py) and include the following:

python
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from httpx import AsyncClient
import asyncio

# Import your application
from your_app.main import app

# For synchronous tests
client = TestClient(app)

# Fixture for async testing
@pytest.fixture
async def async_client():
    async with AsyncClient(app=app, base_url="http://test") as client:
        yield client

# Mark test as async
@pytest.mark.asyncio
async def test_async_endpoint(async_client):
    # Your test code here
    pass

The key components here:

The @pytest.mark.asyncio decorator tells pytest to run this test as an asynchronous function
We create a fixture (async_client) that provides an AsyncClient instance
We use AsyncClient from httpx for testing async endpoints

Testing Async Endpoints

Let's create a simple FastAPI application with an async endpoint to test:

python
# main.py
from fastapi import FastAPI
import asyncio

app = FastAPI()

@app.get("/async-endpoint")
async def async_endpoint():
    # Simulate async operation
    await asyncio.sleep(0.1)
    return {"message": "This is an async response"}

Now, let's write a test for this endpoint:

python
# test_async_endpoints.py
import pytest
from httpx import AsyncClient
from main import app

@pytest.mark.asyncio
async def test_async_endpoint():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        response = await ac.get("/async-endpoint")
        assert response.status_code == 200
        assert response.json() == {"message": "This is an async response"}

Using Fixtures for Cleaner Tests

For cleaner and more reusable tests, let's use fixtures:

python
# conftest.py
import pytest
from httpx import AsyncClient
from main import app

@pytest.fixture
async def async_client():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        yield ac

# test_async_endpoints.py
@pytest.mark.asyncio
async def test_async_endpoint(async_client):
    response = await async_client.get("/async-endpoint")
    assert response.status_code == 200
    assert response.json() == {"message": "This is an async response"}

Mocking in Async Tests

Testing async code often requires mocking external dependencies. Let's see how to use mocks in async tests:

Example with Database Dependency

First, let's create a FastAPI app with a dependency on an async database:

python
# db.py
class AsyncDB:
    async def get_user(self, user_id: int):
        # In real app, this would query a database
        await asyncio.sleep(0.1)  # Simulating DB query
        return {"id": user_id, "name": f"User {user_id}"}

# main.py
from fastapi import FastAPI, Depends
from db import AsyncDB

app = FastAPI()

async def get_db():
    db = AsyncDB()
    return db

@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncDB = Depends(get_db)):
    user = await db.get_user(user_id)
    return user

Now, let's test this endpoint by mocking the database dependency:

python
# test_async_db.py
import pytest
from unittest.mock import AsyncMock, patch
from httpx import AsyncClient
from main import app

@pytest.mark.asyncio
async def test_get_user():
    # Create async mock
    mock_db = AsyncMock()
    mock_db.get_user.return_value = {"id": 1, "name": "Test User"}
    
    # Patch the dependency
    with patch("main.get_db", return_value=mock_db):
        async with AsyncClient(app=app, base_url="http://test") as ac:
            response = await ac.get("/users/1")
            
            assert response.status_code == 200
            assert response.json() == {"id": 1, "name": "Test User"}
            mock_db.get_user.assert_called_once_with(1)

Testing Async Background Tasks

FastAPI supports background tasks, which can also be tested asynchronously. Let's create an endpoint with a background task:

python
# background_tasks.py
from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

async def process_notification(message: str):
    # This would normally send an email, push notification, etc.
    # For demo, we'll just store in a global variable
    global notifications
    notifications.append(message)

notifications = []

@app.post("/send-notification/")
async def send_notification(message: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(process_notification, message)
    return {"status": "Notification sent"}

Here's how we can test this endpoint:

python
# test_background_tasks.py
import pytest
from httpx import AsyncClient
from background_tasks import app, notifications

@pytest.fixture(autouse=True)
def reset_notifications():
    notifications.clear()
    yield

@pytest.mark.asyncio
async def test_send_notification():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        response = await ac.post("/send-notification/", params={"message": "Hello"})
        
        assert response.status_code == 200
        assert response.json() == {"status": "Notification sent"}
        
        # We need to wait a bit for the background task to complete
        import asyncio
        await asyncio.sleep(0.1)
        
        assert "Hello" in notifications

Real-World Example: Testing a Complete Async API

Let's put everything together and create a more complete example with a FastAPI application that uses async database operations and async external API calls:

python
# dependencies.py
from typing import List, Dict, Any
import httpx
import asyncio

class AsyncDatabase:
    def __init__(self):
        # Simulate a database with some data
        self.users = {
            1: {"id": 1, "name": "John", "email": "[email protected]"},
            2: {"id": 2, "name": "Jane", "email": "[email protected]"}
        }
    
    async def get_user(self, user_id: int) -> Dict[str, Any]:
        await asyncio.sleep(0.1)  # Simulate DB latency
        return self.users.get(user_id)
    
    async def get_all_users(self) -> List[Dict[str, Any]]:
        await asyncio.sleep(0.1)
        return list(self.users.values())

class ExternalService:
    async def get_weather(self, city: str) -> Dict[str, Any]:
        # Normally this would call an external API
        async with httpx.AsyncClient() as client:
            # In a real app, this would be a real API call
            await asyncio.sleep(0.2)  # Simulate API call
            return {"city": city, "temperature": 72, "condition": "Sunny"}

# app.py
from fastapi import FastAPI, Depends, HTTPException
from dependencies import AsyncDatabase, ExternalService

app = FastAPI()

# Dependencies
async def get_db():
    return AsyncDatabase()

async def get_external_service():
    return ExternalService()

@app.get("/users/{user_id}")
async def get_user(
    user_id: int, 
    db: AsyncDatabase = Depends(get_db)
):
    user = await db.get_user(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

@app.get("/users/{user_id}/weather")
async def get_user_weather(
    user_id: int,
    city: str,
    db: AsyncDatabase = Depends(get_db),
    external_service: ExternalService = Depends(get_external_service)
):
    user = await db.get_user(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    
    weather = await external_service.get_weather(city)
    
    return {
        "user": user["name"],
        "city": city,
        "temperature": weather["temperature"],
        "condition": weather["condition"]
    }

Now let's write tests for these endpoints:

python
# test_app.py
import pytest
from unittest.mock import AsyncMock, patch
from httpx import AsyncClient
from app import app

@pytest.fixture
async def async_client():
    async with AsyncClient(app=app, base_url="http://test") as client:
        yield client

# Test get_user endpoint
@pytest.mark.asyncio
async def test_get_user(async_client):
    # Create mock database
    mock_db = AsyncMock()
    mock_db.get_user.return_value = {"id": 1, "name": "John", "email": "[email protected]"}
    
    # Patch dependency
    with patch("app.get_db", return_value=mock_db):
        response = await async_client.get("/users/1")
        
        assert response.status_code == 200
        assert response.json() == {"id": 1, "name": "John", "email": "[email protected]"}
        mock_db.get_user.assert_called_once_with(1)

# Test user not found
@pytest.mark.asyncio
async def test_get_user_not_found(async_client):
    mock_db = AsyncMock()
    mock_db.get_user.return_value = None
    
    with patch("app.get_db", return_value=mock_db):
        response = await async_client.get("/users/999")
        
        assert response.status_code == 404
        assert response.json() == {"detail": "User not found"}

# Test get_user_weather endpoint
@pytest.mark.asyncio
async def test_get_user_weather(async_client):
    # Mock both dependencies
    mock_db = AsyncMock()
    mock_db.get_user.return_value = {"id": 1, "name": "John", "email": "[email protected]"}
    
    mock_service = AsyncMock()
    mock_service.get_weather.return_value = {
        "city": "New York", 
        "temperature": 75, 
        "condition": "Cloudy"
    }
    
    # Patch both dependencies
    with patch("app.get_db", return_value=mock_db), \
         patch("app.get_external_service", return_value=mock_service):
        
        response = await async_client.get("/users/1/weather?city=New%20York")
        
        assert response.status_code == 200
        assert response.json() == {
            "user": "John",
            "city": "New York",
            "temperature": 75,
            "condition": "Cloudy"
        }
        
        mock_db.get_user.assert_called_once_with(1)
        mock_service.get_weather.assert_called_once_with("New York")

Best Practices for FastAPI Async Testing

Use pytest-asyncio: It's the most efficient way to test async code with pytest.
Always await async functions: Forgetting to await an async function can lead to unexpected behavior and hard-to-debug issues.
Use AsyncMock for mocking async functions: Regular Mock objects don't handle coroutines properly.
Isolate tests: Each test should be independent of others, especially when testing async code.
Test error conditions: Async code can fail in different ways, so test error handling carefully.
Use dependency injection for easier mocking: FastAPI's dependency injection system makes it easier to replace components in tests.
Be careful with event loops: Some testing environments might have issues with event loops. If you encounter errors, try using asyncio.set_event_loop_policy().
Handle timeouts: Add timeouts to prevent tests from hanging indefinitely if an async operation doesn't complete.

Common Testing Patterns

Testing Database Operations

When testing database operations, you might want to use an in-memory database or a test database for integration tests:

python
# test_database.py
import pytest
import asyncio
from databases import Database
from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String

# Setup test database
metadata = MetaData()
users = Table(
    "users",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String),
    Column("email", String)
)

@pytest.fixture
async def database():
    # Use SQLite in-memory database for testing
    database_url = "sqlite:///test.db"
    database = Database(database_url)
    engine = create_engine(database_url)
    metadata.create_all(engine)
    
    await database.connect()
    
    # Insert test data
    query = users.insert().values(id=1, name="Test User", email="[email protected]")
    await database.execute(query)
    
    yield database
    
    await database.disconnect()
    metadata.drop_all(engine)

@pytest.mark.asyncio
async def test_get_user_from_db(database):
    query = users.select().where(users.c.id == 1)
    result = await database.fetch_one(query)
    
    assert result is not None
    assert result["name"] == "Test User"
    assert result["email"] == "[email protected]"

Testing WebSocket Endpoints

FastAPI supports WebSockets, which also need async testing:

python
# websocket_app.py
from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Message received: {data}")

# test_websocket.py
import pytest
from fastapi.testclient import TestClient
from websocket_app import app

def test_websocket():
    client = TestClient(app)
    with client.websocket_connect("/ws") as websocket:
        websocket.send_text("Hello")
        data = websocket.receive_text()
        assert data == "Message received: Hello"

Summary

Testing asynchronous code in FastAPI requires specific techniques and tools, but it's essential for ensuring your async APIs work correctly. In this guide, we've covered:

Setting up async testing with pytest-asyncio
Creating and using async test fixtures
Testing async endpoints with AsyncClient
Mocking async dependencies with AsyncMock
Testing background tasks
A complete example of testing a FastAPI application with async components
Best practices for FastAPI async testing

With these techniques, you should be able to write robust tests for your FastAPI applications, ensuring that your asynchronous code works as expected.

Additional Resources

Exercises

Create a FastAPI application with an async endpoint that fetches data from a mock database and test it using pytest-asyncio.
Write tests for a FastAPI endpoint that makes multiple concurrent async requests to external services.
Implement and test a FastAPI application with WebSocket support for real-time communication.
Create a FastAPI endpoint that uses background tasks and write tests to verify the background tasks are executed correctly.
Build a complete REST API with async database operations and write a comprehensive test suite for it.

Happy testing!

If you spot any mistakes on this website, please let me know at [email protected]. I’d greatly appreciate your feedback! :)

Introduction​

Prerequisites​

Setting Up Async Tests​

Basic Setup​

Testing Async Endpoints​

Using Fixtures for Cleaner Tests​

Mocking in Async Tests​

Example with Database Dependency​

Testing Async Background Tasks​

Real-World Example: Testing a Complete Async API​

Best Practices for FastAPI Async Testing​

Common Testing Patterns​

Testing Database Operations​

Testing WebSocket Endpoints​

Summary​

Additional Resources​

Exercises​