测试数据管理与驱动策略
大约 8 分钟
测试数据管理与驱动策略
前言：测试数据，自动化测试的"粮草"
俗话说"兵马未动，粮草先行"，在接口自动化测试中，测试数据就是我们的"粮草"。我见过太多项目因为测试数据管理混乱而导致自动化测试维护困难、执行不稳定的情况。
还记得我刚开始做自动化测试时，测试数据都是硬编码在代码里的：
def test_user_login():
response = requests.post('/login', json={
'username': 'testuser123', # 这个用户存在吗？
'password': 'password123' # 密码对吗？
})
assert response.status_code == 200结果就是：今天测试通过，明天就失败了，因为测试数据被其他同事修改或删除了。经过多年的踩坑和总结，我形成了一套完整的测试数据管理策略，今天就来和大家分享。
测试数据的分类与特点
1. 按数据来源分类
静态数据：预先准备好的固定数据
- 优点：稳定可靠，执行结果可预期
- 缺点：容易冲突，维护成本高
- 适用场景：基础配置数据、字典数据
动态数据：运行时生成的数据
- 优点：避免冲突，数据新鲜
- 缺点：不可预期，调试困难
- 适用场景：用户数据、订单数据
混合数据：静态模板 + 动态生成
- 优点：兼具稳定性和灵活性
- 缺点：实现复杂度较高
- 适用场景：大部分业务场景
2. 按数据生命周期分类
# 会话级数据：整个测试会话期间有效
@pytest.fixture(scope="session")
def admin_user():
return create_admin_user()
# 模块级数据：单个测试模块期间有效
@pytest.fixture(scope="module")
def test_company():
return create_test_company()
# 函数级数据：单个测试函数期间有效
@pytest.fixture(scope="function")
def temp_user():
user = create_temp_user()
yield user
delete_user(user.id) # 测试完成后清理数据驱动测试策略
1. 基于文件的数据驱动
YAML格式（推荐）：
# test_data/user_login.yaml
test_cases:
- name: "正常登录"
input:
username: "admin"
password: "123456"
expected:
status_code: 200
message: "登录成功"
- name: "用户名错误"
input:
username: "wronguser"
password: "123456"
expected:
status_code: 401
message: "用户名或密码错误"
- name: "密码错误"
input:
username: "admin"
password: "wrongpass"
expected:
status_code: 401
message: "用户名或密码错误"对应的测试代码：
import yaml
import pytest
def load_test_data(file_path):
"""加载测试数据"""
with open(file_path, 'r', encoding='utf-8') as f:
return yaml.safe_load(f)
class TestUserLogin:
@pytest.mark.parametrize("test_case", load_test_data("test_data/user_login.yaml")["test_cases"])
def test_user_login(self, api_client, test_case):
"""用户登录测试"""
# 发送请求
response = api_client.post('/login', json=test_case['input'])
# 验证结果
assert response.status_code == test_case['expected']['status_code']
assert test_case['expected']['message'] in response.json()['message']JSON格式：
{
"user_registration": [
{
"description": "正常注册",
"data": {
"username": "newuser",
"email": "newuser@example.com",
"password": "StrongPass123!"
},
"expected": {
"status_code": 201,
"success": true
}
},
{
"description": "邮箱格式错误",
"data": {
"username": "newuser2",
"email": "invalid-email",
"password": "StrongPass123!"
},
"expected": {
"status_code": 400,
"error_code": "INVALID_EMAIL"
}
}
]
}CSV格式（适合简单数据）：
username,password,expected_status,expected_message
admin,123456,200,登录成功
user,password,200,登录成功
"",123456,400,用户名不能为空
admin,"",400,密码不能为空
wronguser,123456,401,用户名或密码错误import csv
import pytest
def load_csv_data(file_path):
"""加载CSV测试数据"""
test_data = []
with open(file_path, 'r', encoding='utf-8') as f:
reader = csv.DictReader(f)
for row in reader:
test_data.append(row)
return test_data
@pytest.mark.parametrize("test_data", load_csv_data("test_data/login_cases.csv"))
def test_login_with_csv(api_client, test_data):
"""使用CSV数据进行登录测试"""
response = api_client.post('/login', json={
'username': test_data['username'],
'password': test_data['password']
})
assert response.status_code == int(test_data['expected_status'])
assert test_data['expected_message'] in response.json()['message']2. 基于数据库的数据驱动
import sqlite3
import pytest
class TestDataManager:
"""测试数据管理器"""
def __init__(self, db_path="test_data.db"):
self.db_path = db_path
self.init_database()
def init_database(self):
"""初始化数据库"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS test_cases (
id INTEGER PRIMARY KEY,
test_suite TEXT,
case_name TEXT,
input_data TEXT,
expected_data TEXT,
enabled BOOLEAN DEFAULT 1
)
''')
conn.commit()
conn.close()
def get_test_cases(self, test_suite):
"""获取测试用例数据"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute('''
SELECT case_name, input_data, expected_data
FROM test_cases
WHERE test_suite = ? AND enabled = 1
''', (test_suite,))
cases = []
for row in cursor.fetchall():
cases.append({
'name': row[0],
'input': json.loads(row[1]),
'expected': json.loads(row[2])
})
conn.close()
return cases
# 使用示例
data_manager = TestDataManager()
@pytest.mark.parametrize("test_case", data_manager.get_test_cases("user_login"))
def test_login_from_db(api_client, test_case):
"""从数据库加载测试数据"""
response = api_client.post('/login', json=test_case['input'])
assert response.status_code == test_case['expected']['status_code']动态数据生成策略
1. 使用Faker生成随机数据
from faker import Faker
import random
fake = Faker('zh_CN') # 中文数据
class DataFactory:
"""数据工厂类"""
@staticmethod
def create_user_data():
"""生成用户数据"""
return {
'username': fake.user_name(),
'email': fake.email(),
'phone': fake.phone_number(),
'name': fake.name(),
'address': fake.address(),
'birthday': fake.date_of_birth(minimum_age=18, maximum_age=65).isoformat(),
'company': fake.company()
}
@staticmethod
def create_product_data():
"""生成商品数据"""
return {
'name': fake.catch_phrase(),
'description': fake.text(max_nb_chars=200),
'price': round(random.uniform(10, 1000), 2),
'category': random.choice(['电子产品', '服装', '食品', '图书', '家居']),
'stock': random.randint(0, 1000),
'sku': fake.ean13()
}
@staticmethod
def create_order_data(user_id=None, product_ids=None):
"""生成订单数据"""
return {
'user_id': user_id or random.randint(1, 1000),
'product_ids': product_ids or [random.randint(1, 100) for _ in range(random.randint(1, 5))],
'total_amount': round(random.uniform(50, 5000), 2),
'shipping_address': fake.address(),
'payment_method': random.choice(['支付宝', '微信支付', '银行卡', '现金']),
'remark': fake.sentence()
}
# 使用示例
@pytest.fixture
def random_user():
"""随机用户数据fixture"""
return DataFactory.create_user_data()
def test_create_user(api_client, random_user):
"""测试创建用户"""
response = api_client.post('/users', json=random_user)
assert response.status_code == 201
created_user = response.json()['data']
assert created_user['email'] == random_user['email']2. 模板化数据生成
import string
import random
from datetime import datetime, timedelta
class TemplateDataGenerator:
"""模板化数据生成器"""
def __init__(self):
self.templates = {
'email': '{username}@{domain}',
'phone': '1{area_code}{number}',
'id_card': '{area_code}{birth_date}{sequence}{check_digit}',
'order_no': 'ORD{timestamp}{random_suffix}'
}
def generate_email(self, username=None, domain=None):
"""生成邮箱地址"""
username = username or self._random_string(8)
domain = domain or random.choice(['example.com', 'test.com', 'demo.org'])
return f"{username}@{domain}"
def generate_phone(self, area_code=None):
"""生成手机号"""
area_code = area_code or random.choice(['138', '139', '150', '151', '188'])
number = ''.join([str(random.randint(0, 9)) for _ in range(8)])
return f"1{area_code}{number}"
def generate_order_no(self):
"""生成订单号"""
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 _random_string(self, length):
"""生成随机字符串"""
return ''.join(random.choices(string.ascii_lowercase, k=length))
# 使用示例
generator = TemplateDataGenerator()
def test_user_registration():
"""测试用户注册"""
user_data = {
'username': generator._random_string(8),
'email': generator.generate_email(),
'phone': generator.generate_phone(),
'password': 'TestPass123!'
}
response = api_client.post('/register', json=user_data)
assert response.status_code == 201测试数据隔离策略
1. 数据库事务隔离
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
@pytest.fixture(scope="function")
def db_transaction():
"""数据库事务fixture"""
engine = create_engine('sqlite:///test.db')
Session = sessionmaker(bind=engine)
session = Session()
# 开始事务
transaction = session.begin()
yield session
# 回滚事务，清理测试数据
transaction.rollback()
session.close()
def test_user_crud(db_transaction):
"""测试用户CRUD操作"""
# 创建用户
user = User(name='测试用户', email='test@example.com')
db_transaction.add(user)
db_transaction.flush() # 获取ID但不提交
# 验证创建
assert user.id is not None
# 更新用户
user.name = '更新后的用户'
db_transaction.flush()
# 验证更新
updated_user = db_transaction.query(User).filter_by(id=user.id).first()
assert updated_user.name == '更新后的用户'
# 测试结束后，事务会自动回滚，数据不会真正保存到数据库2. 命名空间隔离
import uuid
from datetime import datetime
class NamespaceManager:
"""命名空间管理器"""
def __init__(self):
self.namespace = self._generate_namespace()
def _generate_namespace(self):
"""生成唯一命名空间"""
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
unique_id = str(uuid.uuid4())[:8]
return f"test_{timestamp}_{unique_id}"
def get_unique_name(self, base_name):
"""获取唯一名称"""
return f"{self.namespace}_{base_name}"
def cleanup_namespace(self):
"""清理命名空间下的所有数据"""
# 删除所有以namespace开头的数据
pass
@pytest.fixture(scope="function")
def namespace():
"""命名空间fixture"""
ns_manager = NamespaceManager()
yield ns_manager
ns_manager.cleanup_namespace()
def test_create_company(api_client, namespace):
"""测试创建公司"""
company_data = {
'name': namespace.get_unique_name('测试公司'),
'code': namespace.get_unique_name('TEST_COMPANY'),
'description': '这是一个测试公司'
}
response = api_client.post('/companies', json=company_data)
assert response.status_code == 2013. 沙箱环境隔离
class SandboxManager:
"""沙箱环境管理器"""
def __init__(self, api_client):
self.api_client = api_client
self.created_resources = []
def create_user(self, user_data):
"""在沙箱中创建用户"""
response = self.api_client.post('/users', json=user_data)
if response.status_code == 201:
user_id = response.json()['data']['id']
self.created_resources.append(('user', user_id))
return response
def create_company(self, company_data):
"""在沙箱中创建公司"""
response = self.api_client.post('/companies', json=company_data)
if response.status_code == 201:
company_id = response.json()['data']['id']
self.created_resources.append(('company', company_id))
return response
def cleanup(self):
"""清理沙箱中的所有资源"""
for resource_type, resource_id in reversed(self.created_resources):
try:
if resource_type == 'user':
self.api_client.delete(f'/users/{resource_id}')
elif resource_type == 'company':
self.api_client.delete(f'/companies/{resource_id}')
except Exception as e:
print(f"清理资源失败: {resource_type}:{resource_id}, 错误: {e}")
@pytest.fixture
def sandbox(api_client):
"""沙箱fixture"""
sandbox_manager = SandboxManager(api_client)
yield sandbox_manager
sandbox_manager.cleanup()
def test_user_company_relationship(sandbox):
"""测试用户和公司关系"""
# 创建公司
company_data = {'name': '测试公司', 'code': 'TEST_COMPANY'}
company_response = sandbox.create_company(company_data)
company_id = company_response.json()['data']['id']
# 创建用户
user_data = {'name': '测试用户', 'company_id': company_id}
user_response = sandbox.create_user(user_data)
assert user_response.status_code == 201
assert user_response.json()['data']['company_id'] == company_id数据依赖管理
1. 依赖链构建
class DataDependencyManager:
"""数据依赖管理器"""
def __init__(self, api_client):
self.api_client = api_client
self.dependency_cache = {}
def get_or_create_admin_user(self):
"""获取或创建管理员用户"""
if 'admin_user' not in self.dependency_cache:
user_data = {
'username': 'admin',
'password': 'admin123',
'role': 'admin'
}
response = self.api_client.post('/users', json=user_data)
self.dependency_cache['admin_user'] = response.json()['data']
return self.dependency_cache['admin_user']
def get_or_create_test_company(self):
"""获取或创建测试公司"""
if 'test_company' not in self.dependency_cache:
admin_user = self.get_or_create_admin_user()
company_data = {
'name': '测试公司',
'owner_id': admin_user['id']
}
response = self.api_client.post('/companies', json=company_data)
self.dependency_cache['test_company'] = response.json()['data']
return self.dependency_cache['test_company']
def create_employee(self, name):
"""创建员工（依赖公司）"""
company = self.get_or_create_test_company()
employee_data = {
'name': name,
'company_id': company['id'],
'department': '测试部门'
}
response = self.api_client.post('/employees', json=employee_data)
return response.json()['data']
@pytest.fixture(scope="session")
def dependency_manager(api_client):
"""依赖管理器fixture"""
return DataDependencyManager(api_client)
def test_employee_operations(dependency_manager):
"""测试员工操作"""
# 创建员工（会自动创建依赖的公司和管理员）
employee = dependency_manager.create_employee('张三')
assert employee['name'] == '张三'
assert employee['company_id'] is not None总结
测试数据管理是接口自动化测试中的重要环节，好的数据管理策略可以让测试更加稳定、可维护。
关键要点回顾：
- 数据分类：根据来源和生命周期合理分类
- 数据驱动：使用外部文件或数据库管理测试数据
- 动态生成：使用Faker等工具生成随机数据
- 数据隔离：避免测试之间的数据污染
- 依赖管理：处理复杂的数据依赖关系
下一篇文章，我们将学习如何美化和定制allure测试报告，敬请期待！