pytest fixture 进阶实战——scope 管理、conftest、fixture 依赖注入
pytest fixture 进阶实战——scope 管理、conftest、fixture 依赖注入
适读人群:已掌握 pytest 基础的 Python 工程师 | 阅读时长:约 15 分钟 | 核心价值:深入理解 fixture 的生命周期和依赖机制,构建高效可维护的测试基础设施
一次让测试套件从 15 分钟变成 2 分钟的优化
我们团队的 API 测试套件跑得很慢——200 个测试用例,每次要跑 15 分钟。
我做了 profiling,发现 90% 的时间花在了数据库操作上:每个测试用例都在创建连接、初始化 schema、插入基础数据、执行测试、删除数据、关闭连接。
问题出在 fixture 的 scope 设置错了——我们的数据库 fixture 用的是默认的 function scope,导致每个测试用例都重新创建了一次数据库连接和全部 schema。
调整 scope 策略后,测试时间从 15 分钟降到了 2 分钟。
深入理解 fixture scope 的执行顺序
import pytest
@pytest.fixture(scope="session")
def session_fixture():
print("\n[Session] Setup")
yield "session_value"
print("\n[Session] Teardown")
@pytest.fixture(scope="module")
def module_fixture(session_fixture):
print(f"\n[Module] Setup, session={session_fixture}")
yield "module_value"
print("\n[Module] Teardown")
@pytest.fixture(scope="class")
def class_fixture(module_fixture):
print(f"\n[Class] Setup, module={module_fixture}")
yield "class_value"
print("\n[Class] Teardown")
@pytest.fixture
def function_fixture(class_fixture):
print(f"\n[Function] Setup, class={class_fixture}")
yield "function_value"
print("\n[Function] Teardown")执行 3 个测试时,输出顺序是:
[Session] Setup ← 整个会话只执行一次
[Module] Setup ← 每个模块执行一次
[Class] Setup ← 每个测试类执行一次
[Function] Setup ← 每个测试函数执行一次
# 测试 1 执行
[Function] Teardown
[Function] Setup ← 测试 2 开始
# 测试 2 执行
[Function] Teardown
[Class] Teardown
[Module] Teardown
[Session] Teardown ← 整个会话结束后才执行记住这个原则:高 scope fixture 的 teardown 在最后执行,低 scope fixture 的 teardown 先执行。
实战:数据库 fixture 的三层结构
这是我们线上项目的数据库 fixture 架构,经过真实验证:
# tests/conftest.py
import pytest
import sqlalchemy as sa
from sqlalchemy.orm import sessionmaker
from myapp.database import Base, engine_factory
from myapp.models import User, Product, Category
@pytest.fixture(scope="session")
def db_engine(config):
"""
Session scope:整个测试会话共享一个数据库引擎
只在测试开始时创建 schema,结束时删除
"""
engine = engine_factory(config.test_database_url)
# 创建所有表
Base.metadata.create_all(engine)
yield engine
# 测试结束后删除所有表
Base.metadata.drop_all(engine)
engine.dispose()
@pytest.fixture(scope="session")
def db_session_factory(db_engine):
"""
Session scope:整个会话共享连接池
"""
return sessionmaker(bind=db_engine, autocommit=False, autoflush=False)
@pytest.fixture
def db_session(db_session_factory):
"""
Function scope:每个测试有独立的数据库会话
使用事务隔离——测试结束后回滚,保证数据干净
"""
connection = db_session_factory.bind.connect()
transaction = connection.begin()
session = db_session_factory(bind=connection)
yield session
# 无论测试成功还是失败,都回滚事务
session.close()
transaction.rollback()
connection.close()关键设计:
db_engine是sessionscope,整个测试套件只创建一次 schemadb_session是functionscope,每个测试有独立会话,且用事务回滚保证隔离- 每个测试结束后事务回滚,数据库状态回到测试前,无需手动清理
fixture 工厂模式
有时候需要在一个测试中多次创建同类型的 fixture 对象:
@pytest.fixture
def make_user(db_session):
"""
工厂 fixture:返回一个工厂函数,可以多次调用创建不同的用户
"""
created_users = []
def _make_user(email=None, role="buyer", **kwargs):
if email is None:
email = f"test_{uuid.uuid4().hex[:8]}@test.com"
user = User(
email=email,
role=role,
password_hash=hash_password("Test@123"),
**kwargs
)
db_session.add(user)
db_session.flush() # 获取 ID,但不提交
created_users.append(user)
return user
yield _make_user
# 清理所有创建的用户(虽然 db_session 会回滚,显式清理更清晰)
@pytest.fixture
def make_order(db_session, make_user):
"""依赖 make_user 工厂的 make_order 工厂"""
def _make_order(user=None, status="pending", total_amount=100.0, **kwargs):
if user is None:
user = make_user()
order = Order(
user_id=user.id,
status=status,
total_amount=total_amount,
**kwargs
)
db_session.add(order)
db_session.flush()
return order
yield _make_order
# 在测试中使用工厂 fixture
def test_user_order_statistics(make_user, make_order, stats_service):
# 创建两个用户
user_a = make_user(email="user_a@test.com")
user_b = make_user(email="user_b@test.com")
# 为 user_a 创建多个订单
make_order(user=user_a, status="completed", total_amount=100)
make_order(user=user_a, status="completed", total_amount=200)
make_order(user=user_a, status="cancelled", total_amount=300)
# user_b 没有订单
stats_a = stats_service.get_user_stats(user_a.id)
assert stats_a.completed_order_count == 2
assert stats_a.completed_order_total == 300
stats_b = stats_service.get_user_stats(user_b.id)
assert stats_b.completed_order_count == 0fixture 参数化(parametrize + indirect)
@pytest.fixture
def user_with_plan(request, make_user):
"""根据参数创建不同套餐的用户"""
plan = request.param
user = make_user(plan=plan, email=f"{plan}_user@test.com")
return user
@pytest.mark.parametrize("user_with_plan,expected_features", [
("free", ["basic_search", "limited_export"]),
("pro", ["basic_search", "unlimited_export", "api_access"]),
("enterprise", ["basic_search", "unlimited_export", "api_access", "sso", "audit_log"]),
], indirect=["user_with_plan"])
def test_user_features(user_with_plan, expected_features, feature_service):
features = feature_service.get_available_features(user_with_plan.id)
feature_names = [f.name for f in features]
for expected in expected_features:
assert expected in feature_names, f"计划 {user_with_plan.plan} 应该有功能 {expected}"条件 fixture(根据环境切换实现)
@pytest.fixture
def email_service(request):
"""
根据测试环境决定使用真实邮件服务还是 Mock
"""
use_real = request.config.getoption("--use-real-email", default=False)
if use_real:
service = RealEmailService(
smtp_host=os.environ["SMTP_HOST"],
smtp_port=int(os.environ["SMTP_PORT"]),
)
else:
service = MockEmailService()
yield service
if not use_real:
service.clear_sent_emails()
# 在 conftest.py 中注册自定义命令行选项
def pytest_addoption(parser):
parser.addoption(
"--use-real-email",
action="store_true",
default=False,
help="使用真实邮件服务(默认使用 Mock)"
)高级技巧:fixture 内部使用其他 fixture
有时候 fixture 本身需要调用其他 fixture,可以通过 request 对象:
@pytest.fixture
def setup_complete_shop(request):
"""
一键初始化完整的商城数据(聚合多个 fixture)
"""
# 通过 request.getfixturevalue 获取其他 fixture
db = request.getfixturevalue("db_session")
make_user = request.getfixturevalue("make_user")
make_product = request.getfixturevalue("make_product")
# 创建商家
merchant = make_user(role="merchant", email="merchant@shop.com")
# 创建分类
categories = []
for name in ["电子产品", "服装", "图书"]:
cat = Category(name=name, merchant_id=merchant.id)
db.add(cat)
categories.append(cat)
db.flush()
# 创建商品
products = []
for i, category in enumerate(categories):
for j in range(3):
product = make_product(
name=f"{category.name}_商品_{j+1}",
category_id=category.id,
price=float((i+1) * (j+1) * 10),
stock=100
)
products.append(product)
return {
"merchant": merchant,
"categories": categories,
"products": products,
}
def test_product_search(setup_complete_shop, search_service):
products = search_service.search("电子产品")
assert len(products) == 3踩坑实录
坑一:session scope fixture 中出现了数据库事务问题
现象: session scope 的 fixture 创建的测试数据,在某些测试中看不到。
原因: 我们的 db_session 使用了事务回滚策略,但 session scope 创建的数据在不同事务中。如果种子数据是在一个 session-scope fixture 中提交的,后续的 function-scope 事务应该能看到;但如果没有提交,就看不到。
解法: session scope 的种子数据必须显式 commit:
@pytest.fixture(scope="session")
def seed_data(db_engine):
Session = sessionmaker(bind=db_engine)
session = Session()
# 创建基础数据
session.add(Category(name="默认分类", id="default-cat"))
session.commit() # 必须 commit,让后续测试可见
yield
# 清理
session.query(Category).delete()
session.commit()
session.close()坑二:fixture 互相依赖形成循环
现象: fixture A 依赖 B,B 依赖 A,pytest 报循环依赖错误。
解法: 重新设计 fixture 的分层结构,提取公共部分为独立 fixture,打破循环:
# 错误:循环依赖
@pytest.fixture
def fixture_a(fixture_b): # A 依赖 B
return ...
@pytest.fixture
def fixture_b(fixture_a): # B 依赖 A → 循环!
return ...
# 正确:提取公共部分
@pytest.fixture
def common_resource():
return ...
@pytest.fixture
def fixture_a(common_resource):
return ...
@pytest.fixture
def fixture_b(common_resource):
return ...坑三:autouse fixture 的执行顺序不符合预期
现象: 两个 autouse fixture,期望 A 先执行,B 后执行,但实际顺序相反。
解法: 让 B 显式依赖 A,建立依赖关系来控制顺序:
@pytest.fixture(autouse=True)
def setup_environment():
os.environ["APP_ENV"] = "test"
yield
del os.environ["APP_ENV"]
@pytest.fixture(autouse=True)
def setup_database(setup_environment): # 显式依赖,确保环境先初始化
# 这时 APP_ENV 一定已经设置好了
db = create_database_based_on_env()
yield db
db.close()小结
fixture 的 scope 管理是 pytest 性能优化的核心:
- session scope:数据库 engine、浏览器实例、配置对象——整个测试套件只创建一次
- module scope:模块级别的 HTTP 客户端、共享的只读数据
- class scope:测试类级别的共享状态(慎用)
- function scope(默认):测试数据、数据库 session、浏览器上下文——保证隔离
合理的 fixture 架构能让你的测试套件快 3-5 倍,同时保持测试隔离性。
