Testing FastAPI applications with async resources? If you're using Starlette's TestClient, you're mixing sync and async contexts in ways that might break your tests. Here's why httpx.AsyncClient should be your go-to choice for keeping everything properly async.
The Sync/Async Boundary Problem
Here's the problematic code from TestClient's implementation:
1 2 3 4 5 6 7 | class TestClient: def __enter__(self) -> "TestClient": # Creates a sync-to-async bridge - this is where things break! self.portal = anyio.from_thread.start_blocking_portal() self.lifespan = LifespanHandler(self.app) self.portal.start_task_soon(self.lifespan.startup) return self |
The culprit is anyio.from_thread.start_blocking_portal() — this creates a sync-to-async bridge with a background thread. Your async startup events, database connections, and HTTP sessions all get bound to this background thread's event loop, while your async test marked with @pytest.mark.asyncio run in main thread's event loop.
The fundamental issue: You can't safely access async resources created in one event loop from a different event loop, that just won't work.
A Real-World Example That Breaks: Sync Test, Async Resources
Let's see this sync/async mixing problem in action:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 | import aiohttp from fastapi import FastAPI import pytest # Global session that gets initialized during startup app_session = None events = {} async def lifespan(app: FastAPI): """Initialize async resources during startup.""" global app_session # This runs in TestClient's background thread app_session = aiohttp.ClientSession() events["startup"] = "executed" yield if app_session: await app_session.close() events["shutdown"] = "executed" app = FastAPI(lifespan=lifespan) @app.get("/health") async def health_check(): """Simple health check that verifies the session is available.""" if app_session and not app_session.closed: return {"status": "healthy", "session_ready": True} return {"status": "unhealthy", "session_ready": False} @pytest.fixture(scope='session') def test_client(): """Create a test client using Starlette's TestClient.""" with TestClient(app) as client: yield client @pytest.fixture(scope='session') async def async_client(): """Create a properly configured async client.""" async with LifespanManager(app) as manager: async with AsyncClient( base_url="http://test", transport=ASGITransport(app) ) as client: yield client @pytest.mark.asyncio # This test is marked as async, but it's trying to access an async resource initialized in the background thread async def test_with_testclient(test_client): """This test demonstrates the sync/async mixing problem.""" assert app_session is not None assert events["startup"] == "executed" with pytest.raises(RuntimeError, match="Timeout context manager should be used inside a task"): async with app_session.get("http://google.com") as response: pass |
Why it fails: Even though we marked the test as async, the TestClient initializes the aiohttp.ClientSession in the background async context, but we are trying to access it from event loop of the main thread.
The Solution: Stay Fully Async with httpx.AsyncClient
Instead of mixing sync and async contexts, let's keep everything consistently async:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 | import pytest_asyncio from httpx import AsyncClient, ASGITransport from contextlib import asynccontextmanager @pytest_asyncio.fixture(scope='session') async def async_client(): """Create a properly configured async client.""" async with LifespanManager(app) as manager: async with AsyncClient( base_url="http://test", transport=ASGITransport(app) ) as client: yield client @pytest.mark.asyncio # Test is truly async async def test_session_available_in_same_context(async_client): # Client is async """Test that works because we're staying consistently async.""" assert events["startup"] == "executed" assert not app_session.closed async with app_session.get("http://google.com") as response: await response.text() |
Why This Approach Works Better: No Sync/Async Mixing
- 🔄 Consistent Async Context
- Your startup events, async resources, and test code all run in the same async context — no dangerous sync/async boundaries.
- 🛡️ Proper Resource Management
- Async resources created during startup remain in the same async context and are immediately usable in your tests.
- 🔧 Clean Lifecycle
- LifespanManager handles startup and shutdown events in a pure async context, ensuring your app behaves exactly like it would in production.
Best Practices: Keep Sync and Async Separate
✅ Do This: Stay Consistently Async
- Use httpx.AsyncClient with ASGITransport for all FastAPI testing
- Leverage LifespanManager for proper app lifecycle management
- Keep async resources in the same async context as your tests
- Use pytest-asyncio and mark your tests with @pytest.mark.asyncio or set asyncio_mode="auto" in your pytest.ini or pyproject.toml file
- Never mix sync and async contexts - if your app is async, your tests should be too
❌ Avoid This: Dangerous Sync/Async Mixing
- Using TestClient when your app has async resources (forces sync/async mixing) and you are trying to access async resource created in the background thread.
- Ignoring event loop boundaries in your tests
Complete Working Example
Here's a full example showing how to properly test async resources:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 | import asyncio import threading from typing import Dict import aiohttp import pytest import pytest_asyncio from asgi_lifespan import LifespanManager from fastapi import FastAPI from httpx import ASGITransport, AsyncClient from starlette.testclient import TestClient # Global state to track startup events events: Dict[str, str] = {} app_session: aiohttp.ClientSession | None = None def log_thread_info(context: str = "") -> None: """Log information about the current thread and event loop.""" thread_id = threading.get_ident() thread_name = threading.current_thread().name try: loop = asyncio.get_running_loop() loop_thread = getattr(loop, '_thread_id', 'unknown') except RuntimeError: loop_thread = 'no loop' print(f"{context}:") print(f" Current Thread: {thread_id} ({thread_name})") print(f" Event Loop: {loop_thread}") async def lifespan(app: FastAPI): """Lifespan event that initializes an async client session.""" global app_session # Initialize an aiohttp session during startup app_session = aiohttp.ClientSession() events["startup"] = "executed" log_thread_info("startup") yield if app_session: await app_session.close() events["shutdown"] = "executed" # Create FastAPI app with startup events app = FastAPI(lifespan=lifespan) # Test fixtures @pytest_asyncio.fixture(scope='session') async def async_client(): """Create an async client using httpx.AsyncClient.""" async with LifespanManager(app) as manager: async with AsyncClient( base_url="http://test", transport=ASGITransport(app) ) as client: yield client @pytest.fixture(scope='session') def test_client(): """Create a test client using Starlette's TestClient.""" with TestClient(app) as client: yield client # Tests @pytest.mark.asyncio async def test_async_client_with_session(async_client): """Test using httpx.AsyncClient with aiohttp session.""" # This works because we're in the same event loop context log_thread_info("test_async_client_with_session") assert app_session is not None assert not app_session.closed assert events["startup"] == "executed" async with app_session.get("http://google.com") as response: await response.text() @pytest.mark.asyncio # This test is marked as async, but it's trying to access an async resource initialized in the background thread async def test_with_testclient(test_client): """This test demonstrates the sync/async mixing problem.""" assert app_session is not None assert events["startup"] == "executed" with pytest.raises(RuntimeError, match="Timeout context manager should be used inside a task"): async with app_session.get("http://google.com") as response: pass |
Quick Migration Guide: From Sync/Async Mixing to Pure Async
If you're currently using TestClient and mixing sync/async contexts, here's how to migrate to a pure async approach:
Before (Problematic Sync/Async Mixing):
1 2 3 4 | def test_endpoint(test_client): # Sync test function # TestClient forces sync interface over async app response = test_client.get("/api/endpoint") # Sync call to async endpoint assert response.status_code == 200 |
After (Pure Async Context):
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 | @pytest_asyncio.fixture(scope='session') async def async_client(): """Create a properly configured async client.""" async with LifespanManager(app): #to run startup and shutdown events async with AsyncClient( base_url="http://test", transport=ASGITransport(app) ) as client: yield client @pytest.mark.asyncio async def test_endpoint(async_client): # Truly async test function # AsyncClient maintains async context throughout response = await async_client.get("/api/endpoint") # Async call to async endpoint assert response.status_code == 200 |
The Bottom Line: Respect the Async/Sync Boundary
When testing FastAPI applications with async components, sync/async context mixing isn't just inconvenient — it's fundamentally broken. TestClient's approach of wrapping async apps in sync interfaces creates subtle bugs and violates async programming principles.
By using httpx.AsyncClient with LifespanManager, you create a testing environment that respects the async nature of your application.