FastAPI Testing, Background Tasks & WebSockets

FastAPI provides built-in tools for automated testing, in-process background tasks, and real-time bidirectional WebSockets. Built on Starlette, FastAPI uses httpx to test ASGI applications directly in memory without binding real network sockets. Understanding how to structure async unit tests, when to use BackgroundTasks vs full message queues (Celery), and how to manage WebSocket connection lifecycles is essential for senior backend engineers.

This chapter details ASGI in-memory testing (httpx.AsyncClient), BackgroundTasks execution, and WebSocket protocol state machines.

1. ASGI In-Memory Testing Architecture (httpx.AsyncClient)

FastAPI applications can be tested synchronously via Starlette’s TestClient or asynchronously via httpx.AsyncClient.

Instead of starting a live HTTP server on a TCP port, httpx.AsyncClient(app=app, base_url="http://test") uses ASGI In-Memory Transport:

ASGI In-Memory Testing Flow:

[ Pytest Test Runner ]
          |
          v
[ httpx.AsyncClient(app=app) ]
          |
          v (Bypasses OS Network Stack / Sockets!)
[ Direct ASGI Scope & Receive/Send Memory Channel ]
          |
          v
[ FastAPI Application Handler ]

Dependency Overriding in Tests:

FastAPI allows replacing production dependencies (such as database connections or third-party APIs) with mock or test fixtures using app.dependency_overrides:

import pytest
from httpx import AsyncClient
from myapp.main import app
from myapp.database import get_db

async def override_get_db():
    # Return mock or test database session
    yield test_db_session

app.dependency_overrides[get_db] = override_get_db

@pytest.mark.anyio
async def test_read_users():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        response = await ac.get("/users")
    assert response.status_code == 200

2. In-Process Background Tasks (BackgroundTasks)

FastAPI includes BackgroundTasks for executing lightweight operations after returning an HTTP response:

• Execution Model: BackgroundTasks runs tasks in the same process using Starlette’s post-response callback loop.
• async def vs def Tasks: If the background task is async def, it runs on the event loop after the response is sent. If it is def, it runs in Starlette’s threadpool.

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

def write_audit_log(message: str):
    with open("audit.log", "a") as f:
        f.write(message + "\n")

@app.post("/items/")
async def create_item(item_id: str, background_tasks: BackgroundTasks):
    # Enqueue task to run AFTER response is delivered to client
    background_tasks.add_task(write_audit_log, f"Item created: {item_id}")
    return {"message": "Item created"}

Warning: Do not use BackgroundTasks for heavy CPU-bound jobs, retried payment processing, or tasks requiring strict persistence guarantees. Process crashes will destroy pending BackgroundTasks. Use Celery or Arq for critical background work.

3. WebSocket Protocol State Machine (WebSocket)

FastAPI supports real-time bidirectional communication via Starlette’s WebSocket interface:

WebSocket Connection Lifecycle:

[ Client HTTP Upgrade Request (ws://domain/ws) ]
                        |
                        v
[ WebSocket.accept() ]  <-- Handshake completed! Status code 101 Switching Protocols
                        |
            ┌───────────┴───────────┐
            v                       v
[ websocket.receive_text() ]   [ websocket.send_text() ]
(Receives Frame from Client)   (Sends Frame to Client)
            │                       │
            └───────────┬───────────┘
                        v
[ WebSocketDisconnect ] <-- Client closes frame or connection drops

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: str):
    await websocket.accept() # Complete WS Handshake
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Client #{client_id} wrote: {data}")
    except WebSocketDisconnect:
        print(f"Client #{client_id} disconnected")

4. Production Trade-offs & Testing Strategies

• BackgroundTasks vs Celery: Use BackgroundTasks for quick, non-critical side effects (e.g. logging, updating user last_seen). Use Celery/RabbitMQ for tasks requiring retries, task tracking, or heavy CPU workloads.
• WebSocket Scaling: Scaling WebSockets across multiple server nodes requires a central Pub/Sub broker (Redis Pub/Sub or NATS) to broadcast messages across disconnected Uvicorn worker instances.