「單元測試都過了,為什麼整合起來還是壞掉?」資深工程師搖搖頭,「因為你只測試了零件,沒測試組裝。」
今天,我們要用過去 28 天學到的所有 TDD 技巧,打造一個真正可靠的 Todo API 整合測試套件!
基礎測試 [✅] Kata 練習 [✅] 框架實戰 [✅] 整合部署 [🔄]
Day 1-10 Day 11-17 Day 18-27 Day 28-30
↑ 我們在這裡!
# 建立 tests/day29/test_integration_challenge.py
import pytest
from datetime import datetime
def test_why_integration_matters():
"""展示為何需要整合測試"""
# 單元測試 A: ✅ 通過
def format_date(date):
return date.strftime('%Y-%m-%d')
# 單元測試 B: ✅ 通過
def calculate_days(start, end):
return (end - start).days
# 整合時: ❌ 時區問題!
start = datetime(2024, 1, 1, 0, 0, 0)
end = datetime(2024, 1, 2, 23, 59, 59)
days = calculate_days(start, end)
assert days == 1 # 可能失敗!
import pytest
import asyncio
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from src.todo.main import app
from src.todo.database import Base, get_db
TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"
@pytest.fixture(scope="function")
async def test_db():
"""建立測試資料庫"""
engine = create_async_engine(TEST_DATABASE_URL)
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
TestingSessionLocal = sessionmaker(
engine, class_=AsyncSession, expire_on_commit=False
)
async with TestingSessionLocal() as session:
yield session
await engine.dispose()
@pytest.fixture
async def client(test_db: AsyncSession):
"""建立測試客戶端"""
app.dependency_overrides[get_db] = lambda: test_db
async with AsyncClient(app=app, base_url="http://test") as ac:
yield ac
app.dependency_overrides.clear()
import pytest
from httpx import AsyncClient
from datetime import datetime
class TestTodoIntegration:
"""Todo 整合測試套件"""
@pytest.mark.asyncio
async def test_complete_todo_lifecycle(self, client: AsyncClient):
"""測試 Todo 完整生命週期"""
# 建立新 Todo
create_response = await client.post(
"/api/todos",
json={"title": "整合測試任務", "description": "測試流程"}
)
assert create_response.status_code == 201
todo_id = create_response.json()["id"]
# 取得列表
list_response = await client.get("/api/todos")
assert len(list_response.json()) == 1
# 更新 Todo
update_response = await client.put(
f"/api/todos/{todo_id}",
json={"title": "更新任務", "completed": True}
)
assert update_response.json()["completed"] is True
# 刪除並確認
await client.delete(f"/api/todos/{todo_id}")
verify_response = await client.get(f"/api/todos/{todo_id}")
assert verify_response.status_code == 404
@pytest.mark.asyncio
async def test_bulk_operations(self, client: AsyncClient):
"""測試批量操作"""
# 建立 5 個 Todo
ids = []
for i in range(1, 6):
resp = await client.post("/api/todos", json={"title": f"Task {i}"})
ids.append(resp.json()["id"])
# 測試分頁
page1 = await client.get("/api/todos?skip=0&limit=3")
assert len(page1.json()) == 3
# 更新部分項目並測試篩選
for i in [0, 2, 4]:
await client.put(f"/api/todos/{ids[i]}", json={"completed": True})
completed = await client.get("/api/todos?completed=true")
assert len(completed.json()) == 3
import pytest
from httpx import AsyncClient
class TestErrorHandling:
"""錯誤處理整合測試"""
@pytest.mark.asyncio
async def test_not_found_error(self, client: AsyncClient):
"""測試 404 錯誤"""
response = await client.get("/api/todos/999")
assert response.status_code == 404
assert "not found" in response.json()["detail"].lower()
@pytest.mark.asyncio
async def test_validation_error(self, client: AsyncClient):
"""測試驗證錯誤"""
response = await client.post("/api/todos", json={})
assert response.status_code == 422
assert "title" in str(response.json()["detail"])
@pytest.mark.asyncio
async def test_concurrent_updates(self, client: AsyncClient):
"""測試並發更新"""
response = await client.post("/api/todos", json={"title": "Test"})
todo_id = response.json()["id"]
import asyncio
results = await asyncio.gather(
client.put(f"/api/todos/{todo_id}", json={"title": "V1"}),
client.put(f"/api/todos/{todo_id}", json={"title": "V2"}),
return_exceptions=True
)
assert all(not isinstance(r, Exception) for r in results)
import pytest
import time
import asyncio
from httpx import AsyncClient
class TestPerformance:
"""性能整合測試"""
@pytest.mark.asyncio
async def test_response_time(self, client: AsyncClient):
"""測試回應時間"""
await client.get("/api/todos") # 預熱
start = time.perf_counter()
response = await client.get("/api/todos")
elapsed = time.perf_counter() - start
assert response.status_code == 200
assert elapsed < 0.1 # 應在 100ms 內
@pytest.mark.asyncio
async def test_load_handling(self, client: AsyncClient):
"""測試負載處理能力"""
# 建立 20 個 Todo 並同時查詢
ids = []
for i in range(20):
resp = await client.post("/api/todos", json={"title": f"Load {i}"})
ids.append(resp.json()["id"])
# 並發請求
start = time.perf_counter()
results = await asyncio.gather(
*[client.get(f"/api/todos/{id}") for id in ids]
)
elapsed = time.perf_counter() - start
assert all(r.status_code == 200 for r in results)
assert elapsed < 2.0 # 應在 2 秒內完成
import pytest
import asyncio
from httpx import AsyncClient
class TestDataConsistency:
"""資料一致性測試"""
@pytest.mark.asyncio
async def test_timestamp_consistency(self, client: AsyncClient):
"""測試時間戳一致性"""
response = await client.post("/api/todos", json={"title": "Time"})
created_at = response.json()["created_at"]
await asyncio.sleep(0.1)
update = await client.put(
f"/api/todos/{response.json()['id']}",
json={"title": "Updated"}
)
assert update.json()["created_at"] == created_at
assert update.json()["updated_at"] > created_at
@pytest.mark.asyncio
async def test_transaction_consistency(self, client: AsyncClient):
"""測試事務一致性"""
# 無效請求不應影響資料庫
await client.post("/api/todos", json={"title": "x" * 300})
response = await client.get("/api/todos")
assert len(response.json()) == 0
# 完整的整合測試套件程式碼
# (與上面的測試內容結合)
# 測試金字塔概念
def test_pyramid_concept():
# 單元測試:多且快
unit_tests = 100
integration_tests = 20
e2e_tests = 5
assert unit_tests > integration_tests > e2e_tests
今天我們完成了完整的整合測試實戰:
明天是我們 30 天旅程的最後一天!我們將:
記住:好的整合測試就像是系統的守護者,它確保所有組件能夠和諧地協同工作!