fastapi-testing

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

FastAPI Testing

FastAPI测试

Deep Knowledge: Use
mcp__documentation__fetch_docs
with technology:
fastapi-testing
for comprehensive documentation on TestClient, async testing, auth, and HTTP mocking.
深度知识:使用
mcp__documentation__fetch_docs
并指定技术:
fastapi-testing
,获取关于TestClient、异步测试、认证和HTTP模拟的完整文档。

Synchronous TestClient

同步TestClient

python
from fastapi.testclient import TestClient
from myapp.main import app

client = TestClient(app)

def test_read_root():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello World"}

def test_post_item():
    response = client.post(
        "/items/",
        json={"name": "Widget", "price": 9.99},
    )
    assert response.status_code == 201
python
from fastapi.testclient import TestClient
from myapp.main import app

client = TestClient(app)

def test_read_root():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello World"}

def test_post_item():
    response = client.post(
        "/items/",
        json={"name": "Widget", "price": 9.99},
    )
    assert response.status_code == 201

Lifespan Events

生命周期事件

python
def test_with_lifespan():
    with TestClient(app) as client:  # triggers startup/shutdown
        response = client.get("/items/")
        assert response.status_code == 200
python
def test_with_lifespan():
    with TestClient(app) as client:  # 触发启动/关闭事件
        response = client.get("/items/")
        assert response.status_code == 200

Async TestClient (httpx)

异步TestClient(httpx)

python
import pytest
from httpx import ASGITransport, AsyncClient
from myapp.main import app

@pytest.mark.anyio
async def test_async():
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="http://test"
    ) as ac:
        response = await ac.get("/")
    assert response.status_code == 200
python
import pytest
from httpx import ASGITransport, AsyncClient
from myapp.main import app

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

Shared fixture

共享夹具

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

Dependency Overrides

依赖注入重写

python
undefined
python
undefined

Production dependency

生产环境依赖

async def get_db() -> AsyncSession: async with AsyncSessionLocal() as session: yield session
async def get_db() -> AsyncSession: async with AsyncSessionLocal() as session: yield session

Test: override with test session

测试:用测试会话重写

@pytest.fixture def client(db_session): def override_get_db(): yield db_session app.dependency_overrides[get_db] = override_get_db with TestClient(app) as c: yield c app.dependency_overrides.clear()
@pytest.fixture def client(db_session): def override_get_db(): yield db_session app.dependency_overrides[get_db] = override_get_db with TestClient(app) as c: yield c app.dependency_overrides.clear()

Override user authentication

重写用户认证

async def override_current_user(): return User(id=1, username="testuser", is_active=True)
app.dependency_overrides[get_current_user] = override_current_user
undefined
async def override_current_user(): return User(id=1, username="testuser", is_active=True)
app.dependency_overrides[get_current_user] = override_current_user
undefined

Auth Testing — JWT

认证测试 — JWT

python
from datetime import timedelta
from myapp.auth import create_access_token

def get_test_token(username="testuser") -> str:
    return create_access_token(
        data={"sub": username},
        expires_delta=timedelta(minutes=30),
    )

def test_protected_endpoint():
    token = get_test_token()
    response = client.get(
        "/users/me/",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert response.status_code == 200

def test_no_token():
    response = client.get("/users/me/")
    assert response.status_code == 401
python
from datetime import timedelta
from myapp.auth import create_access_token

def get_test_token(username="testuser") -> str:
    return create_access_token(
        data={"sub": username},
        expires_delta=timedelta(minutes=30),
    )

def test_protected_endpoint():
    token = get_test_token()
    response = client.get(
        "/users/me/",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert response.status_code == 200

def test_no_token():
    response = client.get("/users/me/")
    assert response.status_code == 401

WebSocket Testing

WebSocket测试

python
def test_websocket():
    with client.websocket_connect("/ws") as ws:
        ws.send_text("hello")
        data = ws.receive_text()
        assert data == "Echo: hello"
python
def test_websocket():
    with client.websocket_connect("/ws") as ws:
        ws.send_text("hello")
        data = ws.receive_text()
        assert data == "Echo: hello"

File Upload Testing

文件上传测试

python
def test_upload_file():
    response = client.post(
        "/uploadfile/",
        files={"file": ("test.txt", b"hello world", "text/plain")},
    )
    assert response.status_code == 200
    assert response.json() == {"filename": "test.txt", "size": 11}
python
def test_upload_file():
    response = client.post(
        "/uploadfile/",
        files={"file": ("test.txt", b"hello world", "text/plain")},
    )
    assert response.status_code == 200
    assert response.json() == {"filename": "test.txt", "size": 11}

Background Tasks Testing

后台任务测试

python
from unittest.mock import patch

def test_task_called():
    with patch("fastapi.BackgroundTasks.add_task") as mock:
        response = client.post("/send-notification/user@example.com")
    assert response.status_code == 200
    mock.assert_called_once()
python
from unittest.mock import patch

def test_task_called():
    with patch("fastapi.BackgroundTasks.add_task") as mock:
        response = client.post("/send-notification/user@example.com")
    assert response.status_code == 200
    mock.assert_called_once()

HTTP Mocking — respx

HTTP模拟 — respx

python
import httpx
import respx

@respx.mock
def test_external_api():
    respx.get("https://api.example.com/users").mock(
        return_value=httpx.Response(200, json=[{"id": 1, "name": "Alice"}])
    )
    response = client.get("/proxy/users")
    assert response.status_code == 200
python
import httpx
import respx

@respx.mock
def test_external_api():
    respx.get("https://api.example.com/users").mock(
        return_value=httpx.Response(200, json=[{"id": 1, "name": "Alice"}])
    )
    response = client.get("/proxy/users")
    assert response.status_code == 200

With side effects

带副作用

@respx.mock async def test_async_mock(): respx.post("https://api.example.com/data").mock( side_effect=httpx.ConnectError ) with pytest.raises(httpx.ConnectError): async with httpx.AsyncClient() as ac: await ac.post("https://api.example.com/data")
undefined
@respx.mock async def test_async_mock(): respx.post("https://api.example.com/data").mock( side_effect=httpx.ConnectError ) with pytest.raises(httpx.ConnectError): async with httpx.AsyncClient() as ac: await ac.post("https://api.example.com/data")
undefined

HTTP Mocking — responses (requests library)

HTTP模拟 — responses(requests库)

python
import responses
import requests

@responses.activate
def test_with_responses():
    responses.get(
        "https://api.example.com/users",
        json=[{"id": 1}],
        status=200,
    )
    result = my_service.get_users()
    assert len(result) == 1
python
import responses
import requests

@responses.activate
def test_with_responses():
    responses.get(
        "https://api.example.com/users",
        json=[{"id": 1}],
        status=200,
    )
    result = my_service.get_users()
    assert len(result) == 1

HTTP Mocking — pytest-httpserver

HTTP模拟 — pytest-httpserver

python
def test_real_server(httpserver):
    httpserver.expect_request("/data").respond_with_json({"key": "value"})
    response = requests.get(httpserver.url_for("/data"))
    assert response.json() == {"key": "value"}
python
def test_real_server(httpserver):
    httpserver.expect_request("/data").respond_with_json({"key": "value"})
    response = requests.get(httpserver.url_for("/data"))
    assert response.json() == {"key": "value"}

Anti-Patterns

反模式

Anti-PatternSolution
app.dependency_overrides
not cleared
Use
yield
+
app.dependency_overrides.clear()
in fixture
Creating
TestClient
per test
Module or session scope
TestClient
Testing with real external APIsUse respx/responses to mock
Not using
ASGITransport
for async
Required for httpx with ASGI apps
反模式解决方案
未清理
app.dependency_overrides
在夹具中使用
yield
+
app.dependency_overrides.clear()
每个测试创建
TestClient
使用模块或会话作用域的
TestClient
使用真实外部API进行测试使用respx/responses进行模拟
异步测试未使用
ASGITransport
ASGI应用配合httpx必须使用该组件