pytest进阶：fixture/parametrize/conftest深度解析

Python 测试与调试专题 · 掌握pytest核心机制的进阶用法

专题：Python 测试与调试系统学习

关键词：Python, 测试, 调试, pytest, fixture, parametrize, conftest, 钩子函数, 依赖注入, Python测试

一、fixture作用域链

fixture的作用域决定了其生命周期。pytest支持四种内建作用域：function（每个测试函数执行一次）、class（每个测试类执行一次）、module（每个测试模块执行一次）和session（整个测试会话执行一次）。理解作用域链是编写高效测试的关键——高作用域fixture通常用于数据库连接、Web客户端等昂贵资源的初始化，低作用域fixture则用于测试级别的数据准备。

作用域链嵌套规则

当一个低作用域fixture依赖于高作用域fixture时，高作用域fixture的实际缓存周期仍由其自身作用域决定。例如，一个function作用域的fixture调用了一个session作用域的fixture，被调用的fixture在整个会话期间只会被创建一次，而非每个测试函数都重新创建。这种机制使得fixture间可以灵活组合而不损失性能。

@pytest.fixture(scope="session")
def db_conn():
    conn = create_database_connection()
    yield conn
    conn.close()

@pytest.fixture(scope="function")
def user_record(db_conn):
    user = db_conn.insert({"name": "test"})
    yield user
    db_conn.delete(user["id"])

def test_user_creation(user_record):
    assert user_record["name"] == "test"

动态scope与fixture缓存

在某些场景下，fixture的作用域需要根据运行时条件动态决定。pytest允许通过向fixture装饰器传递一个可调用对象来实现动态scope。该可调用对象接收fixture名称和配置参数，返回一个有效的作用域字符串。fixture缓存机制确保相同作用域内的fixture实例被复用：当同一作用域（如module）内的多个测试函数请求同一fixture时，pytest仅执行一次fixture函数，后续调用直接从缓存返回结果。理解这一机制可以帮助避免因fixture状态共享导致的测试间干扰。

def determine_scope(fixture_name, config):
    if config.getoption("--db", None):
        return "session"
    return "function"

@pytest.fixture(scope=determine_scope)
def dynamic_db(request):
    return create_connection()

@pytest.fixture
def counter():
    """验证fixture缓存：每次请求同一实例"""
    print("创建counter实例")
    return {"count": 0}

def test_counter_1(counter):
    counter["count"] += 1
    assert counter["count"] == 1

def test_counter_2(counter):
    # 如果作用域为function，每个测试获得新实例
    assert counter["count"] == 0

自动使用autouse fixture

使用autouse=True可以让fixture在所有测试中自动生效，无需显式声明依赖。这对于全局的环境设置（如环境变量、日志配置）非常实用。autouse fixture的作用域同样遵循scope规则，session级别的autouse fixture在测试会话开始前自动执行。

@pytest.fixture(autouse=True, scope="function")
def setup_env():
    os.environ["TEST_MODE"] = "true"
    yield
    del os.environ["TEST_MODE"]

def test_env():
    assert os.environ["TEST_MODE"] == "true"

二、fixture高级特性

pytest fixture远不止简单的依赖注入，它还提供了丰富的内建功能接口。request对象是fixture函数内部可以访问的一个特殊对象，它提供了有关当前测试请求的上下文信息，包括测试函数名称、所在模块、所在类、标记（mark）信息等。通过request对象，fixture可以根据调用者不同而动态调整行为，实现高度灵活的测试基础设施。

request对象详解

request对象包含了fixture被请求时的完整上下文。request.node指向测试节点对象，可获取测试函数名（function.__name__）、测试类名（cls.__name__）、模块名（module.__name__）以及所在文件路径（fspath）。request.fixturename返回当前fixture的名称。request.param在fixture被参数化时携带当前参数值。此外，request.addfinalizer()可以在fixture yield方式不可用时注册清理函数。掌握了request对象，fixture的元编程能力将大幅提升。

@pytest.fixture
def smart_fixture(request):
    print(f"当前测试: {request.function.__name__}")
    print(f"所在模块: {request.module.__name__}")

    if "integration" in request.keywords:
        # 集成测试使用真实数据源
        return {"mode": "integration", "url": "https://api.example.com"}
    else:
        # 单元测试使用Mock
        return {"mode": "unit", "url": "http://localhost:8000"}

@pytest.mark.integration
def test_integration(smart_fixture):
    assert smart_fixture["mode"] == "integration"

def test_unit(smart_fixture):
    assert smart_fixture["mode"] == "unit"

fixture参数化（params）

fixture的params参数允许fixture自身被参数化，每个参数值都会导致使用该fixture的测试用例被多次执行。fixture函数内部通过request.param获取当前参数值。这种机制非常适用于需要针对多种配置运行同一组测试的场景，例如多种数据库后端、多种API版本或多种语言环境。参数化的fixture会自动生成包含参数值的测试节点ID，便于识别失败用例。

@pytest.fixture(params=["mysql", "postgresql", "sqlite"])
def db_backend(request):
    if request.param == "mysql":
        conn = MySQLConnection("localhost")
    elif request.param == "postgresql":
        conn = PostgreSQLConnection("localhost")
    else:
        conn = SQLiteConnection(":memory:")
    yield conn
    conn.close()

def test_query(db_backend):
    # 该测试会针对三种数据库各执行一次
    result = db_backend.execute("SELECT 1")
    assert result is not None

内置fixture：tmpdir、capsys、monkeypatch

pytest提供了一组强大的内置fixture。tmpdir为每个测试函数提供了一个临时目录，测试结束后自动清理。capsys可以捕获stdout和stderr输出，用于测试命令行工具或日志语句。monkeypatch可以安全地修改对象、字典、环境变量，并在测试结束后自动恢复原始值。这些内置fixture覆盖了测试中常见的需求，避免开发者重复造轮子。

def test_file_operations(tmpdir):
    test_file = tmpdir.join("test.txt")
    test_file.write("Hello, pytest!")
    assert test_file.read() == "Hello, pytest!"

def test_output_capture(capsys):
    print("标准输出")
    import sys; sys.stderr.write("错误输出\n")
    captured = capsys.readouterr()
    assert captured.out == "标准输出\n"
    assert captured.err == "错误输出\n"

def test_env_modification(monkeypatch):
    monkeypatch.setenv("DATABASE_URL", "sqlite:///test.db")
    monkeypatch.setattr("os.path.exists", lambda x: True)
    assert os.environ["DATABASE_URL"] == "sqlite:///test.db"
    assert os.path.exists("/any/path") is True

fixture工厂模式

当fixture需要接收运行时参数时，可以使用工厂模式：fixture返回一个函数，由该函数创建实际的对象。工厂函数可以接收参数，每次调用生成不同的实例。这种模式在需要灵活生成测试数据的场景下非常有用，既保持了fixture的复用性，又提供了参数化能力。

@pytest.fixture
def user_factory():
    created_users = []

    def _create_user(name, age=18, role="user"):
        user = {"name": name, "age": age, "role": role, "id": len(created_users) + 1}
        created_users.append(user)
        return user

    yield _create_user
    # 清理所有创建的用户
    for user in created_users:
        delete_user(user["id"])

def test_admin_user(user_factory):
    admin = user_factory("admin", role="admin")
    assert admin["role"] == "admin"

def test_multiple_users(user_factory):
    u1 = user_factory("Alice")
    u2 = user_factory("Bob", age=25)
    assert u2["id"] == 2
    assert u1["id"] == 1

三、fixture依赖注入

fixture依赖注入是pytest最核心的设计模式之一。一个fixture可以声明对其他fixture的依赖，pytest会自动解析依赖图并按正确顺序执行。这种机制类似于专业DI（依赖注入）框架的功能，但使用起来更加简洁直观。fixture的返回值可以直接作为参数注入到依赖它的fixture或测试函数中，形成了清晰的依赖链。理解依赖注入的解析机制对设计稳定、可维护的测试套件至关重要。

fixture调用fixture

fixture之间的相互调用是构建复杂测试基础设施的基础。例如，一个API客户端fixture可能依赖于认证令牌fixture，而认证令牌fixture又依赖于用户账户fixture。pytest的依赖解析器会自动计算拓扑排序，确保每个fixture在其依赖项就绪后才执行。如果存在循环依赖，pytest会在收集测试阶段就抛出FixtureCycleError，避免运行时的死锁问题。

@pytest.fixture
def user_account():
    return {"username": "test_user", "password": "secret123"}

@pytest.fixture
def auth_token(user_account):
    # 依赖user_account进行登录
    response = login_api(user_account["username"], user_account["password"])
    return response["token"]

@pytest.fixture
def api_client(auth_token):
    # 依赖auth_token创建已认证的客户端
    client = APIClient(base_url="https://api.example.com")
    client.set_token(auth_token)
    return client

def test_get_user_profile(api_client):
    """api_client间接依赖user_account和auth_token"""
    profile = api_client.get("/user/profile")
    assert profile["status"] == 200

fixture间依赖管理与资源共享

在设计大型测试套件时，fixture依赖管理需要遵循一些最佳实践。首先，fixture应该职责单一，每个fixture只负责一个资源的创建和维护。其次，依赖链不应过长，过深的依赖链会导致测试可读性下降和调试困难。推荐的做法是将常用依赖组合为更高层次的复合fixture。例如，将"已登录用户 + 已创建项目 + 已授权访问"这个组合封装为一个seeded_session fixture，而不是在每个测试中分别声明三个fixture依赖。

@pytest.fixture(scope="module")
def database():
    db = create_test_database()
    yield db
    drop_test_database(db)

@pytest.fixture(scope="module")
def seeded_database(database):
    database.execute("INSERT INTO users VALUES (1, 'admin')")
    database.execute("INSERT INTO config VALUES ('feature_x', 'enabled')")
    return database

@pytest.fixture
def user_session(seeded_database):
    session = Session(seeded_database)
    session.login("admin")
    return session

@pytest.fixture
def admin_client(user_session):
    return AdminClient(user_session)

def test_admin_feature(admin_client):
    result = admin_client.toggle_feature("feature_x", False)
    assert result["success"] is True

循环依赖的解决方案

虽然pytest禁止fixture间的循环依赖，但在复杂场景中有时确实需要两个fixture互相引用。解决循环依赖的典型方案有两种：一是使用request对象间接打破循环，通过request.getfixturevalue()在fixture函数体内动态获取其他fixture；二是重新设计fixture结构，将共享状态提取为独立的第三个fixture。后一种方案更加符合单一职责原则，推荐优先使用。

# 方案一：使用request.getfixturevalue() 延迟获取
@pytest.fixture
def service_a(request):
    service_b = request.getfixturevalue("service_b")
    return ServiceA(dependency=service_b)

@pytest.fixture
def service_b(request):
    service_a = request.getfixturevalue("service_a")
    return ServiceB(dependency=service_a)

# 方案二（推荐）：提取共享状态
@pytest.fixture
def shared_state():
    return {"a_ready": False, "b_ready": False}

@pytest.fixture
def service_a(shared_state):
    return ServiceA(state=shared_state)

@pytest.fixture
def service_b(shared_state):
    return ServiceB(state=shared_state)

四、conftest层次化设计

conftest.py是pytest项目配置的核心文件，它允许在目录级别定义fixture、钩子函数和插件配置。conftest的一大特性是其层次化作用域：每个目录下的conftest.py文件中的fixture对该目录及其所有子目录中的测试文件可见。根目录的conftest定义全局fixture，子目录的conftest可以覆盖或扩展父级的行为。这种层次化设计使得大型项目的fixture管理变得井然有序。

目录层级conftest与fixture可见性

conftest的层次化特性遵循就近覆盖原则。如果一个fixture在多个层级的conftest中都有定义，测试文件最近的conftest版本优先级最高。这意味着团队可以在顶层conftest中定义通用的模拟客户端或数据库连接，而每个子模块的conftest可以定义该模块特有的fixture。fixture的可见性仅限于定义它的conftest所在目录及其子目录——兄弟目录无法访问对方conftest中的fixture，这提供了天然的隔离性。

# tests/conftest.py —— 全局fixture
@pytest.fixture(scope="session")
def global_config():
    return {"base_url": "http://localhost", "timeout": 30}

@pytest.fixture
def logger():
    return setup_test_logger()

# tests/api/conftest.py —— API测试专用的fixture
@pytest.fixture
def api_client(global_config):
    # 可见：global_config 来自上级conftest
    return APIClient(base_url=global_config["base_url"])

# tests/db/conftest.py —— 数据库测试专用的fixture
@pytest.fixture
def db_connection(global_config):
    # 可见：global_config 来自上级conftest
    return DBConnection(timeout=global_config["timeout"])

钩子函数在conftest中的应用

conftest.py同样是定义pytest钩子函数的最佳位置。钩子函数是pytest插件系统的核心扩展点，允许用户在测试执行的不同阶段注入自定义逻辑。例如，pytest_configure在测试会话开始时执行，可用于注册自定义标记（mark）或初始化全局资源。pytest_collection_modifyitems在测试用例收集完成后执行，可用于修改测试用例的顺序、添加标记或过滤测试。这些钩子在conftest中定义后会自动注册，无需额外配置。

# conftest.py
def pytest_configure(config):
    """注册自定义标记，避免pytest UnknownMarkWarning"""
    config.addinivalue_line("markers", "slow: 标记慢速测试，默认跳过")
    config.addinivalue_line("markers", "smoke: 冒烟测试标记")

def pytest_collection_modifyitems(config, items):
    """收集后自动添加标记：根据测试文件名分类"""
    for item in items:
        if "integration" in item.nodeid:
            item.add_marker(pytest.mark.slow)
        if item.get_closest_marker("slow"):
            # skip默认环境下跳过慢速测试
            if not config.getoption("--run-slow"):
                item.add_marker(pytest.mark.skip(reason="需要 --run-slow 选项"))

def pytest_addoption(parser):
    parser.addoption(
        "--run-slow", action="store_true", default=False,
        help="运行标记为slow的测试"
    )

配置覆盖规则

pytest的配置遵循"就近优先"的覆盖规则。对于conftest中定义的fixture，子目录中的同名fixture会覆盖父目录的版本。对于pytest.ini或pyproject.toml中指定的配置项，命令行参数优先级最高，其次是项目级配置文件，再次是目录级配置。conftest中的钩子函数同样可以在多个层级定义，并且pytest会依次调用所有层级的钩子实现。理解这些覆盖规则有助于避免配置冲突和意外的行为覆盖。

# tests/conftest.py —— 父级fixture
@pytest.fixture
def data_provider():
    return DefaultDataProvider()

# tests/api/v2/conftest.py —— 子级覆盖
@pytest.fixture
def data_provider():
    return V2DataProvider()  # 覆盖父级DefaultDataProvider

# tests/api/v2/test_endpoints.py —— 使用v2版本
def test_create_item(data_provider):
    # 实际获取的是V2DataProvider实例
    item = data_provider.create("test")
    assert item["version"] == 2

五、参数化进阶

pytest的参数化功能是其最具生产力的特性之一。通过@pytest.mark.parametrize装饰器，可以用极少的代码量覆盖大量输入组合。进阶用法涉及多个parametrize组合的笛卡尔积、从外部数据源（JSON/YAML）动态加载参数、参数化ID的自定义格式化，以及与fixture参数化的联合使用。掌握这些技巧可以显著提升测试覆盖率而不增加代码冗余。

多个parametrize组合（笛卡尔积）

当测试函数上叠加多个@pytest.mark.parametrize装饰器时，pytest会计算所有参数值的笛卡尔积。例如，一个参数有3个值，另一个参数有4个值，则会生成12个独立的测试用例。这种组合方式特别适合测试多维输入空间。需要注意的是，叠加顺序会影响测试用例的执行顺序和生成的测试ID，上层装饰器的参数变化较慢（类似嵌套循环的外层）。

import pytest

@pytest.mark.parametrize("operand_a", [1, 2, 3])
@pytest.mark.parametrize("operand_b", [10, 100])
@pytest.mark.parametrize("operation", ["add", "multiply"])
def test_operations(operation, operand_b, operand_a):
    """
    共生成 2(operation) x 2(operand_b) x 3(operand_a) = 12 个测试用例
    执行顺序：operation最内层变化最快，operand_a最外层变化最慢
    """
    if operation == "add":
        result = operand_a + operand_b
    else:
        result = operand_a * operand_b

    if operation == "add":
        assert result == operand_a + operand_b
    else:
        assert result == operand_a * operand_b

从JSON/YAML加载参数

在实际项目中，测试数据通常不硬编码在测试文件中，而是存储在外部数据文件中。pytest的parametrize可以接受动态生成的参数列表，因此从JSON或YAML文件加载数据非常自然。通过编写一个helper函数读取外部文件并返回参数列表，可以将测试逻辑与测试数据完全分离。这种方式便于非开发人员（如测试分析师）维护测试用例数据集。

import json
import pytest

def load_test_data(json_path):
    """从JSON文件加载测试参数"""
    with open(json_path, "r", encoding="utf-8") as f:
        data = json.load(f)
    # 期望JSON结构: [{"input": ..., "expected": ...}, ...]
    return [(item["input"], item["expected"]) for item in data]

# 从外部数据文件加载测试用例
test_cases = load_test_data("test_data/api_cases.json")

@pytest.mark.parametrize("user_input,expected", test_cases,
                         ids=[f"case_{i}" for i in range(len(test_cases))])
def test_from_json(user_input, expected):
    result = process_input(user_input)
    assert result == expected

import yaml
import pytest

def load_yaml_data(yaml_path):
    with open(yaml_path, "r", encoding="utf-8") as f:
        data = yaml.safe_load(f)
    return data["test_cases"]

# YAML格式:
# test_cases:
#   - name: "正常输入"
#     input: {"username": "alice"}
#     expected: 200
#   - name: "空用户名"
#     input: {"username": ""}
#     expected: 400

@pytest.mark.parametrize("case", load_yaml_data("test_cases.yaml"),
                         ids=lambda c: c["name"])
def test_api_endpoints(case):
    response = call_api(case["input"])
    assert response.status_code == case["expected"]

参数化ID格式化

pytest允许通过ids参数自定义测试用例的显示名称。ids可以是一个字符串列表（长度与参数数量一致），也可以是一个可调用对象，它接收当前参数值并返回字符串。良好的ID命名使测试报告更具可读性，方便在大量失败用例中快速定位问题。ids参数与pytest.param()中的id参数配合使用，可以为每个参数值单独命名，特别适合处理复杂的参数化场景。

import pytest

# 使用可调用对象格式化ID
@pytest.mark.parametrize("text,length", [
    ("hello", 5),
    ("world", 5),
    ("pytest", 6),
    ("", 0),
], ids=lambda params: f"len_{params[1]}_text={params[0] or 'empty'}")
def test_string_length(text, length):
    assert len(text) == length

# 使用pytest.param精确控制ID
@pytest.mark.parametrize("value", [
    pytest.param(0, id="zero"),
    pytest.param(1, id="positive_small"),
    pytest.param(-1, id="negative_small"),
    pytest.param(2**31 - 1, id="max_int32"),
])
def test_abs_value(value):
    assert abs(value) >= 0

六、钩子函数体系

pytest的钩子函数（hooks）是其插件系统的基石。钩子函数本质上是在测试执行流程的关键节点定义的接口，插件或conftest可以实现这些接口来干预或扩展pytest的行为。pytest定义了超过50个钩子函数，覆盖了从测试收集、测试执行到测试报告的完整生命周期。掌握核心钩子函数的用途和调用时机，可以深入定制测试框架的行为以满足项目特定需求。

pytest_runtest_protocol——测试执行协议

pytest_runtest_protocol是测试执行的核心钩子，它定义了单个测试用例从开始到结束的完整协议。默认实现依次调用pytest_runtest_setup、pytest_runtest_call和pytest_runtest_teardown三步。通过自定义此钩子，可以实现测试重试、超时控制、执行追踪等高级功能。该钩子返回True表示测试已处理，pytest不再调用其他实现；返回None则继续执行后续实现。

# conftest.py
import time

def pytest_runtest_protocol(item, nextitem):
    """为每个测试添加执行时间记录"""
    start = time.time()
    # 调用默认的测试执行流程
    reports = pytest_runtest_protocol(item, nextitem)
    duration = time.time() - start
    print(f"  [{item.nodeid}] 耗时: {duration:.3f}s")
    return reports

# 注意：实际自定义时不要直接递归调用此函数
# 下面是通过hookwrapper实现的更好方式
import pytest

@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_call(item):
    """Hookwrapper: 在测试执行前后添加逻辑"""
    start = time.time()
    yield  # 执行实际的测试逻辑
    duration = time.time() - start
    if duration > 2.0:
        print(f"  ⚠ {item.nodeid} 执行过慢: {duration:.2f}s")

pytest_collection_modifyitems——测试收集修改

此钩子在测试用例收集完成后被调用，接收items列表（所有已收集的测试用例）。这是实现测试筛选、排序、分组的最常用钩子。例如，可以根据测试文件名自动添加标记，根据标记重新排序执行顺序（慢速测试放在最后），或者根据环境变量决定跳过某些测试。该钩子接收config对象和items列表，items是一个列表引用，可以直接修改。

def pytest_collection_modifyitems(config, items):
    """自动标记和组织测试用例"""
    # 1. 根据路径自动添加标记
    for item in items:
        if "integration" in str(item.fspath):
            item.add_marker(pytest.mark.integration)
        if "benchmark" in str(item.fspath):
            item.add_marker(pytest.mark.benchmark)

    # 2. 将冒烟测试排在前面执行
    smoke_tests = [it for it in items if it.get_closest_marker("smoke")]
    non_smoke = [it for it in items if not it.get_closest_marker("smoke")]
    items[:] = smoke_tests + non_smoke

    # 3. 为测试添加自定义属性
    for item in items:
        item._my_module = item.getparent(pytest.Module).name.replace(".py", "")

pytest_configure与pytest_report_header

pytest_configure钩子在测试会话初始化期间被调用，常用于注册自定义标记、添加配置选项、初始化全局资源等。pytest_report_header允许向pytest的输出头部添加自定义信息，如数据库连接状态、API版本号、环境标识等。这两个钩子配合使用，可以在测试开始前收集和展示环境信息，便于问题诊断。

def pytest_configure(config):
    """测试会话配置初始化"""
    # 注册自定义标记
    config.addinivalue_line("markers", "regression: 回归测试")
    config.addinivalue_line("markers", "smoke: 冒烟测试")
    config.addinivalue_line("markers", "performance: 性能测试")

    # 根据环境变量设置配置
    env = os.getenv("TEST_ENV", "dev")
    config._test_env = env
    print(f"\n  测试环境: {env}")

def pytest_report_header(config, start_path):
    """向pytest输出头部添加自定义信息"""
    env = getattr(config, "_test_env", "unknown")
    python_version = sys.version.split()[0]
    return [
        f"测试环境: {env}",
        f"Python版本: {python_version}",
        f"工作目录: {start_path}",
    ]

七、自定义标记与插件

pytest的标记（mark）和插件系统使其成为一个高度可扩展的测试框架。自定义标记允许为测试用例附加元数据，并在运行时有选择地执行。插件开发则能将通用的测试基础设施包装为可复用的包，跨项目共享。pytest的插件架构基于钩子函数实现，开发者只需实现对应的钩子并注册即可创建一个功能完整的插件。本节将深入标记注册、自定义命令行选项和插件打包的完整流程。

标记注册与解析

使用自定义标记前必须先注册，否则pytest会发出UnknownMarkWarning警告。注册方式有两种：在pytest.ini中用markers配置项注册，或使用config.addinivalue_line在pytest_configure钩子中注册。运行时可以通过-m选项选择标记表达式，如pytest -m "smoke and not slow"执行冒烟测试但排除慢速测试。标记可以携带参数，通过item.get_closest_marker(name)获取标记对象后访问其args和kwargs属性。

# 在conftest.py中注册自定义标记
def pytest_configure(config):
    config.addinivalue_line("markers",
        "issue(id): 关联的Issue编号")
    config.addinivalue_line("markers",
        "timeout(seconds): 测试超时时间")
    config.addinivalue_line("markers",
        "datafile(path): 关联的测试数据文件")

# 在测试中使用自定义标记
@pytest.mark.issue(1234)
@pytest.mark.timeout(30)
@pytest.mark.datafile("users.json")
def test_user_api():
    # 运行时解析标记
    pass

# 动态解析标记内容
def pytest_runtest_call(item):
    timeout_mark = item.get_closest_marker("timeout")
    if timeout_mark:
        timeout = timeout_mark.args[0]  # 获取30
        signal.alarm(timeout)

自定义命令行选项

pytest允许通过pytest_addoption钩子添加自定义命令行参数。该钩子接收一个parser对象（pytest.Parser实例），可以调用parser.addoption()添加任意选项，选项的语义与argparse兼容。添加的选项值通过config.getoption()在钩子函数或fixture中获取。自定义命令行选项使得测试套件可以根据运行参数灵活切换行为，例如选择测试环境、指定数据库类型、控制日志级别等。

# conftest.py
def pytest_addoption(parser):
    parser.addoption(
        "--env", action="store", default="dev",
        choices=["dev", "staging", "prod"],
        help="指定测试环境: dev, staging, prod"
    )
    parser.addoption(
        "--db-type", action="store", default="sqlite",
        choices=["sqlite", "mysql", "postgres"],
        help="指定数据库类型"
    )
    parser.addoption(
        "--log-level", action="store", default="INFO",
        choices=["DEBUG", "INFO", "WARNING", "ERROR"],
        help="设置日志级别"
    )

@pytest.fixture
def test_config(request):
    return {
        "env": request.config.getoption("--env"),
        "db_type": request.config.getoption("--db-type"),
        "log_level": request.config.getoption("--log-level"),
    }

# 命令行: pytest --env staging --db-type mysql --log-level DEBUG

插件打包与分发

将常用的测试基础设施封装为可安装的插件包，可以跨项目复用。创建一个pytest插件需要：准备一个包含钩子函数的Python模块；配置setup.py或pyproject.toml，将模块注册为pytest11的入口点（entry point）；在conftest.py中注册钩子实现。发布到PyPI后，其他项目只需pip install即可使用。Anthropic内部也可以搭建私有PyPI服务器来分发测试插件。

# my_pytest_plugin/plugin.py
"""一个简单的pytest插件：自动跳过耗时过长的测试"""
import time

def pytest_addoption(parser):
    parser.addoption("--max-test-time", action="store", type=float,
                     default=60.0, help="测试最大运行时间（秒）")

@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_call(item):
    max_time = item.config.getoption("--max-test-time")
    start = time.time()
    yield
    duration = time.time() - start
    if duration > max_time:
        pytest.fail(f"测试超时: {duration:.2f}s > {max_time}s")

# pyproject.toml 配置入口点
"""
[project.entries."pytest11"]
my_plugin = "my_pytest_plugin.plugin"
"""

# 使用方式（安装后自动生效）
# pip install my-pytest-plugin
# pytest --max-test-time 30 tests/

八、测试执行控制

当测试项目规模增长到数百甚至数千个测试用例时，测试执行效率成为关键问题。pytest生态提供了多个第三方插件来应对大规模测试执行挑战：pytest-xdist实现并行执行、pytest-rerunfailures实现失败重试、pytest-timeout实现超时控制。这些插件的配合使用可以大幅缩短CI流水线时间，同时提高测试的稳定性。

失败重试：pytest-rerunfailures

pytest-rerunfailures插件允许在测试失败时自动重试指定次数。这对于处理网络抖动、资源竞争、外部服务不稳定等非确定性失败非常有效。通过--reruns N选项设置全局重试次数，--reruns-delay D设置重试间隔秒数。也可以使用@pytest.mark.flaky(reruns=N, reruns_delay=D)为特定测试指定重试策略。需要注意的是，重试不应掩盖真正的代码缺陷，建议对flaky测试单独标记并定期审查。

# 安装: pip install pytest-rerunfailures

# 命令行用法
# pytest --reruns 3 --reruns-delay 1 tests/

# 标记特定测试为重试策略
@pytest.mark.flaky(reruns=5, reruns_delay=2, condition="CI" in os.environ)
def test_unstable_network_api():
    """网络不稳定的API测试，只在CI环境中重试5次"""
    response = requests.get("https://external-api.example.com/data")
    assert response.status_code == 200

# 在conftest.py中配置条件重试
def pytest_collection_modifyitems(config, items):
    for item in items:
        if item.get_closest_marker("integration"):
            # 集成测试默认重试2次
            item.add_marker(pytest.mark.flaky(reruns=2))

并行执行：pytest-xdist

pytest-xdist通过将测试分发到多个worker进程来实现并行执行，可以显著利用多核CPU缩短测试时间。使用-n auto参数自动检测CPU核心数并启动对应数量的worker，或使用-n N手动指定worker数量。xdist还支持--dist loadscope（按模块分发）和--dist loadfile（按文件分发）等分发策略。需要注意的是，并行执行要求测试具有良好的隔离性——共享资源（如数据库）需要额外处理以避免竞争条件。

# 安装: pip install pytest-xdist

# 常用命令行
# pytest -n auto tests/                    # 使用所有CPU核心
# pytest -n 4 --dist loadscope tests/      # 4个worker，按模块分发
# pytest -n auto --pdb tests/             # 不支持，--pdb与并行冲突

# 标记某些测试不可并行执行
@pytest.mark.xdist_group(name="serial_tests")
def test_database_migration():
    """数据库迁移测试，必须串行执行"""
    run_migration()
    assert check_schema_version() == 2

# conftest.py 中设置worker标识
@pytest.fixture
def worker_id(request):
    """获取当前worker的ID，xdist下为 gw0, gw1, ..."""
    if hasattr(request.config, "workerinput"):
        return request.config.workerinput["workerid"]
    return "master"

# 每个worker使用独立的数据库
@pytest.fixture
def db_name(worker_id):
    return f"test_db_{worker_id}"

超时控制：pytest-timeout

pytest-timeout插件可以为测试用例设置执行超时，防止死循环或挂起的测试阻塞整个测试套件。超时机制支持两种模式：signal（基于Unix信号，默认）和thread（基于线程）。signal模式更精确但仅在Unix系统上可用，thread模式跨平台但可能无法中断某些底层C扩展调用。超时设置可以在命令行（--timeout=N）、配置文件或标记级别（@pytest.mark.timeout(N)）指定。

# 安装: pip install pytest-timeout

# 命令行用法
# pytest --timeout=60 tests/              # 全局60秒超时
# pytest --timeout=120 --timeout-method=thread tests/

# 在pytest.ini中设置
"""
[pytest]
timeout = 300
timeout_method = thread
"""

# 标记特定测试的超时时间
@pytest.mark.timeout(10)  # 10秒超时
def test_quick():
    assert quick_computation() == 42

@pytest.mark.timeout(300, method="signal")  # 5分钟超时，signal模式
def test_heavy_computation():
    result = heavy_computation()
    assert result is not None

# conftest中动态设置超时
def pytest_collection_modifyitems(config, items):
    for item in items:
        if item.get_closest_marker("slow"):
            # 慢速测试默认超时更宽松
            if not item.get_closest_marker("timeout"):
                item.add_marker(pytest.mark.timeout(600))

九、实战案例

理论知识的价值最终体现在实际项目中。本节通过两个实战案例——复杂Web测试的fixture设计和多层conftest项目架构——展示如何将前八节的知识融会贯通。这些案例来自真实项目的测试架构经验，涵盖了fixture链式依赖、conftest层次化组织、参数化数据驱动、钩子函数应用等综合场景。

案例一：复杂Web测试fixture设计

在一个典型的RESTful API测试项目中，我们需要处理用户认证、资源创建、数据清理等多个层次的fixture。核心思路是采用工厂模式的fixture链：config_fixture提供全局配置，auth_fixture处理令牌获取，client_fixture创建带认证的HTTP客户端，resource_fixture负责创建和清理测试资源。每个fixture职责单一，通过依赖注入组合成完整的功能链路。fixture的teardown部分（yield之后的清理代码）确保测试资源被正确释放，即使测试失败也不会留下孤儿数据。

import pytest
import requests

@pytest.fixture(scope="session")
def api_config():
    """全局API配置，从环境变量读取"""
    return {
        "base_url": os.getenv("API_BASE_URL", "http://localhost:8000"),
        "admin_user": os.getenv("ADMIN_USER", "admin"),
        "admin_pass": os.getenv("ADMIN_PASS", "admin123"),
        "timeout": int(os.getenv("API_TIMEOUT", "30")),
    }

@pytest.fixture(scope="session")
def auth_token(api_config):
    """获取管理员认证令牌（session级别，只获取一次）"""
    response = requests.post(
        f"{api_config['base_url']}/auth/login",
        json={
            "username": api_config["admin_user"],
            "password": api_config["admin_pass"],
        },
        timeout=api_config["timeout"],
    )
    response.raise_for_status()
    token = response.json()["access_token"]
    yield token
    # 清理：登出使令牌失效
    requests.post(
        f"{api_config['base_url']}/auth/logout",
        headers={"Authorization": f"Bearer {token}"},
    )

@pytest.fixture
def api_client(api_config, auth_token):
    """创建带认证的API客户端"""
    session = requests.Session()
    session.base_url = api_config["base_url"]
    session.headers.update({
        "Authorization": f"Bearer {auth_token}",
        "Content-Type": "application/json",
    })
    session.timeout = api_config["timeout"]
    yield session
    session.close()

@pytest.fixture
def test_project(api_client):
    """创建测试项目并返回项目数据"""
    response = api_client.post("/api/projects", json={
        "name": "test-project-e2e",
        "description": "E2E测试项目",
    })
    response.raise_for_status()
    project = response.json()
    yield project
    # 清理：删除测试项目及其所有关联资源
    api_client.delete(f"/api/projects/{project['id']}")

def test_create_task_in_project(api_client, test_project):
    """在测试项目中创建任务"""
    response = api_client.post(
        f"/api/projects/{test_project['id']}/tasks",
        json={"title": "测试任务", "priority": "high"},
    )
    assert response.status_code == 201
    task = response.json()
    assert task["project_id"] == test_project["id"]

    # 验证任务已创建
    response = api_client.get(
        f"/api/projects/{test_project['id']}/tasks/{task['id']}"
    )
    assert response.status_code == 200

案例二：多层conftest项目架构

大型项目通常采用如下目录结构：顶层conftest定义全局资源（数据库、外部服务Mock等）；按功能模块划分的子目录中各有一个conftest，定义该模块特有的fixture；最底层的测试文件仅包含测试逻辑。如果项目需要同时支持MySQL和SQLite两种数据库后端，可以在顶层conftest中提供一个可参数化的数据库fixture，子模块无需关心数据库具体实现。这种分层设计使得单个测试文件的复杂度大大降低，只需关注业务逻辑验证。

# 项目目录结构：
"""
tests/
├── conftest.py              # 全局: 数据库、日志、配置
├── test_config.py
├── api/
│   ├── conftest.py          # API层: 客户端、认证
│   ├── test_users.py
│   ├── test_products.py
│   └── v2/
│       ├── conftest.py      # API v2: 覆盖API客户端v2版本
│       └── test_users_v2.py
├── services/
│   ├── conftest.py          # 服务层: Mock外部服务
│   ├── test_payment.py
│   └── test_notification.py
└── e2e/
    ├── conftest.py          # E2E: 完整环境准备
    └── test_full_flow.py
"""

# tests/conftest.py —— 全局
@pytest.fixture(scope="session")
def db_engine(pytestconfig):
    """数据库引擎，通过命令行选择后端"""
    db_type = pytestconfig.getoption("--db-type")
    if db_type == "mysql":
        engine = create_mysql_engine()
    else:
        engine = create_sqlite_engine()
    yield engine
    engine.dispose()

# tests/api/conftest.py —— API层
@pytest.fixture
def api_client(db_engine, request):
    """API客户端基类，子目录可以覆盖"""
    return APIClient(db_engine=db_engine, version="v1")

# tests/api/v2/conftest.py —— API v2覆盖
@pytest.fixture
def api_client(db_engine):
    """v2版本的API客户端"""
    return APIClient(db_engine=db_engine, version="v2")

# tests/api/v2/test_users_v2.py —— 最终测试
def test_list_users_v2(api_client):
    """使用v2版本的api_client"""
    users = api_client.get("/users")
    assert users.status_code == 200
    # 验证v2特有的响应格式
    assert "pagination" in users.json()

案例三：结合参数化的数据驱动测试

在真实项目中，同一个业务逻辑需要验证大量输入输出组合。通过fixture链提供测试基础设施，再通过parametrize驱动数据，可以实现高效的数据驱动测试。以下案例展示了一个用户注册接口的完整测试：api_client链式fixture提供HTTP客户端，register_cases从外部YAML加载多种注册场景数据，通过parametrize自动生成每个场景的测试用例，验证各种正常和异常输入的处理。

import yaml
import pytest

# 从YAML加载测试用例
def load_test_scenarios():
    scenarios = yaml.safe_load("""
    valid_registrations:
      - name: "正常用户"
        input: {"username": "newuser", "password": "Pass1234!", "email": "new@test.com"}
        expected: 201
      - name: "邮箱带加号"
        input: {"username": "plususer", "password": "Pass1234!", "email": "test+tag@test.com"}
        expected: 201
    invalid_registrations:
      - name: "用户名太短"
        input: {"username": "ab", "password": "Pass1234!", "email": "a@test.com"}
        expected: 400
      - name: "密码太弱"
        input: {"username": "stronguser", "password": "123", "email": "s@test.com"}
        expected: 400
      - name: "邮箱格式错误"
        input: {"username": "emailuser", "password": "Pass1234!", "email": "not-an-email"}
        expected: 400
    """)
    return scenarios

@pytest.fixture(scope="module")
def scenarios():
    return load_test_scenarios()

@pytest.mark.parametrize("case", [
    pytest.param(c, id=c["name"])
    for c in load_test_scenarios()["valid_registrations"]
])
def test_valid_registrations(api_client, case):
    response = api_client.post("/api/register", json=case["input"])
    assert response.status_code == case["expected"]
    # 验证用户确实已创建
    user_id = response.json()["id"]
    get_resp = api_client.get(f"/api/users/{user_id}")
    assert get_resp.status_code == 200

@pytest.mark.parametrize("case", [
    pytest.param(c, id=c["name"])
    for c in load_test_scenarios()["invalid_registrations"]
])
def test_invalid_registrations(api_client, case):
    response = api_client.post("/api/register", json=case["input"])
    assert response.status_code == case["expected"]
    # 验证错误信息格式
    error_body = response.json()
    assert "message" in error_body
    assert "error_code" in error_body

最佳实践总结：

1. fixture作用域：昂贵资源用session或module scope，测试数据用function scope。

2. conftest层次化：全局资源放顶层conftest，模块专属放子目录conftest，就近覆盖。

3. 参数化策略：简单场景用parametrize装饰器，复杂场景从JSON/YAML文件加载。

4. 钩子函数：测试标记注册、自定义命令行、测试排序在conftest中通过钩子函数实现。

5. 插件开发：将通用测试基础设施封装为pytest插件，通过entry points注册。

6. 执行控制：xdist并行加速，rerunfailures处理不稳定测试，timeout防止挂死。

7. 依赖注入：保持fixture职责单一，利用pytest自动解析依赖图。