集成测试与端到端测试
集成测试与端到端测试
集成测试验证组件之间的协作，端到端测试验证整个系统流程。本文将介绍测试数据库、API测试和浏览器自动化测试。
测试数据库配置
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.models import Base, User, Product
@pytest.fixture(scope="session")
def engine():
"""创建测试数据库引擎"""
engine = create_engine("sqlite:///test.db")
Base.metadata.create_all(engine)
yield engine
Base.metadata.drop_all(engine)
engine.dispose()
@pytest.fixture(scope="function")
def db_session(engine):
"""每个测试函数使用独立会话"""
Session = sessionmaker(bind=engine)
session = Session()
yield session
session.rollback()
session.close()
@pytest.fixture(scope="session")
def test_data():
"""测试数据工厂"""
return {
"users": [
{"id": 1, "name": "Alice", "email": "alice@test.com"},
{"id": 2, "name": "Bob", "email": "bob@test.com"},
],
"products": [
{"id": 1, "name": "Laptop", "price": 999.99},
{"id": 2, "name": "Phone", "price": 599.99},
]
}
# 使用Docker进行集成测试
@pytest.fixture(scope="session")
def docker_postgres():
"""使用Docker运行PostgreSQL"""
import docker
client = docker.from_env()
container = client.containers.run(
"postgres:15",
environment={
"POSTGRES_DB": "testdb",
"POSTGRES_USER": "test",
"POSTGRES_PASSWORD": "test"
},
ports={"5432/tcp": 5432},
detach=True
)
# 等待数据库就绪
import time
time.sleep(5)
yield f"postgresql://test:test@localhost:5432/testdb"
container.stop()
container.remove()
API测试
import pytest
from fastapi.testclient import TestClient
from unittest.mock import Mock, patch
from app import app
from app.database import get_db
client = TestClient(app)
# 测试数据库依赖注入
def override_get_db():
db = Mock()
db.query.return_value.filter.return_value.first.return_value = Mock(
id=1, name="Test User"
)
return db
app.dependency_overrides[get_db] = override_get_db
class TestUserAPI:
"""用户API测试套件"""
def test_get_users(self):
"""测试获取用户列表"""
response = client.get("/api/users")
assert response.status_code == 200
assert isinstance(response.json(), list)
def test_create_user(self):
"""测试创建用户"""
user_data = {
"name": "Alice",
"email": "alice@example.com",
"password": "secure123"
}
response = client.post("/api/users", json=user_data)
assert response.status_code == 201
data = response.json()
assert data["name"] == "Alice"
assert "id" in data
def test_create_user_invalid_email(self):
"""测试无效邮箱"""
user_data = {
"name": "Bob",
"email": "invalid-email",
"password": "secure123"
}
response = client.post("/api/users", json=user_data)
assert response.status_code == 422
assert "email" in response.json()["detail"][0]["loc"]
def test_get_user_not_found(self):
"""测试用户不存在"""
response = client.get("/api/users/999")
assert response.status_code == 404
def test_update_user(self):
"""测试更新用户"""
update_data = {"name": "Updated Name"}
response = client.put("/api/users/1", json=update_data)
assert response.status_code == 200
assert response.json()["name"] == "Updated Name"
def test_delete_user(self):
"""测试删除用户"""
response = client.delete("/api/users/1")
assert response.status_code == 204
# 测试认证
class TestAuthentication:
"""认证测试"""
@pytest.fixture
def auth_token(self):
"""获取认证token"""
response = client.post("/auth/token", data={
"username": "testuser",
"password": "testpass"
})
return response.json()["access_token"]
def test_protected_endpoint(self, auth_token):
"""测试受保护端点"""
headers = {"Authorization": f"Bearer {auth_token}"}
response = client.get("/api/profile", headers=headers)
assert response.status_code == 200
def test_unauthorized_access(self):
"""测试未授权访问"""
response = client.get("/api/profile")
assert response.status_code == 401
测试异步API
import pytest
from httpx import AsyncClient
from app import app
@pytest.mark.asyncio
async def test_async_endpoint():
"""测试异步端点"""
async with AsyncClient(app=app, base_url="http://test") as client:
response = await client.get("/api/async")
assert response.status_code == 200
@pytest.mark.asyncio
async def test_websocket():
"""测试WebSocket"""
async with AsyncClient(app=app, base_url="http://test") as client:
async with client.websocket_connect("/ws") as websocket:
await websocket.send_text("Hello")
data = await websocket.receive_text()
assert data == "Echo: Hello"
Selenium浏览器自动化
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.chrome.options import Options
import pytest
@pytest.fixture(scope="module")
def browser():
"""配置浏览器驱动"""
options = Options()
options.add_argument("--headless") # 无头模式
options.add_argument("--no-sandbox")
options.add_argument("--disable-dev-shm-usage")
driver = webdriver.Chrome(options=options)
driver.implicitly_wait(10)
yield driver
driver.quit()
class TestWebApplication:
"""Web应用端到端测试"""
def test_homepage_loads(self, browser):
"""测试首页加载"""
browser.get("http://localhost:8000")
# 等待页面加载
WebDriverWait(browser, 10).until(
EC.presence_of_element_located((By.TAG_NAME, "h1"))
)
# 验证标题
title = browser.find_element(By.TAG_NAME, "h1")
assert title.text == "Welcome"
def test_user_login(self, browser):
"""测试用户登录"""
browser.get("http://localhost:8000/login")
# 填写表单
username = browser.find_element(By.NAME, "username")
password = browser.find_element(By.NAME, "password")
submit = browser.find_element(By.TYPE, "submit")
username.send_keys("testuser")
password.send_keys("testpass123")
submit.click()
# 等待登录成功
WebDriverWait(browser, 10).until(
EC.presence_of_element_located((By.CLASS_NAME, "dashboard"))
)
# 验证登录状态
welcome = browser.find_element(By.CLASS_NAME, "user-welcome")
assert "testuser" in welcome.text
def test_search_functionality(self, browser):
"""测试搜索功能"""
browser.get("http://localhost:8000")
search_box = browser.find_element(By.ID, "search-input")
search_button = browser.find_element(By.ID, "search-button")
search_box.send_keys("Python tutorial")
search_button.click()
# 等待搜索结果
WebDriverWait(browser, 10).until(
EC.presence_of_element_located((By.CLASS_NAME, "search-results"))
)
results = browser.find_elements(By.CLASS_NAME, "result-item")
assert len(results) > 0
# 页面对象模式
class LoginPage:
"""登录页面对象"""
def __init__(self, driver):
self.driver = driver
self.wait = WebDriverWait(driver, 10)
def navigate(self):
self.driver.get("http://localhost:8000/login")
def login(self, username, password):
username_field = self.driver.find_element(By.NAME, "username")
password_field = self.driver.find_element(By.NAME, "password")
submit_button = self.driver.find_element(By.TYPE, "submit")
username_field.send_keys(username)
password_field.send_keys(password)
submit_button.click()
return self.wait.until(
EC.presence_of_element_located((By.CLASS_NAME, "dashboard"))
)
def get_error_message(self):
return self.driver.find_element(By.CLASS_NAME, "error-message").text
# 使用页面对象
def test_login_with_page_object(browser):
login_page = LoginPage(browser)
login_page.navigate()
dashboard = login_page.login("testuser", "wrongpass")
assert login_page.get_error_message() == "Invalid credentials"
测试数据管理
import factory
from app.models import User, Product
# 使用factory_boy创建测试数据
class UserFactory(factory.Factory):
class Meta:
model = User
id = factory.Sequence(lambda n: n + 1)
name = factory.Faker("name")
email = factory.Faker("email")
password = factory.PostGenerationMethodCall("hash_password", "testpass")
class ProductFactory(factory.Factory):
class Meta:
model = Product
id = factory.Sequence(lambda n: n + 1)
name = factory.Faker("word")
price = factory.Faker("pydecimal", left_digits=4, right_digits=2, positive=True)
stock = factory.Faker("random_int", min=0, max=1000)
# 使用测试数据
@pytest.fixture
def sample_users():
return UserFactory.create_batch(5)
@pytest.fixture
def sample_products():
return ProductFactory.create_batch(10)
def test_user_product_association(sample_users, sample_products):
user = sample_users[0]
product = sample_products[0]
user.products.append(product)
assert product in user.products
测试性能
import time
import statistics
class TestPerformance:
"""性能测试"""
def test_api_response_time(self):
"""测试API响应时间"""
times = []
for _ in range(100):
start = time.time()
response = client.get("/api/users")
end = time.time()
assert response.status_code == 200
times.append(end - start)
avg_time = statistics.mean(times)
p95_time = sorted(times)[int(len(times) * 0.95)]
assert avg_time < 0.1 # 平均响应时间<100ms
assert p95_time < 0.2 # 95%请求<200ms
def test_database_query_performance(self, db_session):
"""测试数据库查询性能"""
# 插入大量数据
for i in range(10000):
user = User(name=f"User {i}", email=f"user{i}@test.com")
db_session.add(user)
db_session.commit()
start = time.time()
users = db_session.query(User).filter(User.name.like("User 1%")).all()
end = time.time()
assert len(users) > 0
assert end - start < 0.5 # 查询时间<500ms
测试报告和覆盖
# pytest-html报告
# pytest --html=report.html --self-contained-html
# 覆盖率报告
# pytest --cov=src --cov-report=html --cov-report=xml
# 使用pytest-benchmark
def test_benchmark_sorting(benchmark):
"""基准测试"""
data = list(range(10000))
benchmark(lambda: sorted(data))
最佳实践
- 测试隔离：每个测试独立，无共享状态
- 测试数据管理：使用工厂模式创建测试数据
- 页面对象模式：Selenium测试使用POM提高可维护性
- 测试环境：使用Docker提供一致的测试环境
- 测试报告：生成详细的测试报告和覆盖率报告
总结
集成测试和端到端测试确保系统整体功能正确。通过测试数据库、API测试和浏览器自动化，可以验证从数据库到前端的完整流程。