Testing (AsyncClient)
TL;DR — Pour tester une app FastAPI moderne (async), on n'utilise plus
TestClient(synchrone, qui démarre une boucle d'événements parallèle et casse vite). On branche un clienthttpx.AsyncClientdirectement sur l'app viaASGITransport: pas de serveur réseau, pas de port, mais du vrai async de bout en bout. La discipline senior tient en quatre piliers : (1) surcharger les dépendances avecapp.dependency_overridesplutôt que mocker à la sauvage, (2) une base de données réelle jetable par test (transaction rollback ou Testcontainers) plutôt qu'un mock de l'ORM, (3) mocker uniquement la frontière externe — l'appel HTTP au LLM, le paiement, le mail — au plus près du réseau, et (4) tester la forme du contrat (status, schéma Pydantic, en-têtes) plutôt que des chaînes exactes. Pour une app qui sert un agent Anthropic, le test critique n'est pas « est-ce que Claude répond bien » (non déterministe) mais « est-ce que ma boucle tool-use, mon streaming SSE et mes retries se comportent correctement quand le SDK renvoie X ».
🧠 Mental model
Un test, c'est un banc d'essai moteur. On ne teste pas la voiture sur l'autoroute (ça, c'est la prod, l'e2e) : on sort le moteur, on le boulonne sur un bâti, et on lui injecte du carburant contrôlé pour mesurer sa réponse. Le moteur, c'est votre code applicatif. Le bâti, c'est AsyncClient + ASGITransport. Le carburant contrôlé, ce sont les dependency overrides et les mocks de frontière.
La question de fond en test FastAPI est toujours : où est la frontière de mon banc d'essai ? Trop serrée (je mocke mes propres fonctions), je teste mon mock, pas mon code. Trop large (je tape la vraie API Anthropic, la vraie DB de prod), ce n'est plus un test unitaire, c'est flaky et lent.
╔═══════════════════════════════════╗
test() ║ VOTRE APP (sous test, RÉELLE) ║
┌────────┐ ║ ║
│ Async │ ASGI ║ router → service → repo ║ mock ici
│ Client │─────►║ │ │ │ ║ ┌──────────┐
└────────┘ (in- ║ │ │ └─► DB ───────╫──►│ Postgres │ jetable
proc)║ │ └─► AnthropicClient ───╫──►│ (mock) │ frontière
║ └─► auth dep (OVERRIDE) ─────────╫─ └──────────┘ externe
╚═══════════════════════════════════╝
▲ pas de socket TCP : httpx parle ASGI directement à l'appTout ce qui est à l'intérieur de la boîte est exécuté pour de vrai. Seules les arêtes qui sortent (réseau externe, horloge, aléatoire) sont contrôlées.
Le socle : AsyncClient + ASGITransport
L'app de référence pour tout le chapitre :
# app/main.py
from __future__ import annotations
from dataclasses import dataclass
from fastapi import Depends, FastAPI, HTTPException, status
from pydantic import BaseModel, Field
class ItemIn(BaseModel):
name: str = Field(min_length=1, max_length=100)
price_cents: int = Field(gt=0)
class ItemOut(ItemIn):
id: int
@dataclass
class CurrentUser:
id: int
is_admin: bool
# Dépendance d'auth — c'est ELLE qu'on surchargera dans les tests
async def get_current_user() -> CurrentUser:
# En vrai : décode un JWT, tape Redis, etc.
raise HTTPException(status.HTTP_401_UNAUTHORIZED)
# Repo en mémoire pour l'exemple (en vrai : une DB)
class ItemRepo:
def __init__(self) -> None:
self._items: dict[int, ItemOut] = {}
self._seq = 0
async def create(self, data: ItemIn) -> ItemOut:
self._seq += 1
item = ItemOut(id=self._seq, **data.model_dump())
self._items[item.id] = item
return item
async def get(self, item_id: int) -> ItemOut | None:
return self._items.get(item_id)
_repo = ItemRepo()
async def get_repo() -> ItemRepo:
return _repo
app = FastAPI()
@app.post("/items", response_model=ItemOut, status_code=201)
async def create_item(
body: ItemIn,
user: CurrentUser = Depends(get_current_user),
repo: ItemRepo = Depends(get_repo),
) -> ItemOut:
return await repo.create(body)
@app.get("/items/{item_id}", response_model=ItemOut)
async def read_item(
item_id: int,
repo: ItemRepo = Depends(get_repo),
) -> ItemOut:
item = await repo.get(item_id)
if item is None:
raise HTTPException(status.HTTP_404_NOT_FOUND, "item not found")
return item❌ La façon datée (TestClient synchrone)
# NE FAITES PLUS ÇA dans une app async
from fastapi.testclient import TestClient
from app.main import app
client = TestClient(app)
def test_read():
r = client.get("/items/1")
assert r.status_code == 404TestClient (basé sur Starlette/requests) est synchrone : il fait tourner une boucle d'événements dans un thread à part pour exécuter votre code async. Ça marche pour des cas triviaux, mais ça explose dès que vos tests doivent eux-mêmes faire de l'async (une fixture DB async, un await sur un mock, du streaming) : vous vous retrouvez avec deux boucles, des RuntimeError: Event loop is closed, des deadlocks. Le test n'est plus le reflet de la prod.
✅ La façon idiomatique (async de bout en bout)
# tests/conftest.py
from __future__ import annotations
from collections.abc import AsyncIterator
import pytest
from httpx import ASGITransport, AsyncClient
from app.main import app, get_current_user, CurrentUser
@pytest.fixture
def anyio_backend() -> str:
return "asyncio" # on ne teste pas sur trio
@pytest.fixture
async def client() -> AsyncIterator[AsyncClient]:
# ASGITransport = httpx parle directement à l'app, AUCUN socket réseau.
# base_url est cosmétique mais obligatoire (httpx exige un host absolu).
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as ac:
# Surcharge l'auth : tout test est authentifié par défaut
app.dependency_overrides[get_current_user] = lambda: CurrentUser(
id=42, is_admin=False
)
yield ac
app.dependency_overrides.clear() # ESSENTIEL : sinon fuite entre tests# tests/test_items.py
import pytest
pytestmark = pytest.mark.anyio # tous les tests du module sont async
async def test_create_then_read(client):
r = await client.post("/items", json={"name": "Paté", "price_cents": 1290})
assert r.status_code == 201
item_id = r.json()["id"]
r = await client.get(f"/items/{item_id}")
assert r.status_code == 200
body = r.json()
assert body["name"] == "Paté"
assert body["price_cents"] == 1290
async def test_validation_422(client):
# price_cents négatif → Pydantic rejette AVANT le handler
r = await client.post("/items", json={"name": "x", "price_cents": -1})
assert r.status_code == 422
# On teste la FORME de l'erreur, pas le texte exact
detail = r.json()["detail"][0]
assert detail["loc"] == ["body", "price_cents"]
assert detail["type"] == "greater_than"
async def test_not_found(client):
r = await client.get("/items/999999")
assert r.status_code == 404Côté config, le plus court chemin avec pytest-anyio (ou anyio + pytest) :
# pyproject.toml
[tool.pytest.ini_options]
addopts = "-q --strict-markers"
markers = ["anyio: async tests via anyio"]pip install pytest anyio httpx
pytestPourquoi
anyioet paspytest-asyncio? FastAPI/Starlette tournent sur AnyIO. Utiliser le même backend dans les tests évite les surprises de boucle.pytest-asynciomarche aussi (@pytest.mark.asyncio+asyncio_mode = "auto"), mais soyez cohérent dans tout le repo — mélanger les deux est la source n°1 de tests flaky async.
Dependency overrides : votre arme principale
app.dependency_overrides est un simple dict[Callable, Callable]. FastAPI consulte ce dict avant de résoudre une dépendance. C'est toujours préférable à unittest.mock.patch parce que ça respecte le système d'injection, le typage, et le cycle de vie.
# Tester un endpoint admin-only sans fabriquer un vrai JWT admin
async def test_admin_route(client):
from app.main import get_current_user, CurrentUser
app.dependency_overrides[get_current_user] = lambda: CurrentUser(
id=1, is_admin=True
)
r = await client.delete("/admin/items/1")
assert r.status_code in (200, 204)Règle d'or : on surcharge la dépendance, on ne patche pas la fonction métier. Si vous vous surprenez à écrire patch("app.services.compute_price"), demandez-vous pourquoi compute_price n'est pas une dépendance injectée.
Base de données : réelle et jetable, pas mockée
Mocker un ORM async (AsyncSession) est un piège : vous réimplémentez SQLAlchemy en moins bien, et vous ne testez ni vos requêtes, ni vos contraintes, ni vos migrations. Deux stratégies senior :
1. Rollback par test (rapide, même process). Chaque test ouvre une transaction, fait son travail, et roll back à la fin. La DB reste vierge.
# tests/conftest.py (extrait DB)
from collections.abc import AsyncIterator
import pytest
from sqlalchemy.ext.asyncio import (
AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine,
)
from app.db import Base, get_session
# Postgres jetable via Testcontainers en CI ; ici un engine de test
TEST_URL = "postgresql+asyncpg://test:test@localhost:5432/test"
@pytest.fixture(scope="session")
async def engine() -> AsyncIterator[AsyncEngine]:
eng = create_async_engine(TEST_URL)
async with eng.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
yield eng
await eng.dispose()
@pytest.fixture
async def session(engine: AsyncEngine) -> AsyncIterator[AsyncSession]:
# Connexion + transaction externe qu'on rollback : isolation parfaite
async with engine.connect() as conn:
trans = await conn.begin()
maker = async_sessionmaker(bind=conn, expire_on_commit=False)
async with maker() as s:
yield s
await trans.rollback() # tout disparaît, même les COMMIT internes
@pytest.fixture
async def client(session: AsyncSession) -> AsyncIterator[AsyncClient]:
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as ac:
# L'app utilise LA session du test → mêmes données, même transaction
app.dependency_overrides[get_session] = lambda: session
yield ac
app.dependency_overrides.clear()2. Testcontainers (fidèle, plus lent). En CI, lancez un vrai Postgres éphémère. C'est le standard pour les tests d'intégration : vous testez sur le même moteur qu'en prod (JSONB, contraintes, ON CONFLICT…), pas sur SQLite qui mentira sur la moitié des comportements.
SQLite en mémoire pour « aller vite » est une fausse économie : dialecte différent, pas de vraies contraintes FK par défaut, pas de types Postgres. Vous gagnez 2 secondes et perdez la confiance dans vos tests.
⚙️ En production
- Isolation et fuite d'état. Le bug le plus fréquent : oublier
app.dependency_overrides.clear(). Un override posé dans un test « contamine » les suivants, et vous obtenez des échecs qui dépendent de l'ordre des tests. Toujours nettoyer dans la fixture (viayield), jamais à la main dans le test. Pareil pour les singletons modules-level (_repoci-dessus) : préférez une fabrique par test. - Flakiness async. Boucle d'événements partagée mal scoppée,
asyncio.sleepréel dans le code,datetime.now()non gelé → tests instables. Injectez l'horloge (now: Callable[[], datetime]en dépendance), et gelez-la (freezegunou un fake). Bannissez lessleepréels : si vous testez un timeout, mockez le timer. - Performance. L'
ASGITransportest in-process, donc rapide ; le coût est presque toujours la DB. Mettez le schéma en place une fois (scope="session") et isolez par rollback de transaction. Parallélisez avecpytest-xdist(-n auto) — mais alors chaque worker a besoin de sa base (suffixez le nom de DB parPYTEST_XDIST_WORKER). - Sécurité. Ne testez pas seulement le chemin heureux : écrivez explicitement les tests « 401 sans token », « 403 user non-admin », « 404 vs 403 » (ne jamais révéler l'existence d'une ressource à un non-autorisé). Vérifiez les en-têtes de sécurité (CORS,
Content-Type) et que les secrets ne fuitent jamais dans les réponses d'erreur. - Observabilité. Capturez les logs (
caplog) pour vérifier qu'une erreur émet bien le bon log structuré avec lerequest_id. Vérifiez que les spans/metrics sont émis (mock de l'exporter OTel). Un bug silencieux qui n'apparaît pas dans les logs est pire qu'un crash. - Couverture vs valeur. 100 % de couverture sur des getters triviaux ne vaut rien. Le ROI est dans les bords : pagination limite, concurrence, erreurs réseau du LLM, retries, désérialisation d'un payload partiel. Ciblez la couverture de branches sur le code métier risqué, pas la couverture de lignes globale.
- Tradeoff de la pyramide. Beaucoup de tests unitaires de services purs (rapides, déterministes), une couche moyenne de tests d'endpoints via
AsyncClient(le sweet spot FastAPI), une poignée d'e2e contre un environnement réel. Inverser la pyramide = suite lente et flaky que personne ne fait tourner.
Tester un endpoint qui sert un agent Anthropic
C'est ici que la discipline « mocker à la frontière » devient vitale. Votre endpoint appelle Claude via AsyncAnthropic ; la réponse de Claude est non déterministe et payante. On ne tape donc jamais l'API réelle dans les tests : on injecte un faux client à la frontière, et on teste notre logique (boucle tool-use, parsing, retries, streaming).
L'app, conçue pour être testable (le client Anthropic est une dépendance injectée, pas un import global) :
# app/agent.py
from __future__ import annotations
from collections.abc import AsyncIterator
from anthropic import AsyncAnthropic
from fastapi import APIRouter, Depends
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
router = APIRouter()
MODEL = "claude-sonnet-4-6" # flagship dispo : claude-opus-4-8
def get_anthropic() -> AsyncAnthropic:
# Surchargé dans les tests via dependency_overrides
return AsyncAnthropic()
class ChatIn(BaseModel):
prompt: str
@router.post("/chat/stream")
async def chat_stream(
body: ChatIn,
anthropic: AsyncAnthropic = Depends(get_anthropic),
) -> StreamingResponse:
async def sse() -> AsyncIterator[str]:
async with anthropic.messages.stream(
model=MODEL,
max_tokens=1024,
messages=[{"role": "user", "content": body.prompt}],
) as stream:
async for text in stream.text_stream:
yield f"data: {text}\n\n"
yield "data: [DONE]\n\n"
return StreamingResponse(sse(), media_type="text/event-stream")Mocker le streaming Anthropic
Le SDK expose messages.stream(...) comme context manager async dont .text_stream est un itérateur async. On fabrique un faux qui imite exactement cette forme :
# tests/test_agent.py
import pytest
from collections.abc import AsyncIterator
from httpx import ASGITransport, AsyncClient
from app.main import app
from app.agent import get_anthropic
pytestmark = pytest.mark.anyio
class FakeStream:
"""Imite le retour de AsyncAnthropic.messages.stream(...)."""
def __init__(self, chunks: list[str]) -> None:
self._chunks = chunks
async def __aenter__(self) -> "FakeStream":
return self
async def __aexit__(self, *exc) -> None:
return None
@property
async def text_stream(self) -> AsyncIterator[str]: # noqa: parodie
for c in self._chunks:
yield c
class FakeMessages:
def __init__(self, chunks: list[str]) -> None:
self._chunks = chunks
def stream(self, **_: object) -> FakeStream:
return FakeStream(self._chunks)
class FakeAnthropic:
def __init__(self, chunks: list[str]) -> None:
self.messages = FakeMessages(chunks)
async def test_chat_stream_emits_sse():
fake = FakeAnthropic(chunks=["Bon", "jour", " 👋"])
app.dependency_overrides[get_anthropic] = lambda: fake
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as ac:
async with ac.stream("POST", "/chat/stream",
json={"prompt": "salut"}) as r:
assert r.status_code == 200
assert r.headers["content-type"].startswith("text/event-stream")
received = [line async for line in r.aiter_lines()]
body = "".join(received)
# `aiter_lines()` retire les séparateurs de ligne : on cherche le payload nu.
assert "data: Bon" in body
assert "data: [DONE]" in body
app.dependency_overrides.clear()Notez : on teste que chaque token devient un event SSE et que le sentinel [DONE] est bien émis — le contrat de notre couche transport. On ne teste pas le contenu généré par Claude.
Tester la boucle tool-use et les retries
La vraie valeur d'un agent est sa boucle : Claude renvoie stop_reason="tool_use", on exécute l'outil, on renvoie le tool_result, on reboucle. C'est du code à nous, donc 100 % testable de façon déterministe en scriptant les réponses du faux client.
# app/loop.py — boucle tool-use minimale
from collections.abc import Awaitable, Callable
from typing import Any
from anthropic import AsyncAnthropic
# Un outil = une coroutine qui prend des kwargs et renvoie n'importe quoi de sérialisable.
Tool = Callable[..., Awaitable[Any]]
async def run_agent(client: AsyncAnthropic, prompt: str,
tools: dict[str, Tool], max_turns: int = 5) -> str:
messages: list[dict[str, Any]] = [{"role": "user", "content": prompt}]
for _ in range(max_turns):
msg = await client.messages.create(
model="claude-sonnet-4-6", max_tokens=1024,
tools=[...], messages=messages,
)
if msg.stop_reason != "tool_use":
return "".join(b.text for b in msg.content if b.type == "text")
# exécuter chaque tool_use, accumuler les tool_result, reboucler
messages.append({"role": "assistant", "content": msg.content})
results = []
for block in msg.content:
if block.type == "tool_use":
out = await tools[block.name](**block.input)
results.append({"type": "tool_result",
"tool_use_id": block.id, "content": str(out)})
messages.append({"role": "user", "content": results})
raise RuntimeError("max_turns dépassé sans réponse finale")Pour le tester, on scripte une séquence de réponses : « d'abord demande l'outil, puis réponds ». On vérifie que l'outil est appelé une fois avec les bons arguments, et que la boucle s'arrête. On ajoute un test où Claude boucle indéfiniment (stop_reason toujours tool_use) → on vérifie que max_turns lève bien RuntimeError (sinon : boucle infinie + facture explosée en prod).
Pour les retries : le SDK Anthropic réessaie déjà les 429/5xx/erreurs réseau avec backoff. En test, faites que le faux client lève anthropic.RateLimitError puis réussisse, et vérifiez que votre code propage ou dégrade correctement (fallback claude-haiku-4-5, message d'erreur propre, pas un 500 nu). Utilisez les exceptions typées du SDK (RateLimitError, APIStatusError, APIConnectionError) — jamais un except Exception fourre-tout.
🏋️ Exercices
1. Le harnais de base — Objectif : monter le banc d'essai
Crée une app FastAPI avec POST /signup (Pydantic v2, validation email + mot de passe ≥ 8) et un conftest.py avec une fixture client (AsyncClient + ASGITransport). Écris 3 tests : succès 201, email invalide 422 (vérifie loc et type, pas le texte), doublon 409. Indice/Solution : modèle EmailStr ; le 409 nécessite un repo en mémoire partagé via fixture — attention à ne pas le partager entre tests (refais-le à chaque test ou via override de get_repo).
2. DB jetable par transaction — Objectif : tester le vrai SQL
Ajoute SQLAlchemy async + Postgres (Testcontainers ou local). Implémente la fixture session qui ouvre une connexion, démarre une transaction, et rollback à la fin. Branche get_session via override. Écris un test qui crée 2 users et vérifie qu'un 3e test démarre sur une DB vide. Indice/Solution : la clé est engine.connect() + conn.begin() puis async_sessionmaker(bind=conn) ; même les commit() internes restent dans la transaction externe et disparaissent au rollback.
3. Auth et chemins refusés — Objectif : sécurité par les tests
Protège DELETE /items/{id} (admin-only). Écris : 401 sans override d'auth, 403 user non-admin, 204 admin. Bonus : vérifie qu'un non-admin reçoit 403 et non 404 révélant l'existence de la ressource (ou l'inverse selon ta politique — documente le choix dans le test). Indice/Solution : trois variantes d'override de get_current_user ; pour le 401, retire l'override et laisse la vraie dépendance lever.
4. Streaming SSE déterministe — Objectif : tester le transport LLM sans LLM
Reprends chat_stream. Avec un FakeAnthropic scriptable, vérifie : (a) chaque chunk → un event data:, (b) le sentinel [DONE], (c) le Content-Type est text/event-stream. Bonus : simule une coupure client (annule la requête en cours de stream) et vérifie qu'aucune exception ne remonte côté serveur. Indice/Solution : utilise ac.stream("POST", ...) + aiter_lines() ; pour la coupure, lève asyncio.CancelledError dans le générateur fake et entoure ta logique serveur d'un try/finally qui ferme proprement le stream Anthropic.
5. Boucle tool-use bornée — Objectif : production-grade, casser puis réparer
Teste run_agent : (a) un tour avec outil puis réponse finale → l'outil est appelé 1× avec les bons args ; (b) Claude qui demande l'outil à l'infini → max_turns lève RuntimeError (sans boucle infinie dans le test). Puis casse : retire la borne max_turns et observe le test (a) qui passe mais (b) qui pend → remets la borne. Indice/Solution : le faux messages.create retourne des objets avec .stop_reason et .content scriptés via une liste consommée à chaque appel ; mocke l'outil avec un AsyncMock et asserte mock.assert_awaited_once_with(...).
6. Retries et dégradation — Objectif : résilience
Fais que le faux client lève anthropic.RateLimitError deux fois puis réussisse. Vérifie que ta couche fait un fallback vers claude-haiku-4-5 après N échecs (ou propage une 503 propre, jamais un 500 nu). Vérifie aussi qu'une APIConnectionError produit une réponse d'erreur sans fuiter la stack trace. Indice/Solution : capture l'exception typée du SDK, mappe-la vers une HTTPException(503) via un exception handler FastAPI ; teste le handler en provoquant l'erreur et en asssertant le body JSON normalisé.
🎤 En entretien
Q : Pourquoi httpx.AsyncClient plutôt que TestClient pour une app FastAPI async ?TestClient est synchrone et pilote une boucle d'événements à part, ce qui casse dès que les tests ou fixtures sont eux-mêmes async ; AsyncClient + ASGITransport parle ASGI in-process en vrai async, sans socket, et reflète fidèlement la prod.
Q : Comment isoler la base de données entre tests sans la recréer à chaque fois ? On crée le schéma une fois par session, et chaque test tourne dans une transaction qu'on rollback en teardown — la session de l'app est branchée sur la même connexion via dependency_overrides, donc tout disparaît à la fin, y compris les commits internes.
Q : Comment tester un endpoint qui appelle un LLM non déterministe et payant ? On injecte le client Anthropic comme dépendance et on le remplace par un faux scriptable à la frontière réseau ; on teste notre logique déterministe (boucle tool-use bornée, mapping SSE, retries, dégradation sur exceptions typées), jamais le contenu généré.
Q : Quel est le piège n°1 des suites de tests FastAPI et comment l'éviter ? La fuite d'app.dependency_overrides entre tests, qui crée des échecs dépendants de l'ordre : on nettoie toujours via yield dans la fixture (overrides.clear()), et on évite les singletons module-level au profit de fabriques par test.