数据驱动测试与Mock服务
大约 16 分钟
数据驱动测试与Mock服务
前言：让测试数据"活"起来
还记得我刚开始做自动化测试时，每个测试用例都要硬编码一堆测试数据，就像这样：
def test_user_registration():
user_data = {
"username": "testuser123", # 这个用户名会不会重复？
"email": "test@example.com", # 这个邮箱存在吗？
"phone": "13800138000" # 这个手机号被占用了吗？
}
# ...结果就是今天测试通过，明天就失败了，因为数据冲突了。后来我意识到，好的测试数据就像活水，应该是流动的、新鲜的、不会产生冲突的。
今天我们就来探讨这个pytest框架中的数据驱动测试和Mock服务实现，看看如何让测试数据"活"起来，让测试更加稳定和独立。
数据驱动测试：让数据成为测试的"燃料"
设计理念
"""
数据驱动测试的设计理念
就像汽车需要汽油一样，测试需要数据作为"燃料"
"""
data_driven_benefits = {
"数据与逻辑分离": "测试逻辑专注于验证，数据管理独立进行",
"提高测试覆盖": "同一个测试逻辑可以验证多组数据",
"便于维护": "修改测试数据不需要改动测试代码",
"支持大规模测试": "可以轻松生成大量测试数据",
"真实性": "使用Faker生成接近真实的测试数据"
}1. 数据生成器设计
# src/utils/data_driver.py
"""
数据驱动测试模块 - 测试数据的"工厂"
让测试数据生成变得简单而强大
"""
from faker import Faker
import random
import string
import uuid
from datetime import datetime, timedelta
from typing import Dict, List, Any, Optional, Callable
import yaml
import json
class DataGenerator:
"""数据生成器 - 测试数据的魔法师"""
def __init__(self, locale: str = 'zh_CN'):
self.fake = Faker(locale)
self.fake.seed_instance(random.randint(1, 10000)) # 随机种子
def generate_user_data(self, **overrides) -> Dict[str, Any]:
"""生成用户数据"""
base_data = {
"username": self._generate_unique_username(),
"email": self.fake.email(),
"phone": self._generate_phone(),
"name": self.fake.name(),
"age": random.randint(18, 65),
"gender": random.choice(['male', 'female']),
"address": {
"province": self.fake.province(),
"city": self.fake.city(),
"district": self.fake.district(),
"street": self.fake.street_address()
},
"created_at": self.fake.date_time_between(start_date='-1y', end_date='now').isoformat()
}
# 应用覆盖参数
base_data.update(overrides)
return base_data
def generate_product_data(self, **overrides) -> Dict[str, Any]:
"""生成商品数据"""
categories = ['电子产品', '服装', '食品', '图书', '家居', '运动', '美妆']
base_data = {
"name": self.fake.catch_phrase(),
"description": self.fake.text(max_nb_chars=200),
"price": round(random.uniform(10, 1000), 2),
"category": random.choice(categories),
"stock": random.randint(0, 1000),
"sku": self._generate_sku(),
"brand": self.fake.company(),
"weight": round(random.uniform(0.1, 10.0), 2),
"dimensions": {
"length": round(random.uniform(1, 100), 1),
"width": round(random.uniform(1, 100), 1),
"height": round(random.uniform(1, 100), 1)
},
"tags": [self.fake.word() for _ in range(random.randint(1, 5))]
}
base_data.update(overrides)
return base_data
def generate_order_data(self, user_id: int = None, product_ids: List[int] = None, **overrides) -> Dict[str, Any]:
"""生成订单数据"""
if user_id is None:
user_id = random.randint(1, 1000)
if product_ids is None:
product_ids = [random.randint(1, 100) for _ in range(random.randint(1, 5))]
items = []
total_amount = 0
for product_id in product_ids:
quantity = random.randint(1, 3)
price = round(random.uniform(10, 500), 2)
items.append({
"product_id": product_id,
"quantity": quantity,
"price": price,
"subtotal": round(price * quantity, 2)
})
total_amount += items[-1]["subtotal"]
base_data = {
"order_no": self._generate_order_no(),
"user_id": user_id,
"items": items,
"total_amount": round(total_amount, 2),
"status": random.choice(['pending', 'paid', 'shipped', 'delivered', 'cancelled']),
"payment_method": random.choice(['alipay', 'wechat', 'credit_card', 'cash']),
"shipping_address": {
"recipient": self.fake.name(),
"phone": self._generate_phone(),
"address": self.fake.address()
},
"created_at": self.fake.date_time_between(start_date='-30d', end_date='now').isoformat(),
"remark": self.fake.sentence()
}
base_data.update(overrides)
return base_data
def _generate_unique_username(self) -> str:
"""生成唯一用户名"""
timestamp = int(datetime.now().timestamp())
random_suffix = ''.join(random.choices(string.ascii_lowercase, k=4))
return f"user_{timestamp}_{random_suffix}"
def _generate_phone(self) -> str:
"""生成手机号"""
prefixes = ['130', '131', '132', '133', '134', '135', '136', '137', '138', '139',
'150', '151', '152', '153', '155', '156', '157', '158', '159',
'180', '181', '182', '183', '184', '185', '186', '187', '188', '189']
prefix = random.choice(prefixes)
suffix = ''.join([str(random.randint(0, 9)) for _ in range(8)])
return f"{prefix}{suffix}"
def _generate_sku(self) -> str:
"""生成SKU编码"""
letters = ''.join(random.choices(string.ascii_uppercase, k=3))
numbers = ''.join(random.choices(string.digits, k=6))
return f"{letters}{numbers}"
def _generate_order_no(self) -> str:
"""生成订单号"""
timestamp = datetime.now().strftime('%Y%m%d%H%M%S')
random_suffix = ''.join(random.choices(string.digits, k=4))
return f"ORD{timestamp}{random_suffix}"
def generate_batch_data(self, data_type: str, count: int, **kwargs) -> List[Dict[str, Any]]:
"""批量生成数据"""
generators = {
'user': self.generate_user_data,
'product': self.generate_product_data,
'order': self.generate_order_data
}
if data_type not in generators:
raise ValueError(f"不支持的数据类型: {data_type}")
generator = generators[data_type]
return [generator(**kwargs) for _ in range(count)]
class DataTemplate:
"""数据模板管理器"""
def __init__(self, template_dir: str = "data_templates"):
self.template_dir = template_dir
self.templates = {}
self._load_templates()
def _load_templates(self):
"""加载数据模板"""
import os
from pathlib import Path
template_path = Path(self.template_dir)
if not template_path.exists():
return
for file_path in template_path.glob("*.yaml"):
with open(file_path, 'r', encoding='utf-8') as f:
template_name = file_path.stem
self.templates[template_name] = yaml.safe_load(f)
def get_template(self, template_name: str) -> Dict[str, Any]:
"""获取数据模板"""
if template_name not in self.templates:
raise ValueError(f"模板不存在: {template_name}")
return self.templates[template_name].copy()
def fill_template(self, template_name: str, **variables) -> Dict[str, Any]:
"""填充模板变量"""
template = self.get_template(template_name)
return self._replace_variables(template, variables)
def _replace_variables(self, data: Any, variables: Dict[str, Any]) -> Any:
"""递归替换模板变量"""
if isinstance(data, dict):
return {k: self._replace_variables(v, variables) for k, v in data.items()}
elif isinstance(data, list):
return [self._replace_variables(item, variables) for item in data]
elif isinstance(data, str) and data.startswith('${') and data.endswith('}'):
var_name = data[2:-1]
return variables.get(var_name, data)
else:
return data
# 数据模板示例文件
"""
# data_templates/user_registration.yaml
username: "${username}"
email: "${email}"
password: "${password}"
profile:
name: "${name}"
age: ${age}
phone: "${phone}"
address:
province: "${province}"
city: "${city}"
# data_templates/api_request.yaml
method: "POST"
url: "/api/users"
headers:
Content-Type: "application/json"
Authorization: "Bearer ${token}"
data:
username: "${username}"
email: "${email}"
expected:
status_code: 201
response:
code: 0
message: "success"
"""
class DataProvider:
"""数据提供者 - 统一的数据获取接口"""
def __init__(self):
self.generator = DataGenerator()
self.template = DataTemplate()
self._data_cache = {}
def get_user_data(self, template_name: str = None, **kwargs) -> Dict[str, Any]:
"""获取用户数据"""
if template_name:
# 使用模板生成数据
template_data = self.template.get_template(template_name)
# 用生成的数据填充模板变量
generated_data = self.generator.generate_user_data(**kwargs)
return self.template.fill_template(template_name, **generated_data)
else:
# 直接生成数据
return self.generator.generate_user_data(**kwargs)
def get_test_dataset(self, dataset_name: str) -> List[Dict[str, Any]]:
"""获取测试数据集"""
if dataset_name in self._data_cache:
return self._data_cache[dataset_name]
# 根据数据集名称生成相应的数据
if dataset_name == "user_registration_cases":
dataset = self._generate_user_registration_cases()
elif dataset_name == "product_search_cases":
dataset = self._generate_product_search_cases()
elif dataset_name == "order_creation_cases":
dataset = self._generate_order_creation_cases()
else:
raise ValueError(f"未知的数据集: {dataset_name}")
self._data_cache[dataset_name] = dataset
return dataset
def _generate_user_registration_cases(self) -> List[Dict[str, Any]]:
"""生成用户注册测试用例数据"""
cases = []
# 正常注册案例
for _ in range(5):
cases.append({
"case_name": "正常注册",
"data": self.generator.generate_user_data(),
"expected": {"status_code": 201, "success": True}
})
# 异常案例
invalid_cases = [
{
"case_name": "用户名为空",
"data": self.generator.generate_user_data(username=""),
"expected": {"status_code": 400, "error_code": "INVALID_USERNAME"}
},
{
"case_name": "邮箱格式错误",
"data": self.generator.generate_user_data(email="invalid-email"),
"expected": {"status_code": 400, "error_code": "INVALID_EMAIL"}
},
{
"case_name": "手机号格式错误",
"data": self.generator.generate_user_data(phone="123"),
"expected": {"status_code": 400, "error_code": "INVALID_PHONE"}
}
]
cases.extend(invalid_cases)
return cases
def _generate_product_search_cases(self) -> List[Dict[str, Any]]:
"""生成商品搜索测试用例数据"""
return [
{
"case_name": "按名称搜索",
"params": {"keyword": "手机", "page": 1, "size": 10},
"expected": {"status_code": 200, "has_results": True}
},
{
"case_name": "按分类搜索",
"params": {"category": "电子产品", "page": 1, "size": 20},
"expected": {"status_code": 200, "has_results": True}
},
{
"case_name": "价格范围搜索",
"params": {"min_price": 100, "max_price": 500, "page": 1, "size": 10},
"expected": {"status_code": 200, "has_results": True}
},
{
"case_name": "无结果搜索",
"params": {"keyword": "不存在的商品名称12345", "page": 1, "size": 10},
"expected": {"status_code": 200, "has_results": False}
}
]
def _generate_order_creation_cases(self) -> List[Dict[str, Any]]:
"""生成订单创建测试用例数据"""
cases = []
# 正常订单案例
for _ in range(3):
order_data = self.generator.generate_order_data()
cases.append({
"case_name": "正常创建订单",
"data": order_data,
"expected": {"status_code": 201, "success": True}
})
# 异常案例
invalid_cases = [
{
"case_name": "用户ID不存在",
"data": self.generator.generate_order_data(user_id=999999),
"expected": {"status_code": 404, "error_code": "USER_NOT_FOUND"}
},
{
"case_name": "商品库存不足",
"data": self.generator.generate_order_data(),
"expected": {"status_code": 400, "error_code": "INSUFFICIENT_STOCK"}
}
]
cases.extend(invalid_cases)
return cases
# 全局数据提供者实例
data_provider = DataProvider()Mock服务：测试的"替身演员"
Mock服务设计理念
"""
Mock服务的设计理念
就像电影中的替身演员，在真正的演员不在场时提供相同的表演
"""
mock_service_benefits = {
"独立性": "不依赖外部服务，测试更稳定",
"可控性": "可以模拟各种场景，包括异常情况",
"速度": "本地响应，执行速度快",
"成本": "不消耗外部服务资源",
"隔离性": "避免测试数据污染生产环境"
}1. Mock服务器实现
# src/utils/mock_server.py
"""
Mock服务器 - 测试的"替身演员"
提供可控的、稳定的API响应
"""
import json
import threading
import time
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import urlparse, parse_qs
from typing import Dict, Any, List, Callable, Optional
import re
class MockRule:
"""Mock规则定义"""
def __init__(self, method: str, path_pattern: str, response_data: Dict[str, Any],
status_code: int = 200, delay: float = 0, headers: Dict[str, str] = None):
self.method = method.upper()
self.path_pattern = re.compile(path_pattern)
self.response_data = response_data
self.status_code = status_code
self.delay = delay
self.headers = headers or {}
self.call_count = 0
self.request_history = []
def matches(self, method: str, path: str) -> bool:
"""检查请求是否匹配此规则"""
return (self.method == method.upper() and
self.path_pattern.match(path) is not None)
def get_response(self, request_data: Dict[str, Any] = None) -> Dict[str, Any]:
"""获取响应数据"""
self.call_count += 1
if request_data:
self.request_history.append(request_data)
# 模拟延迟
if self.delay > 0:
time.sleep(self.delay)
# 支持动态响应
if callable(self.response_data):
return self.response_data(request_data)
return self.response_data.copy()
class MockRequestHandler(BaseHTTPRequestHandler):
"""Mock请求处理器"""
def do_GET(self):
self._handle_request('GET')
def do_POST(self):
self._handle_request('POST')
def do_PUT(self):
self._handle_request('PUT')
def do_DELETE(self):
self._handle_request('DELETE')
def do_PATCH(self):
self._handle_request('PATCH')
def _handle_request(self, method: str):
"""处理HTTP请求"""
try:
# 解析请求
parsed_url = urlparse(self.path)
path = parsed_url.path
query_params = parse_qs(parsed_url.query)
# 读取请求体
content_length = int(self.headers.get('Content-Length', 0))
request_body = self.rfile.read(content_length).decode('utf-8') if content_length > 0 else ''
request_data = {
'method': method,
'path': path,
'query_params': query_params,
'headers': dict(self.headers),
'body': request_body
}
# 查找匹配的规则
mock_rule = self.server.find_matching_rule(method, path)
if mock_rule:
response_data = mock_rule.get_response(request_data)
self._send_response(mock_rule.status_code, response_data, mock_rule.headers)
else:
# 没有匹配的规则，返回404
self._send_error_response(404, "Mock rule not found")
except Exception as e:
self._send_error_response(500, f"Mock server error: {str(e)}")
def _send_response(self, status_code: int, data: Dict[str, Any], headers: Dict[str, str]):
"""发送响应"""
self.send_response(status_code)
# 设置默认头部
self.send_header('Content-Type', 'application/json')
self.send_header('Access-Control-Allow-Origin', '*')
self.send_header('Access-Control-Allow-Methods', 'GET, POST, PUT, DELETE, PATCH, OPTIONS')
self.send_header('Access-Control-Allow-Headers', 'Content-Type, Authorization')
# 设置自定义头部
for key, value in headers.items():
self.send_header(key, value)
self.end_headers()
# 发送响应体
response_json = json.dumps(data, ensure_ascii=False, indent=2)
self.wfile.write(response_json.encode('utf-8'))
def _send_error_response(self, status_code: int, message: str):
"""发送错误响应"""
error_data = {
"error": {
"code": status_code,
"message": message
}
}
self._send_response(status_code, error_data, {})
def log_message(self, format, *args):
"""重写日志方法，避免控制台输出"""
pass
class MockServer:
"""Mock服务器"""
def __init__(self, host: str = 'localhost', port: int = 8888):
self.host = host
self.port = port
self.rules: List[MockRule] = []
self.server: Optional[HTTPServer] = None
self.server_thread: Optional[threading.Thread] = None
self.is_running = False
def add_rule(self, method: str, path_pattern: str, response_data: Dict[str, Any],
status_code: int = 200, delay: float = 0, headers: Dict[str, str] = None) -> MockRule:
"""添加Mock规则"""
rule = MockRule(method, path_pattern, response_data, status_code, delay, headers)
self.rules.append(rule)
return rule
def find_matching_rule(self, method: str, path: str) -> Optional[MockRule]:
"""查找匹配的规则"""
for rule in self.rules:
if rule.matches(method, path):
return rule
return None
def start(self):
"""启动Mock服务器"""
if self.is_running:
return
self.server = HTTPServer((self.host, self.port), MockRequestHandler)
self.server.find_matching_rule = self.find_matching_rule
self.server_thread = threading.Thread(target=self.server.serve_forever)
self.server_thread.daemon = True
self.server_thread.start()
self.is_running = True
print(f"🚀 Mock服务器启动成功: http://{self.host}:{self.port}")
def stop(self):
"""停止Mock服务器"""
if not self.is_running:
return
if self.server:
self.server.shutdown()
self.server.server_close()
if self.server_thread:
self.server_thread.join(timeout=5)
self.is_running = False
print("🛑 Mock服务器已停止")
def clear_rules(self):
"""清空所有规则"""
self.rules.clear()
def get_rule_stats(self) -> List[Dict[str, Any]]:
"""获取规则统计信息"""
stats = []
for i, rule in enumerate(self.rules):
stats.append({
"index": i,
"method": rule.method,
"pattern": rule.path_pattern.pattern,
"call_count": rule.call_count,
"request_history": rule.request_history
})
return stats
# Mock数据生成器
class MockDataGenerator:
"""Mock数据生成器"""
@staticmethod
def success_response(data: Any = None, message: str = "success") -> Dict[str, Any]:
"""生成成功响应"""
return {
"code": 0,
"message": message,
"data": data,
"timestamp": int(time.time())
}
@staticmethod
def error_response(error_code: str, message: str, status_code: int = 400) -> Dict[str, Any]:
"""生成错误响应"""
return {
"code": status_code,
"message": message,
"error": {
"code": error_code,
"message": message
},
"timestamp": int(time.time())
}
@staticmethod
def pagination_response(items: List[Any], total_count: int, page: int = 1, size: int = 10) -> Dict[str, Any]:
"""生成分页响应"""
return MockDataGenerator.success_response({
"items": items,
"pagination": {
"current_page": page,
"page_size": size,
"total_count": total_count,
"total_pages": (total_count + size - 1) // size
}
})
# Mock规则预设
class MockRulePresets:
"""Mock规则预设"""
@staticmethod
def user_api_rules(mock_server: MockServer, data_generator: DataGenerator):
"""用户API的Mock规则"""
# 用户注册
def register_response(request_data):
try:
request_body = json.loads(request_data['body'])
user_data = data_generator.generate_user_data(**request_body)
user_data['id'] = random.randint(1000, 9999)
return MockDataGenerator.success_response(user_data, "注册成功")
except:
return MockDataGenerator.error_response("INVALID_DATA", "请求数据格式错误")
mock_server.add_rule('POST', r'/api/users$', register_response, 201)
# 获取用户信息
def get_user_response(request_data):
user_id = request_data['path'].split('/')[-1]
if user_id.isdigit():
user_data = data_generator.generate_user_data(id=int(user_id))
return MockDataGenerator.success_response(user_data)
else:
return MockDataGenerator.error_response("USER_NOT_FOUND", "用户不存在", 404)
mock_server.add_rule('GET', r'/api/users/\d+$', get_user_response)
# 用户列表
def list_users_response(request_data):
query_params = request_data['query_params']
page = int(query_params.get('page', [1])[0])
size = int(query_params.get('size', [10])[0])
users = data_generator.generate_batch_data('user', size)
return MockDataGenerator.pagination_response(users, 100, page, size)
mock_server.add_rule('GET', r'/api/users$', list_users_response)
@staticmethod
def product_api_rules(mock_server: MockServer, data_generator: DataGenerator):
"""商品API的Mock规则"""
# 商品搜索
def search_products_response(request_data):
query_params = request_data['query_params']
keyword = query_params.get('keyword', [''])[0]
page = int(query_params.get('page', [1])[0])
size = int(query_params.get('size', [10])[0])
if keyword == "不存在的商品名称12345":
# 模拟无结果
return MockDataGenerator.pagination_response([], 0, page, size)
else:
# 模拟有结果
products = data_generator.generate_batch_data('product', size)
return MockDataGenerator.pagination_response(products, 50, page, size)
mock_server.add_rule('GET', r'/api/products/search$', search_products_response)
@staticmethod
def order_api_rules(mock_server: MockServer, data_generator: DataGenerator):
"""订单API的Mock规则"""
# 创建订单
def create_order_response(request_data):
try:
request_body = json.loads(request_data['body'])
# 模拟用户不存在的情况
if request_body.get('user_id') == 999999:
return MockDataGenerator.error_response("USER_NOT_FOUND", "用户不存在", 404)
order_data = data_generator.generate_order_data(**request_body)
order_data['id'] = random.randint(10000, 99999)
return MockDataGenerator.success_response(order_data, "订单创建成功")
except:
return MockDataGenerator.error_response("INVALID_DATA", "请求数据格式错误")
mock_server.add_rule('POST', r'/api/orders$', create_order_response, 201)实战应用：数据驱动 + Mock服务
1. 数据驱动测试实战
# tests/test_data_driven.py
"""
数据驱动测试实战示例
展示如何使用数据驱动进行全面的接口测试
"""
import pytest
from src.client.base_client import BaseClient
from src.utils.assertion import assert_success_response, assert_error_response
from src.utils.data_driver import data_provider
class TestDataDrivenAPI:
"""数据驱动API测试"""
def setup_method(self):
self.client = BaseClient("http://localhost:8888") # Mock服务器地址
@pytest.mark.parametrize("test_case", data_provider.get_test_dataset("user_registration_cases"))
def test_user_registration(self, test_case):
"""用户注册数据驱动测试"""
case_name = test_case["case_name"]
user_data = test_case["data"]
expected = test_case["expected"]
print(f"🧪 执行测试用例: {case_name}")
# 发送注册请求
response = self.client.post("/api/users", json=user_data)
# 根据期望结果进行断言
if expected["status_code"] == 201:
(assert_success_response(response)
.assert_status_code(201)
.assert_json_path("code", 0)
.assert_json_path("message", "注册成功")
.assert_json_path_exists("data.id")
.assert_json_path("data.username", user_data["username"])
.validate())
else:
(assert_error_response(response, expected["status_code"])
.assert_json_path("error.code", expected["error_code"])
.validate())
@pytest.mark.parametrize("test_case", data_provider.get_test_dataset("product_search_cases"))
def test_product_search(self, test_case):
"""商品搜索数据驱动测试"""
case_name = test_case["case_name"]
params = test_case["params"]
expected = test_case["expected"]
print(f"🔍 执行搜索测试: {case_name}")
# 发送搜索请求
response = self.client.get("/api/products/search", params=params)
# 验证响应
assertion = assert_success_response(response)
if expected["has_results"]:
assertion.assert_json_path_custom(
"data.items",
lambda items: len(items) > 0,
"应该有搜索结果"
)
else:
assertion.assert_json_path("data.items", [])
assertion.validate()
def test_batch_user_creation(self):
"""批量用户创建测试"""
# 生成批量测试数据
users_data = data_provider.generator.generate_batch_data('user', 5)
created_users = []
for i, user_data in enumerate(users_data):
print(f"👤 创建第 {i+1} 个用户: {user_data['username']}")
response = self.client.post("/api/users", json=user_data)
(assert_success_response(response)
.assert_status_code(201)
.assert_json_path_exists("data.id")
.validate())
created_users.append(response.json()["data"])
# 验证所有用户都创建成功
assert len(created_users) == 5
# 验证用户名唯一性
usernames = [user["username"] for user in created_users]
assert len(set(usernames)) == len(usernames), "用户名应该唯一"
class TestTemplateBasedTesting:
"""基于模板的测试"""
def setup_method(self):
self.client = BaseClient("http://localhost:8888")
def test_user_registration_with_template(self):
"""使用模板进行用户注册测试"""
# 使用模板生成测试数据
user_data = data_provider.get_user_data("user_registration")
response = self.client.post("/api/users", json=user_data)
(assert_success_response(response)
.assert_status_code(201)
.assert_json_path("data.username", user_data["username"])
.validate())
def test_api_request_with_template(self):
"""使用API请求模板进行测试"""
# 获取API请求模板
request_template = data_provider.template.get_template("api_request")
# 填充模板变量
test_data = data_provider.generator.generate_user_data()
filled_template = data_provider.template.fill_template("api_request", **test_data)
# 发送请求
response = self.client.request(
method=filled_template["method"],
endpoint=filled_template["url"],
headers=filled_template["headers"],
json=filled_template["data"]
)
# 验证响应
expected = filled_template["expected"]
(assert_success_response(response)
.assert_status_code(expected["status_code"])
.assert_json_path("code", expected["response"]["code"])
.validate())
# pytest fixture集成
@pytest.fixture(scope="session")
def mock_server():
"""Mock服务器fixture"""
from src.utils.mock_server import MockServer, MockRulePresets
from src.utils.data_driver import DataGenerator
# 启动Mock服务器
server = MockServer(host='localhost', port=8888)
generator = DataGenerator()
# 添加预设规则
MockRulePresets.user_api_rules(server, generator)
MockRulePresets.product_api_rules(server, generator)
MockRulePresets.order_api_rules(server, generator)
server.start()
yield server
# 清理
server.stop()
@pytest.fixture
def test_data():
"""测试数据fixture"""
return data_provider
# 使用示例
def test_with_fixtures(mock_server, test_data):
"""使用fixture的测试示例"""
client = BaseClient("http://localhost:8888")
# 使用测试数据
user_data = test_data.generator.generate_user_data()
response = client.post("/api/users", json=user_data)
assert response.status_code == 201
# 检查Mock服务器统计
stats = mock_server.get_rule_stats()
print("Mock服务器调用统计:", stats)2. 高级Mock场景
# tests/test_advanced_mock.py
"""
高级Mock场景测试
展示复杂的Mock服务使用场景
"""
import pytest
import time
from src.utils.mock_server import MockServer, MockDataGenerator
from src.client.base_client import BaseClient
class TestAdvancedMockScenarios:
"""高级Mock场景测试"""
def setup_method(self):
self.mock_server = MockServer(port=8889)
self.client = BaseClient("http://localhost:8889")
self.mock_server.start()
def teardown_method(self):
self.mock_server.stop()
def test_dynamic_response(self):
"""测试动态响应"""
call_count = 0
def dynamic_response(request_data):
nonlocal call_count
call_count += 1
if call_count == 1:
return MockDataGenerator.success_response({"message": "第一次调用"})
elif call_count == 2:
return MockDataGenerator.success_response({"message": "第二次调用"})
else:
return MockDataGenerator.error_response("TOO_MANY_CALLS", "调用次数过多", 429)
# 添加动态响应规则
self.mock_server.add_rule('GET', r'/api/dynamic$', dynamic_response)
# 第一次调用
response1 = self.client.get("/api/dynamic")
assert response1.status_code == 200
assert response1.json()["data"]["message"] == "第一次调用"
# 第二次调用
response2 = self.client.get("/api/dynamic")
assert response2.status_code == 200
assert response2.json()["data"]["message"] == "第二次调用"
# 第三次调用
response3 = self.client.get("/api/dynamic")
assert response3.status_code == 429
assert response3.json()["error"]["code"] == "TOO_MANY_CALLS"
def test_delayed_response(self):
"""测试延迟响应"""
# 添加延迟响应规则
self.mock_server.add_rule(
'GET',
r'/api/slow$',
MockDataGenerator.success_response({"message": "慢响应"}),
delay=2.0
)
start_time = time.time()
response = self.client.get("/api/slow")
end_time = time.time()
assert response.status_code == 200
assert end_time - start_time >= 2.0 # 验证延迟
def test_conditional_response(self):
"""测试条件响应"""
def conditional_response(request_data):
headers = request_data['headers']
auth_header = headers.get('Authorization', '')
if auth_header.startswith('Bearer valid-token'):
return MockDataGenerator.success_response({"message": "认证成功"})
else:
return MockDataGenerator.error_response("UNAUTHORIZED", "认证失败", 401)
self.mock_server.add_rule('GET', r'/api/protected$', conditional_response)
# 无认证头
response1 = self.client.get("/api/protected")
assert response1.status_code == 401
# 错误的认证头
response2 = self.client.get("/api/protected", headers={'Authorization': 'Bearer invalid-token'})
assert response2.status_code == 401
# 正确的认证头
response3 = self.client.get("/api/protected", headers={'Authorization': 'Bearer valid-token'})
assert response3.status_code == 200
def test_stateful_mock(self):
"""测试有状态的Mock"""
# 模拟一个简单的计数器服务
counter_state = {"value": 0}
def get_counter(request_data):
return MockDataGenerator.success_response({"counter": counter_state["value"]})
def increment_counter(request_data):
counter_state["value"] += 1
return MockDataGenerator.success_response({"counter": counter_state["value"]})
def reset_counter(request_data):
counter_state["value"] = 0
return MockDataGenerator.success_response({"counter": counter_state["value"]})
# 添加规则
self.mock_server.add_rule('GET', r'/api/counter$', get_counter)
self.mock_server.add_rule('POST', r'/api/counter/increment$', increment_counter)
self.mock_server.add_rule('POST', r'/api/counter/reset$', reset_counter)
# 测试计数器功能
# 初始值
response = self.client.get("/api/counter")
assert response.json()["data"]["counter"] == 0
# 增加
self.client.post("/api/counter/increment")
response = self.client.get("/api/counter")
assert response.json()["data"]["counter"] == 1
# 再次增加
self.client.post("/api/counter/increment")
response = self.client.get("/api/counter")
assert response.json()["data"]["counter"] == 2
# 重置
self.client.post("/api/counter/reset")
response = self.client.get("/api/counter")
assert response.json()["data"]["counter"] == 0
class TestMockServerManagement:
"""Mock服务器管理测试"""
def test_rule_statistics(self):
"""测试规则统计功能"""
mock_server = MockServer(port=8890)
mock_server.start()
try:
# 添加规则
rule1 = mock_server.add_rule('GET', r'/api/test1$', {"message": "test1"})
rule2 = mock_server.add_rule('POST', r'/api/test2$', {"message": "test2"})
client = BaseClient("http://localhost:8890")
# 调用API
client.get("/api/test1")
client.get("/api/test1")
client.post("/api/test2", json={})
# 检查统计信息
stats = mock_server.get_rule_stats()
assert len(stats) == 2
assert stats[0]["call_count"] == 2 # test1被调用2次
assert stats[1]["call_count"] == 1 # test2被调用1次
finally:
mock_server.stop()
def test_rule_management(self):
"""测试规则管理功能"""
mock_server = MockServer(port=8891)
# 添加规则
rule1 = mock_server.add_rule('GET', r'/api/test$', {"message": "original"})
assert len(mock_server.rules) == 1
# 清空规则
mock_server.clear_rules()
assert len(mock_server.rules) == 0
# 重新添加规则
rule2 = mock_server.add_rule('GET', r'/api/test$', {"message": "new"})
assert len(mock_server.rules) == 1最佳实践与经验总结
1. 数据管理策略
"""
数据管理最佳实践
"""
data_management_practices = {
"数据隔离": {
"原则": "每个测试使用独立的数据，避免相互影响",
"实现": "使用时间戳、UUID等生成唯一标识",
"示例": "username = f'user_{int(time.time())}_{uuid.uuid4().hex[:8]}'"
},
"数据清理": {
"原则": "测试完成后及时清理数据",
"实现": "使用pytest fixture的teardown机制",
"示例": "yield data; cleanup_data(data)"
},
"数据复用": {
"原则": "合理复用数据，提高测试效率",
"实现": "使用缓存机制，避免重复生成",
"示例": "session级别的fixture提供共享数据"
},
"数据真实性": {
"原则": "测试数据应该接近真实数据",
"实现": "使用Faker生成符合格式的数据",
"示例": "使用中文locale生成中文姓名和地址"
}
}2. Mock服务最佳实践
"""
Mock服务最佳实践
"""
mock_service_practices = {
"规则设计": {
"原则": "Mock规则应该简单明确",
"建议": "一个规则只处理一种场景",
"避免": "在一个规则中处理多种复杂逻辑"
},
"状态管理": {
"原则": "谨慎使用有状态的Mock",
"建议": "优先使用无状态的Mock",
"场景": "只在必要时使用状态，如测试工作流"
},
"错误模拟": {
"原则": "充分模拟各种错误场景",
"包括": "网络错误、服务器错误、业务错误",
"目的": "提高测试覆盖率和系统健壮性"
},
"性能考虑": {
"原则": "Mock服务应该快速响应",
"建议": "只在测试延迟场景时添加delay",
"优化": "使用编译后的正则表达式"
}
}总结
数据驱动测试和Mock服务就像测试的"双翼"，让测试既有丰富的数据支撑，又有稳定的环境保障。通过这篇文章，我们深入了解了：
- 数据生成策略：Faker集成、模板系统、批量生成
- Mock服务设计：规则管理、动态响应、状态控制
- 实战应用：数据驱动测试、高级Mock场景
- 最佳实践：数据管理、服务设计、性能优化
- 工程化实现：fixture集成、统计监控、规则管理
这套数据驱动和Mock服务系统不仅让测试更加稳定和可控，更重要的是提供了丰富的测试场景覆盖能力，让我们能够充分验证系统的各种边界情况。
下一篇文章，我们将探讨性能测试与并发优化，看看如何让测试框架在大规模场景下依然表现出色。
推荐阅读