在使用 pytest-xdist 進行并行測試時，數(shù)據(jù)污染（如共享數(shù)據(jù)庫/文件被多個進程同時修改）是一個常見問題。以下是系統(tǒng)化的解決方案，結(jié)合不同場景的代碼示例：

一、數(shù)據(jù)污染的根本原因

二、解決方案與代碼實踐

1. 隔離測試數(shù)據(jù)（核心策略）

方案A：動態(tài)生成唯一數(shù)據(jù)

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：參數(shù)化隔離數(shù)據(jù)

@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):
    # 每個測試函數(shù)創(chuàng)建獨立數(shù)據(jù)庫
    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. 數(shù)據(jù)庫事務(wù)回滾

@pytest.fixture
def db_transaction(db_connection):
    # 開始事務(wù)
    db_connection.begin()
    yield db_connection
    # 測試結(jié)束后回滾
    db_connection.rollback()

def test_payment(db_transaction):
    db_transaction.execute("INSERT INTO payments VALUES (...)")
    # 無論測試成功與否，數(shù)據(jù)都不會持久化

三、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. 同步鎖控制關(guān)鍵段

from filelock import FileLock

def test_with_shared_resource(tmp_path):
    lock_file = tmp_path / "lock"
    with FileLock(lock_file):
        # 臨界區(qū)代碼（只有一個進程能進入）
        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 字典協(xié)調(diào)資源

四、不同場景的解決方案對比

五、調(diào)試與驗證

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. 資源監(jiān)控腳本

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

六、最佳實踐總結(jié)

1. 隔離優(yōu)先：始終假設(shè)測試會并行運行，提前設(shè)計隔離策略。
2. 原子操作：單個測試應(yīng)包含完整的 setup/action/assert 流程。
3. 清理保障：使用 Fixture 的 yield 或 addfinalizer 確保資源釋放。
4. 避免全局狀態(tài)：禁用單例模式，改用依賴注入。
5. 選擇性并行：對資源敏感的測試標記為 @pytest.mark.serial，用 -m "not serial" 過濾。

通過以上方法，可以在享受 pytest-xdist 并行加速的同時，徹底解決數(shù)據(jù)污染問題。