Pytest-xdist：如何在所有Worker结束后执行tearDown操作

针对pytest-xdist并行场景下，每个Worker都会执行session级fixture的teardown（数据库降级）的问题，我们可以通过共享计数文件+文件锁的方式，确保只有最后一个退出的Worker执行数据库降级操作。以下是具体实现方案：

核心思路 #

1. 用一个共享文件记录当前活跃的Worker数量
2. 每个Worker启动时，通过文件锁原子递增计数；只有计数从0变为1时，执行数据库升级（对应你提到的setup文件锁方案）
3. 每个Worker退出时，通过文件锁原子递减计数；当计数变为0时，执行数据库降级操作

实现代码 #

import pytest
import os
import subprocess
import fcntl
from pathlib import Path

# 定义共享计数文件路径，放在项目根目录
WORKER_COUNT_FILE = Path(__file__).parent / ".worker_count"

@pytest.fixture(autouse=True, scope='session')
def handle_database():
    """Create necessary tables in db (using migration) and drop them after all tests"""
    original_dir = os.getcwd()
    migration_dir = os.path.join(original_dir, 'test/api_simulator')
    
    # --- Setup: 仅首个Worker执行数据库升级 ---
    with open(WORKER_COUNT_FILE, "a+") as f:
        fcntl.flock(f, fcntl.LOCK_EX)  # 获取排他锁
        f.seek(0)
        count = int(f.read() or 0)
        if count == 0:
            # 首个Worker执行升级
            os.chdir(migration_dir)
            subprocess.run(['flask', 'db', 'upgrade'], check=True)
            print('\n\nUpgrade migration completed\n\n')
            os.chdir(original_dir)
        # 递增计数并写入
        count += 1
        f.seek(0)
        f.truncate()
        f.write(str(count))
        fcntl.flock(f, fcntl.LOCK_UN)  # 释放锁

    yield

    # --- Teardown: 仅最后一个Worker执行数据库降级 ---
    with open(WORKER_COUNT_FILE, "a+") as f:
        fcntl.flock(f, fcntl.LOCK_EX)
        f.seek(0)
        count = int(f.read() or 0)
        count -= 1
        if count == 0:
            # 最后一个Worker执行降级
            os.chdir(migration_dir)
            subprocess.run(['flask', 'db', 'downgrade'], check=True)
            print('\n\nDowngrade migration completed\n\n')
            os.chdir(original_dir)
            # 清空计数文件
            f.truncate()
        else:
            # 更新计数
            f.seek(0)
            f.truncate()
            f.write(str(count))
        fcntl.flock(f, fcntl.LOCK_UN)

# 可选：测试结束后清理计数文件
def pytest_sessionfinish(session, exitstatus):
    if WORKER_COUNT_FILE.exists():
        try:
            WORKER_COUNT_FILE.unlink()
        except:
            pass

关键细节说明 #

• 文件锁的作用：fcntl.LOCK_EX确保多Worker对计数文件的读写是原子操作，避免并发导致的计数错误
• 计数逻辑：每个Worker启动时计数+1，退出时-1；只有计数归0时，说明所有Worker都已完成测试，此时执行降级
• 路径处理：用绝对路径替代相对路径，避免Worker进程的工作目录不一致导致的问题
• 异常处理：subprocess.run添加check=True，确保迁移命令执行失败时直接抛出异常，避免测试在数据库状态异常的情况下继续运行
• 清理钩子：pytest_sessionfinish用于在整个测试会话结束后删除计数文件，避免残留文件影响下次测试

内容的提问来源于stack exchange，提问作者chlebastian