穿越来找茬
35.39MB · 2025-09-13
FastAPI 的集成测试核心在于模拟真实环境中的多个服务模块交互,尤其针对认证、数据库、外部服务等场景。
pytest
+ httpx
+ asyncio
组合unittest.mock
替换外部依赖pytest-asyncio
管理异步事务回滚# 安装依赖
# pip install pytest==7.4.0 httpx==0.25.0 pytest-asyncio==0.21.1
flowchart TB
A[测试入口] --> B[模拟认证模块]
B --> C[调用用户服务模块]
C --> D[触发支付服务模块]
D --> E[验证结果]
from fastapi.testclient import TestClient
from unittest.mock import patch
import pytest
# 依赖模拟
@pytest.fixture
def mock_payment_gateway():
with patch("app.services.payment.PaymentGateway.charge") as mock:
mock.return_value = {"status": "success"}
yield
# 测试逻辑
def test_payment_flow(client: TestClient, mock_payment_gateway):
# 1. 获取认证Token
auth = client.post("/auth/login", json={"username": "test", "password": "pass"})
token = auth.json()["access_token"]
# 2. 创建订单
order = client.post(
"/orders",
json={"product_id": "p123", "qty": 2},
headers={"Authorization": f"Bearer {token}"}
)
order_id = order.json()["id"]
# 3. 执行支付
payment = client.post(
f"/orders/{order_id}/pay",
json={"card": "4111111111111111"},
headers={"Authorization": f"Bearer {token}"}
)
# 4. 验证结果
assert payment.status_code == 200
assert payment.json()["status"] == "completed"
关键点说明:
mock_payment_gateway
隔离第三方支付服务TestClient
维护请求上下文fixture
管理测试生命周期FastAPI 通过依赖注入系统管理认证流程:
flowchart LR
R[请求] --> D[认证依赖]
D -->|成功| U[用户对象]
D -->|失败| E[401错误]
from fastapi.security import OAuth2PasswordBearer
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")
async def get_current_user(token: str = Depends(oauth2_scheme)):
# 验证逻辑(实际项目需连接数据库)
return {"username": "test"} if token == "valid_token" else None
# 测试中伪造Token
def test_protected_route():
client = TestClient(app)
response = client.get("/protected", headers={"Authorization": "Bearer valid_token"})
assert response.status_code == 200
class AuthContext:
def __init__(self, client: TestClient):
self.client = client
self.token = None
def login(self, username, password):
res = self.client.post("/auth/login", json={"username": username, "password": password})
self.token = res.json().get("access_token")
return self
def get_headers(self):
return {"Authorization": f"Bearer {self.token}"}
# 测试用例
def test_user_profile():
auth = AuthContext(client).login("test", "pass")
res = client.get("/profile", headers=auth.get_headers())
assert res.json()["username"] == "test"
问题:如何避免认证测试中的Token硬编码风险?
解析:采用动态Token生成策略:
@pytest.fixture
动态获取Token正确实践:
@pytest.fixture
def valid_token(client: TestClient):
res = client.post("/auth/login", json={"username": "test", "password": "pass"})
return res.json()["access_token"]
def test_with_dynamic_token(client: TestClient, valid_token):
res = client.get("/protected", headers={"Authorization": f"Bearer {valid_token}"})
assert res.status_code == 200
原因分析:
解决方案:
# 错误示例
client.post("/orders", json={"product": "p123"}) # 缺少必要字段qty
# 正确方式
client.post("/orders", json={"product_id": "p123", "qty": 1})
预防建议:
response_model
自动校验@app.post("/orders", response_model=OrderResponse)
class Order(BaseModel):
product_id: str
qty: int = 1 # 默认值
notes: Optional[str] = None # 可选字段
解决方案:
get_current_user
逻辑@app.middleware("http")
async def debug_middleware(request: Request, call_next):
print("Headers:", request.headers)
return await call_next(request)