How to Test HTTPX Async Client Pool Reuse with Mocks in Python
Mock an httpx.AsyncClient to verify connection pool reuse by asserting GET calls share a single client instance across concurrent async requests.
pip install httpx
Python code
26 linesimport asyncio
import httpx
from unittest.mock import AsyncMock, patch, Mock
async def fetch_with_pool(client, url, n_reuses=3):
results = []
for i in range(n_reuses):
resp = await client.get(url)
results.append(resp.status_code)
await asyncio.sleep(0) # yield to loop to mimic real usage
return results
async def main():
mock_response = Mock()
mock_response.status_code = 200
mock_client = AsyncMock(spec=httpx.AsyncClient)
mock_client.get.return_value = mock_response
results = await fetch_with_pool(mock_client, "https://api.example.com/data", n_reuses=5)
print(f"Status codes: {results}")
print(f"GET call count: {mock_client.get.call_count}")
print(f"Reused same client: {mock_client is mock_client}")
if __name__ == "__main__":
asyncio.run(main())
Output
Status codes: [200, 200, 200, 200, 200]
GET call count: 5
Reused same client: True
How it works
AsyncMock(spec=httpx.AsyncClient) creates a mock that mimics the async client's interface, so await client.get() returns the configured mock response. The mock_client.get.return_value sets a Mock with status_code=200, and each loop iteration awaits it, simulating real sequential requests. mock_client.get.call_count increments per call, proving the client is reused — no new instance is created. The await asyncio.sleep(0) yields control to the event loop, matching production async behavior where I/O boundaries occur. This pattern validates that your code reuses connections instead of opening a new one per request, which is critical for performance.
Common mistakes
- Using a plain Mock instead of AsyncMock, which breaks awaited calls with a TypeError.
- Forgetting to set the return value on the mock, causing None status_code errors.
- Mocking the client inside the request loop, defeating the purpose of testing reuse.
Variations
- Use `pytest.mark.asyncio` with async test functions instead of manual asyncio.run().
- Patch `httpx.AsyncClient` globally with `patch` to verify it's instantiated once.
Real-world use cases
- Verifying that a shared HTTP client in a FastAPI service reuses its connection pool across concurrent endpoints.
- Testing a background worker that makes many sequential API calls to confirm it doesn't exhaust connections.
- Ensuring an SDK's async methods reuse a single client instance for rate-limited third-party APIs.
Sponsored
More from Concurrency & performance
- Benchmark list.append vs deque.append in Python medium
- Build a Python Performance Profiler That Generates Readable Reports medium
- Graceful Shutdown Executor Context Manager in Python medium
- How to Build a Producer-Consumer Pattern with asyncio.Queue in Python medium
- How to Cancel an asyncio Task with Graceful Cleanup in Python medium
- How to Convert Data in Parallel with ThreadPoolExecutor in Python easy
Keep learning
Related tutorials and quizzes for this topic.