Pytest-xdist如何处理数据污染

在使用 pytest-xdist 进行并行测试时，数据污染（如共享数据库/文件被多个进程同时修改）是一个常见问题。以下是系统化的解决方案，结合不同场景的代码示例：

一、数据污染的根本原因

二、解决方案与代码实践

1. 隔离测试数据（核心策略）

方案A：动态生成唯一数据

import uuid

def test_user_creation(db_connection):
    # 每个测试生成唯一用户名
    username = f"user_{uuid.uuid4().hex[:8]}"
    db_connection.create_user(username)
    assert db_connection.get_user(username) is not None

方案B：参数化隔离数据

@pytest.mark.parametrize("user_id", ["test_001", "test_002", "test_003"])
def test_delete_user(db_connection, user_id):
    db_connection.delete_user(user_id)
    assert not db_connection.user_exists(user_id)

2. 使用进程隔离的 Fixture

@pytest.fixture(scope="function")
def temp_db(request):
    # 每个测试函数创建独立数据库
    db_name = f"test_db_{os.getpid()}_{hash(request.node.name)}"
    db = create_database(db_name)
    yield db
    db.drop()

3. 文件操作防护

@pytest.fixture
def isolated_temp_file(tmp_path):
    # 每个测试获取唯一文件路径
    file_path = tmp_path / f"data_{os.getpid()}.txt"
    with open(file_path, "w") as f:
        f.write("initial data")
    return file_path

def test_file_operations(isolated_temp_file):
    with open(isolated_temp_file, "a") as f:
        f.write("_appended")
    # 其他进程不会操作同一文件

4. 数据库事务回滚

@pytest.fixture
def db_transaction(db_connection):
    # 开始事务
    db_connection.begin()
    yield db_connection
    # 测试结束后回滚
    db_connection.rollback()

def test_payment(db_transaction):
    db_transaction.execute("INSERT INTO payments VALUES (...)")
    # 无论测试成功与否，数据都不会持久化

三、pytest-xdist 专用技巧

1. 通过 worker_id 隔离资源

def test_worker_specific_data(request):
    worker_id = request.config.workerinput["workerid"]
    data = f"data_for_{worker_id}"
    assert process_data(data) == expected_result

2. 同步锁控制关键段

from filelock import FileLock

def test_with_shared_resource(tmp_path):
    lock_file = tmp_path / "lock"
    with FileLock(lock_file):
        # 临界区代码（只有一个进程能进入）
        modify_shared_resource()

3. 全局资源池管理

import pytest
from multiprocessing import Manager

@pytest.fixture(scope="session")
def resource_pool():
    with Manager() as manager:
        pool = manager.dict()
        yield pool

def test_use_resource(resource_pool):
    resource_id = f"res_{os.getpid()}"
    resource_pool[resource_id] = allocate_resource()
    # 其他进程通过 pool 字典协调资源

四、不同场景的解决方案对比

五、调试与验证

1. 检测并行冲突

# 运行测试并打印 Worker ID
pytest -n 2 --dist=loadfile -v

2. 日志追踪

def test_with_logging(request):
    worker_id = request.config.workerinput.get("workerid", "local")
    print(f"\n[Worker-{worker_id}] Running test: {request.node.name}")
    # 测试逻辑...

3. 资源监控脚本

# conftest.py
@pytest.hookimpl(tryfirst=True)
def pytest_runtest_protocol(item):
    print(f"Worker {os.getpid()} handling {item.nodeid}")

六、最佳实践总结

1. 隔离优先：始终假设测试会并行运行，提前设计隔离策略。
2. 原子操作：单个测试应包含完整的 setup/action/assert 流程。
3. 清理保障：使用 Fixture 的 yield 或 addfinalizer 确保资源释放。
4. 避免全局状态：禁用单例模式，改用依赖注入。
5. 选择性并行：对资源敏感的测试标记为 @pytest.mark.serial，用 -m "not serial" 过滤。

通过以上方法，可以在享受 pytest-xdist 并行加速的同时，彻底解决数据污染问题。