一、pytest测试数据清理失效的常见坑
做接口或者业务逻辑测试的时候,很多人都遇到过这种情况:上一个测试用例跑的时候插了一条订单数据到数据库,结果跑完没清干净,下一个测试用例本来要测“查询空订单列表”,结果拿到了上一个用例留的脏数据,直接报错。更坑的是,有时候你明明写了清理代码,比如用finally块删数据,结果还是会漏——比如测试过程中抛出异常,清理代码没执行到;或者多个用例并行跑,你删的时候刚好另一个用例也在改同一条数据,删错了或者没删干净。
我之前踩过最离谱的坑是测电商的库存扣减逻辑,有个用例跑完没清掉扣减后的库存记录,导致下一个用例测“库存不足不能下单”的时候,系统认为库存够,用例直接失败。后来查了半天,发现是测试代码里的清理逻辑写在try块里,测试用例中途抛了个业务异常,清理代码根本没执行。
1.1 常见清理逻辑的缺陷
很多人一开始写清理逻辑,会用这几种方式,结果都踩了坑: 第一种是用finally块清理。比如测试完之后,在finally里写删除数据库的代码,但如果测试用例本身有多个分支,或者中间有其他异常打断,finally可能没覆盖到所有需要清理的场景?不对,其实finally是一定会执行的,但问题是如果清理逻辑本身也出错了——比如你要删的那条数据已经被其他并行用例删了,清理代码抛异常,反而会导致整个测试任务失败,还把原来的测试结果搞混。 第二种是用teardown方法。pytest里的teardown是测试类或者模块跑完之后执行的,但如果一个模块里有好几个用例,每个用例都改数据库,teardown的时候要清所有数据,很容易漏,比如某个用例插了数据没被记录到,teardown的时候就删不掉。 第三种是手动清库,比如每次跑测试前把数据库删了重建。但这种方法太耗时,尤其是测试用例多的时候,每次跑都要等数据库重建,效率极低,而且如果是共享的测试环境,你删库的时候别人的测试可能也在跑,直接影响别人。
二、fixture终结器:解决清理问题的核心思路
pytest的fixture是什么?简单说就是测试用例的“前置准备+后置清理”工具。比如你需要一个登录后的token,就可以写个fixture,先登录拿到token给用例用,用例跑完之后再把token注销掉。而终结器(teardown)就是fixture里专门负责后置清理的部分,和普通的清理逻辑不一样,fixture的终结器是pytest保证一定会执行的——不管用例是成功还是失败,不管有没有抛异常,只要fixture被用了,终结器就会跑。
更重要的是,fixture可以实现“用例级”的清理,也就是每个用例跑完之后,只清理这个用例自己产生的脏数据,不会影响其他用例。比如用例A插了一条订单,fixture的终结器就只删用例A插的那条,用例B插的订单不会被误删。
2.1 终结器的两种写法
fixture的终结器有两种常见写法,一种是用yield关键字,另一种是用request.addfinalizer方法。两种写法效果差不多,只是yield的写法更简洁,适合大部分场景。
先举个例子,用yield写的fixture结构大概是这样:
import pytest
@pytest.fixture
def my_fixture():
# 前置逻辑:比如插测试数据、初始化资源
print("前置:准备测试数据")
yield # 这里会暂停,把前置逻辑的结果传给测试用例
# 后置逻辑(终结器):比如删测试数据、释放资源
print("后置:清理测试数据")
当测试用例调用my_fixture的时候,会先执行yield之前的代码,然后把yield后面的值(如果有的话)传给用例,等用例跑完(不管成功失败),再执行yield之后的代码。
另一种用request.addfinalizer的写法,适合需要更灵活控制终结器的场景,比如同一个fixture里需要加多个终结器:
import pytest
@pytest.fixture
def my_fixture(request):
# 前置逻辑
print("前置:准备测试数据")
# 定义终结器函数
def finalizer():
print("后置:清理测试数据")
# 把终结器注册到fixture
request.addfinalizer(finalizer)
这种写法和yield的效果一样,都是保证终结器一定会执行。
三、用终结器实现数据库自动回滚:完整示例
接下来我们结合实际的数据库测试场景,写一个完整的示例,用fixture的终结器自动清理数据库的脏数据。这里我们用Python的pytest框架,搭配SQLAlchemy操作MySQL数据库,所有示例都用这个技术栈。
3.1 示例准备:技术栈说明
本次示例统一使用以下技术栈:
- 测试框架:pytest(用于编写和执行测试用例)
- 数据库操作:SQLAlchemy(Python的ORM框架,用于操作MySQL)
- 数据库:MySQL(测试用的数据库)
首先我们需要准备一个简单的测试用数据库表,比如一个订单表,结构如下:
CREATE TABLE orders (
id INT AUTO_INCREMENT PRIMARY KEY COMMENT '订单ID',
order_no VARCHAR(32) NOT NULL COMMENT '订单编号',
user_id INT NOT NULL COMMENT '用户ID',
amount DECIMAL(10,2) NOT NULL COMMENT '订单金额',
create_time DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间'
);
3.2 写fixture实现自动清理
我们先写一个fixture,负责插入一条测试订单数据,然后在终结器里删除这条数据,保证每个用例跑完之后,自己插的订单都会被删掉。
首先是数据库连接的配置,我们先写一个公共的数据库初始化fixture:
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, DECIMAL, DateTime
import datetime
# 配置数据库连接(请根据自己的测试环境修改)
DB_URL = "mysql+pymysql://test_user:test_password@localhost:3306/test_db?charset=utf8mb4"
# 初始化ORM的基类
Base = declarative_base()
# 定义订单模型,对应数据库的orders表
class Order(Base):
__tablename__ = "orders"
id = Column(Integer, primary_key=True, autoincrement=True, comment="订单ID")
order_no = Column(String(32), nullable=False, comment="订单编号")
user_id = Column(Integer, nullable=False, comment="用户ID")
amount = Column(DECIMAL(10,2), nullable=False, comment="订单金额")
create_time = Column(DateTime, default=datetime.datetime.now, comment="创建时间")
# 数据库连接的fixture,整个测试模块共享一个连接
@pytest.fixture(scope="module")
def db_session():
# 前置逻辑:创建数据库引擎和会话
engine = create_engine(DB_URL, echo=False) # echo=True会打印SQL语句,方便调试
Session = sessionmaker(bind=engine)
session = Session()
yield session # 把会话传给测试用例
# 后置逻辑:测试模块跑完后关闭会话
session.close()
engine.dispose()
接下来写一个针对单个测试用例的fixture,负责插入测试订单,然后自动删除:
# 针对单个用例的fixture,插入测试订单,自动清理
@pytest.fixture(scope="function") # scope="function"表示每个用例跑一次这个fixture
def test_order(db_session):
# 前置逻辑:插入一条测试订单
test_order = Order(
order_no="TEST_ORDER_001",
user_id=1001,
amount=99.99
)
db_session.add(test_order)
db_session.commit() # 提交到数据库,让其他查询能拿到这条数据
yield test_order # 把测试订单对象传给测试用例
# 后置逻辑(终结器):删除这条测试订单
db_session.delete(test_order)
db_session.commit()
3.3 写测试用例验证效果
现在我们写两个测试用例,第一个用例查询这条测试订单,第二个用例查询空订单列表,验证第二个用例不会拿到第一个用例的脏数据:
# 测试用例1:查询测试订单
def test_get_test_order(db_session, test_order):
# 用例逻辑:查询订单ID为test_order.id的订单
order = db_session.query(Order).filter(Order.id == test_order.id).first()
assert order is not None # 断言订单存在
assert order.order_no == "TEST_ORDER_001" # 断言订单编号正确
# 测试用例2:查询不存在的订单
def test_get_nonexistent_order(db_session):
# 用例逻辑:查询用户ID为9999的订单(测试环境里这个用户没有订单)
orders = db_session.query(Order).filter(Order.user_id == 9999).all()
assert len(orders) == 0 # 断言订单列表为空
如果我们没有用fixture的终结器,第一个用例跑完之后,测试订单会留在数据库里,第二个用例可能会拿到这条数据(如果测试环境里有其他脏数据的话),但现在有了终结器,第一个用例跑完之后,测试订单会被自动删除,第二个用例的断言就会正常通过。
3.4 进阶:批量清理和异常场景处理
如果一个用例需要插入多条测试数据,我们可以在fixture里把所有要清理的数据存到一个列表里,然后在终结器里批量删除:
@pytest.fixture(scope="function")
def multiple_test_orders(db_session):
# 前置逻辑:插入多条测试订单
test_orders = []
for i in range(3):
order = Order(
order_no=f"TEST_ORDER_{i+1}",
user_id=1001,
amount=10 * (i+1)
)
db_session.add(order)
test_orders.append(order)
db_session.commit()
yield test_orders # 把订单列表传给用例
# 后置逻辑:批量删除所有测试订单
for order in test_orders:
db_session.delete(order)
db_session.commit()
另外,我们还可以处理清理逻辑本身出错的情况,比如要删的订单已经被其他用例删了,我们可以在终结器里加异常捕获,避免清理错误影响测试结果:
@pytest.fixture(scope="function")
def test_order_safe(db_session):
test_order = Order(
order_no="TEST_ORDER_SAFE",
user_id=1002,
amount=199.99
)
db_session.add(test_order)
db_session.commit()
yield test_order
# 后置逻辑:加异常捕获,避免清理错误影响测试
try:
db_session.delete(test_order)
db_session.commit()
except Exception as e:
# 清理出错可以打印日志,不用抛异常
print(f"清理测试订单出错:{e}")
四、技术应用场景、优缺点和注意事项
4.1 应用场景
这个fixture终结器的清理方案,适合很多测试场景: 第一是数据库相关的测试,比如接口测试、业务逻辑测试,只要用例需要修改数据库,都可以用这个方案自动清理脏数据; 第二是需要创建临时资源的测试,比如测试上传文件的功能,fixture可以先创建一个临时文件,用例跑完之后自动删除; 第三是共享测试环境的场景,比如多个开发或者测试人员共用一个测试数据库,每个用例自己的脏数据自己清,不会影响别人的测试; 第四是并行测试的场景,多个用例同时跑,每个用例的清理只针对自己产生的数据,不会互相干扰。
4.2 技术优缺点
这个方案的优点很明显: 第一是可靠性高,fixture的终结器是pytest保证一定会执行的,不管用例成功还是失败,不管有没有抛异常,都能清理脏数据; 第二是精准性高,每个用例只清理自己产生的脏数据,不会误删其他用例的数据; 第三是灵活性高,可以根据需要调整fixture的作用域(scope),比如module级的fixture可以清理整个模块的脏数据,function级的fixture可以清理单个用例的脏数据; 第四是效率高,不用每次跑测试都重建数据库,只清理脏数据,节省时间。
当然也有一些缺点: 第一是需要写fixture,增加了测试代码的复杂度,尤其是复杂的测试场景,需要写很多fixture; 第二是如果用例产生的脏数据太多,批量清理的速度可能会比较慢; 第三是如果数据库的约束比较复杂(比如外键约束),清理的时候可能会遇到问题,比如要删的订单有对应的子订单,需要先删子订单再删主订单。
4.3 注意事项
用这个方案的时候,有几个注意事项: 第一是fixture的作用域要选对,比如如果fixture的scope是module,那么整个模块跑完之后才会执行终结器,如果模块里有多个用例,每个用例都产生脏数据,那么这些脏数据会留在数据库里直到模块跑完,可能会影响同一个模块里的其他用例,所以一般单个用例的脏数据清理,用scope="function"; 第二是清理逻辑要和前置逻辑对应,比如前置逻辑插了一条订单,后置逻辑就要删这条订单,不能漏; 第三是要处理清理逻辑的异常,比如要删的数据已经不存在了,不要让清理错误导致整个测试任务失败; 第四是如果是并行测试,要保证每个用例产生的脏数据是唯一的,比如订单编号用唯一的测试编号,避免多个用例产生的脏数据冲突; 第五是不要在终结器里写太复杂的逻辑,终结器的作用就是清理,不要加业务逻辑,避免影响清理的可靠性。
五、文章总结
pytest测试数据清理失效的问题,本质上是因为普通的清理逻辑没有覆盖所有场景,或者不够精准。而fixture的终结器机制,通过pytest保证一定会执行的后置逻辑,实现了每个用例的精准清理,解决了脏数据的问题,保证了测试用例的幂等性——也就是同一个用例不管跑多少次,结果都是一样的,不会因为之前的测试数据影响结果。
这个方案不仅适用于数据库测试,还适用于很多需要临时资源的测试场景,只要合理设计fixture的前置和后置逻辑,就能大大提高测试的可靠性和效率。
评论
围绕“pytest测试数据清理为何经常失效,利用fixture的终结器机制自动回滚数据库残留记录确保用例幂等”参与讨论