测试基础 #

安装依赖 #

bash

pip install pytest
pip install httpx

TestClient #

基本使用 #

python

from fastapi import FastAPI
from fastapi.testclient import TestClient

app = FastAPI()

@app.get('/')
def read_root():
    return {'message': 'Hello World'}

client = TestClient(app)

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

测试 POST 请求 #

python

from fastapi import FastAPI
from fastapi.testclient import TestClient
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float

@app.post('/items/')
def create_item(item: Item):
    return item

client = TestClient(app)

def test_create_item():
    response = client.post(
        '/items/',
        json={'name': 'Item 1', 'price': 10.5}
    )
    assert response.status_code == 200
    assert response.json() == {'name': 'Item 1', 'price': 10.5}

测试查询参数 #

python

@app.get('/items/')
def read_items(skip: int = 0, limit: int = 10):
    return {'skip': skip, 'limit': limit}

def test_read_items():
    response = client.get('/items/?skip=0&limit=5')
    assert response.status_code == 200
    assert response.json() == {'skip': 0, 'limit': 5}

测试路径参数 #

python

@app.get('/items/{item_id}')
def read_item(item_id: int):
    return {'item_id': item_id}

def test_read_item():
    response = client.get('/items/1')
    assert response.status_code == 200
    assert response.json() == {'item_id': 1}

测试请求头 #

python

@app.get('/headers/')
def read_headers(token: str = Header(...)):
    return {'token': token}

def test_read_headers():
    response = client.get('/headers/', headers={'token': 'secret'})
    assert response.status_code == 200
    assert response.json() == {'token': 'secret'}

pytest 使用 #

基本测试文件 #

python

from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_read_root():
    response = client.get('/')
    assert response.status_code == 200

def test_read_item():
    response = client.get('/items/1')
    assert response.status_code == 200
    assert response.json()['item_id'] == 1

运行测试 #

bash

pytest test_main.py

pytest -v test_main.py

pytest -v --tb=short test_main.py

使用 fixture #

python

import pytest
from fastapi.testclient import TestClient
from main import app

@pytest.fixture
def client():
    return TestClient(app)

def test_read_root(client):
    response = client.get('/')
    assert response.status_code == 200

依赖覆盖 #

覆盖依赖 #

python

from fastapi import FastAPI, Depends
from fastapi.testclient import TestClient

app = FastAPI()

async def get_db():
    return 'production_db'

@app.get('/items/')
def read_items(db: str = Depends(get_db)):
    return {'db': db}

def override_get_db():
    return 'test_db'

app.dependency_overrides[get_db] = override_get_db

client = TestClient(app)

def test_read_items():
    response = client.get('/items/')
    assert response.json() == {'db': 'test_db'}

清除覆盖 #

python

def test_read_items():
    response = client.get('/items/')
    assert response.json() == {'db': 'test_db'}
    app.dependency_overrides = {}

使用 fixture 管理覆盖 #

python

@pytest.fixture
def client():
    app.dependency_overrides[get_db] = override_get_db
    yield TestClient(app)
    app.dependency_overrides = {}

测试认证 #

测试 OAuth2 #

python

from fastapi.security import OAuth2PasswordBearer

app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl='token')

@app.get('/users/me')
def read_users_me(token: str = Depends(oauth2_scheme)):
    return {'token': token}

def test_with_token():
    response = client.get(
        '/users/me',
        headers={'Authorization': 'Bearer test-token'}
    )
    assert response.status_code == 200

测试登录 #

python

from fastapi.security import OAuth2PasswordRequestForm

@app.post('/token')
def login(form_data: OAuth2PasswordRequestForm = Depends()):
    return {'access_token': form_data.username, 'token_type': 'bearer'}

def test_login():
    response = client.post(
        '/token',
        data={'username': 'john', 'password': 'secret'}
    )
    assert response.status_code == 200
    assert 'access_token' in response.json()

测试文件上传 #

python

from fastapi import FastAPI, File, UploadFile
from fastapi.testclient import TestClient

app = FastAPI()

@app.post('/files/')
async def create_file(file: bytes = File(...)):
    return {'file_size': len(file)}

client = TestClient(app)

def test_create_file():
    response = client.post(
        '/files/',
        files={'file': ('test.txt', b'file content')}
    )
    assert response.status_code == 200
    assert response.json() == {'file_size': 12}

测试 UploadFile #

python

@app.post('/upload/')
async def upload_file(file: UploadFile = File(...)):
    content = await file.read()
    return {'filename': file.filename, 'size': len(content)}

def test_upload_file():
    response = client.post(
        '/upload/',
        files={'file': ('test.txt', b'file content')}
    )
    assert response.status_code == 200
    assert response.json() == {'filename': 'test.txt', 'size': 12}

测试异常 #

测试 404 #

python

@app.get('/items/{item_id}')
def read_item(item_id: int):
    if item_id == 0:
        raise HTTPException(status_code=404, detail='Item not found')
    return {'item_id': item_id}

def test_item_not_found():
    response = client.get('/items/0')
    assert response.status_code == 404
    assert response.json()['detail'] == 'Item not found'

测试验证错误 #

python

@app.post('/items/')
def create_item(item: Item):
    return item

def test_validation_error():
    response = client.post('/items/', json={'name': 'Item'})
    assert response.status_code == 422

测试异步代码 #

异步测试 #

python

import pytest
import httpx
from main import app

@pytest.mark.asyncio
async def test_async():
    async with httpx.AsyncClient(app=app, base_url='http://test') as client:
        response = await client.get('/')
        assert response.status_code == 200

pytest-asyncio #

bash

pip install pytest-asyncio

python

import pytest
from httpx import AsyncClient
from main import app

@pytest.mark.asyncio
async def test_async_root():
    async with AsyncClient(app=app, base_url='http://test') as client:
        response = await client.get('/')
        assert response.status_code == 200

完整测试示例 #

python

import pytest
from fastapi import FastAPI, Depends, HTTPException
from fastapi.testclient import TestClient
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float

items_db = {}

oauth2_scheme = OAuth2PasswordBearer(tokenUrl='token')

def get_current_user(token: str = Depends(oauth2_scheme)):
    if token != 'valid-token':
        raise HTTPException(status_code=401, detail='Invalid token')
    return {'username': 'testuser'}

@app.get('/')
def read_root():
    return {'message': 'Hello World'}

@app.post('/items/', response_model=Item)
def create_item(item: Item):
    items_db[item.name] = item
    return item

@app.get('/items/{item_name}', response_model=Item)
def read_item(item_name: str):
    if item_name not in items_db:
        raise HTTPException(status_code=404, detail='Item not found')
    return items_db[item_name]

@app.get('/protected/')
def protected_route(user: dict = Depends(get_current_user)):
    return {'user': user}

@pytest.fixture
def client():
    items_db.clear()
    return TestClient(app)

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

def test_create_item(client):
    response = client.post('/items/', json={'name': 'Item 1', 'price': 10.5})
    assert response.status_code == 200
    assert response.json() == {'name': 'Item 1', 'price': 10.5}

def test_read_item(client):
    client.post('/items/', json={'name': 'Item 1', 'price': 10.5})
    response = client.get('/items/Item 1')
    assert response.status_code == 200
    assert response.json() == {'name': 'Item 1', 'price': 10.5}

def test_read_item_not_found(client):
    response = client.get('/items/nonexistent')
    assert response.status_code == 404

def test_protected_route_unauthorized(client):
    response = client.get('/protected/')
    assert response.status_code == 401

def test_protected_route_authorized(client):
    response = client.get(
        '/protected/',
        headers={'Authorization': 'Bearer valid-token'}
    )
    assert response.status_code == 200
    assert response.json()['user']['username'] == 'testuser'

下一步 #

现在你已经掌握了测试基础，接下来学习 Docker 部署，了解 FastAPI 的部署方法！