Python 测试策略：pytest fixture 作用域与 pytest-xdist 并行化

import pytest

# function 作用域：每个测试函数执行一次
@pytest.fixture
def function_scoped():
    print("Created")
    yield
    print("Destroyed")

# module 作用域：整个模块只执行一次
@pytest.fixture(scope="module")
def module_scoped():
    print("Module fixture created")
    yield
    print("Module fixture destroyed")

# session 作用域：整个测试会话只执行一次
@pytest.fixture(scope="session")
def session_scoped():
    print("Session fixture created")
    yield
    print("Session fixture destroyed")

@pytest.fixture
def base():
    print("1-base created")
    yield
    print("1-base destroyed")

@pytest.fixture
def derived(base):
    print("2-derived created")
    yield
    print("2-derived destroyed")

@pytest.fixture
def top(derived):
    print("3-top created")
    yield
    print("3-top destroyed")

# 测试函数使用 top
def test_order(top):
    pass

# 输出顺序：
# 创建: 1-base -> 2-derived -> 3-top
# 销毁: 3-top -> 2-derived -> 1-base (LIFO)

# autouse fixture：自动为所有测试创建数据库事务回滚
@pytest.fixture(autouse=True)
def db_transaction(db_connection):
    # 每个测试开始时开启事务
    transaction = db_connection.begin()
    yield
    # 测试结束后回滚，不污染数据库
    transaction.rollback()

# 条件 fixture：仅当特定模块可用时
@pytest.fixture
def redis_client():
    redis = pytest.importorskip("redis")
    client = redis.Redis()
    pytest.skipif(not client.ping(), "Redis not available")
    yield client

# 目录结构示例
# tests/
#   conftest.py        <- 被所有测试加载
#   unit/
#     conftest.py      <- 只被 unit/ 下测试加载
#     test_math.py
#   integration/
#     conftest.py      <- 只被 integration/ 下测试加载
#     test_api.py

# conftest.py 内容自动对同级及子目录测试可用
# 无需 import，pytest 自动加载

# tests/conftest.py
@pytest.fixture
def db():
    return "parent_db"

# tests/unit/conftest.py
@pytest.fixture
def db():
    return "unit_db"

# tests/unit/test_math.py
def test_db(db):
    print(db)  # "unit_db" - 子目录覆盖了父目录

# tests/conftest.py

# 在测试收集完成后修改测试项
def pytest_collection_modifyitems(config, items):
    for item in items:
        # 为所有异步测试添加 asyncio 标记
        if asyncio.iscoroutinefunction(item.function):
            item.add_marker(pytest.mark.asyncio)

# 配置 hook
def pytest_configure(config):
    config.addinivalue_line("markers", "slow: marks tests as slow")
    config.addinivalue_line("markers", "integration: marks tests as integration tests")

# session 级别的 hook
def pytest_sessionstart(session):
    print("Test session starting...")

def pytest_sessionfinish(session, exitstatus):
    print(f"Test session finished with exit status {exitstatus}")

@pytest.mark.parametrize("input,expected", [
    (1, 2),
    (2, 4),
    (3, 6),
])
def test_double(input, expected):
    assert input * 2 == expected

# 运行结果：3 个测试用例
# test_double[1-2] PASSED
# test_double[2-4] PASSED
# test_double[3-6] PASSED

@pytest.mark.parametrize("x", [1, 2])
@pytest.mark.parametrize("y", ["a", "b", "c"])
def test_xy(x, y):
    pass

# 生成 2 * 3 = 6 个测试用例:
# test_xy[1-a] test_xy[1-b] test_xy[1-c]
# test_xy[2-a] test_xy[2-b] test_xy[2-c]

# 命名组合使用 ids
@pytest.mark.parametrize("x,y", [
    (1, "a"),
    (2, "b"),
], ids=["case_small", "case_large"])
def test_named(x, y):
    pass

# 生成: test_named[case_small] test_named[case_large]

# 当 parametrize 值是 "db"，pytest 会查找名为 "db" 的 fixture
@pytest.fixture
def db(request):
    # request.param 接收 parametrize 传来的值
    db_type = request.param
    return Database(db_type)

@pytest.mark.parametrize("db", ["sqlite", "postgres"], indirect=True)
def test_db_query(db):
    # db fixture 会根据参数创建不同类型的数据库
    result = db.query("SELECT 1")
    assert result == [1]

# 安装
# pip install pytest-xdist

# 使用 CPU 核数作为 worker 数
pytest -n auto

# 指定 4 个 worker
pytest -n 4

# 在模块级别设置（pytest.ini 或 pyproject.toml）
# addopts = -n auto

# 只运行空闲的 worker
pytest -n auto --maxprocesses 8

# session 作用域 fixture - 所有 worker 共享
@pytest.fixture(scope="session")
def shared_config():
    # 只创建一次，所有 worker 复用
    return Config.load()

# module 作用域 fixture - 每个 worker 模块独立
@pytest.fixture(scope="module")
def worker_module_db():
    # 每个 worker 进程创建一次
    return TestDatabase()

# function 作用域 fixture - 每个测试函数独立
@pytest.fixture
def test_data():
    # 每个测试函数执行一次
    return generate_test_data()

# pip install pytest-cov coverage

# 运行覆盖率测试
pytest --cov=mypackage --cov-report=html

# 多格式报告
pytest --cov=mypackage \
      --cov-report=html \
      --cov-report=term-missing \
      --cov-report=xml

# 结合并行执行
pytest -n auto --cov=mypackage --cov-report=

# 推荐：使用 loadfile 分发策略
# 同一文件的所有测试在同一 worker 中运行
pytest -n auto --cov=mypackage --dist=loadfile

# coverage 配置文件 .coveragerc
# [run]
# source = mypackage
# parallel = True  # 启用并行模式
# [report]
# exclude_lines =
#     pragma: no cover
#     def __repr__
#     raise NotImplementedError

# 启用分支覆盖
# coverage run --branch tests/

# 代码示例
def process(value):
    if value > 0:           # Branch: value>0 True/False
        return value * 2
    else:
        return value / 2     # 需要测试 value<=0 分支

# 结合 pytest-cov
# pytest --cov=mypackage --cov-branch --cov-report=term-missing