一、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的前置和后置逻辑,就能大大提高测试的可靠性和效率。