测试策略：单元测试、集成测试、端到端测试
测试金字塔
/\
/E2E\ 少量端到端测试
/------\
/集成测试\ 中等数量集成测试
/----------\
/ 单元测试 \ 大量单元测试
/--------------\单元测试
测试 Pydantic 模型
python
from pydantic import ValidationError
import pytest
def test_user_create_valid():
user = UserCreate(
email="test@example.com",
username="testuser",
password="password123"
)
assert user.email == "test@example.com"
def test_user_create_invalid_email():
with pytest.raises(ValidationError):
UserCreate(
email="invalid-email",
username="testuser",
password="password123"
)测试业务逻辑
python
def test_password_hashing():
password = "secret123"
hashed = get_password_hash(password)
assert hashed != password
assert verify_password(password, hashed)
assert not verify_password("wrong", hashed)集成测试
使用 TestClient
python
from fastapi.testclient import TestClient
from main import app
client = TestClient(app)
def test_create_user():
response = client.post(
"/users/",
json={
"email": "test@example.com",
"username": "testuser",
"password": "password123"
}
)
assert response.status_code == 201
data = response.json()
assert data["email"] == "test@example.com"
assert "id" in data
def test_create_user_duplicate_email():
# 第一次创建
client.post("/users/", json={
"email": "duplicate@example.com",
"username": "user1",
"password": "password123"
})
# 第二次创建（重复邮箱）
response = client.post("/users/", json={
"email": "duplicate@example.com",
"username": "user2",
"password": "password123"
})
assert response.status_code == 400
assert "already registered" in response.json()["detail"]测试数据库
使用测试数据库：
python
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from database import Base, get_db
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base.metadata.create_all(bind=engine)
def override_get_db():
try:
db = TestingSessionLocal()
yield db
finally:
db.close()
app.dependency_overrides[get_db] = override_get_db
@pytest.fixture(scope="function")
def db():
Base.metadata.drop_all(bind=engine)
Base.metadata.create_all(bind=engine)
yield测试认证
python
def test_login_success():
# 先注册用户
client.post("/users/", json={
"email": "login@example.com",
"username": "loginuser",
"password": "password123"
})
# 登录
response = client.post(
"/token",
data={"username": "loginuser", "password": "password123"}
)
assert response.status_code == 200
data = response.json()
assert "access_token" in data
assert data["token_type"] == "bearer"
def test_access_protected_route():
# 获取 token
login_response = client.post(
"/token",
data={"username": "loginuser", "password": "password123"}
)
token = login_response.json()["access_token"]
# 访问受保护路由
response = client.get(
"/users/me",
headers={"Authorization": f"Bearer {token}"}
)
assert response.status_code == 200
assert response.json()["username"] == "loginuser"pytest 配置
conftest.py：
python
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from main import app
from database import Base, get_db
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
@pytest.fixture(scope="function")
def db_session():
Base.metadata.drop_all(bind=engine)
Base.metadata.create_all(bind=engine)
db = TestingSessionLocal()
try:
yield db
finally:
db.close()
@pytest.fixture(scope="function")
def client(db_session):
def override_get_db():
try:
yield db_session
finally:
pass
app.dependency_overrides[get_db] = override_get_db
yield TestClient(app)
app.dependency_overrides.clear()覆盖率
bash
pip install pytest-cov
# 运行测试并生成覆盖率报告
pytest --cov=. --cov-report=html
# 查看报告
open htmlcov/index.html返回第四章概览