Python测试用例如何实现并发加锁_利用pytest-xdist共享信号-Python教程-PHP中文网

pytest-xdist 默认以独立进程运行 worker，threading.lock 和 multiprocessing.lock 均无效；应使用文件锁（flock）或 pytest-xdist 的 --dist=loadgroup 分组机制实现跨进程同步。

python测试用例如何实现并发加锁_利用pytest-xdist共享信号

pytest-xdist 默认不共享进程间状态

你写的 threading.Lock 或 multiprocessing.Lock 在 pytest-xdist 下基本没用——因为默认每个 worker 是独立子进程，内存不共享，锁对象各自实例化，形同虚设。

常见错误现象：FileExistsError 写临时文件时撞车、sqlite3.OperationalError: database is locked、或者测试逻辑里靠全局变量计数/标记状态却始终为初始值。

pytest-xdist 的 -n 启动的是多个独立 Python 进程（不是线程），threading 模块完全无效
multiprocessing.Lock 仅在 fork 模式下可能复用，但 xdist 默认用 spawn（尤其 Windows/macOS），锁无法继承或传递
真正能跨 worker 共享的只有外部资源：文件、数据库、Redis、本地 socket 或文件锁（flock）

用文件锁（flock）做轻量级跨进程互斥

不需要引入 Redis 或启动额外服务，一个本地文件 + flock 就能实现可靠加锁，适合控制测试中“只允许一个 worker 执行某段逻辑”这类场景。

使用场景：初始化一次数据库 schema、生成一份共享 fixture 数据、清理某个临时目录、调用一次外部 mock server 启停。

macOS UI 自动化技能包 —— 用 Python 控制你的 Mac

让 macOS 自动化像说话一样简单 —— 用代码控制你的 Mac，无需学习 AppleScript 系统要求 macOS 14+ (Sonoma) Python 3.9+ 需要开启：辅助功能 + 屏幕录制权限

下载

Python 标准库 fcntl.flock 只支持 Unix/Linux/macOS；Windows 需用 msvcrt.locking 或改用 portalocker 库
锁文件路径必须所有 worker 进程可读写，推荐放在项目根目录下明确路径，比如 ./.test-lock
务必在 finally 块中释放锁，否则一个 worker 挂掉会导致锁永久占用

import fcntl
import os

def acquire_global_lock(lock_path="./.test-lock"):
    fd = os.open(lock_path, os.O_CREAT | os.O_RDWR)
    try:
        fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
        return fd
    except (OSError, IOError):
        os.close(fd)
        raise RuntimeError(f"Failed to acquire lock on {lock_path}")

def release_global_lock(fd):
    fcntl.flock(fd, fcntl.LOCK_UN)
    os.close(fd)

pytest fixture 中安全封装共享锁逻辑

把锁操作包装成 session-scoped fixture，让需要串行执行的测试自动排队，避免在每个 test 函数里手写锁逻辑出错。

参数差异：注意 scope="session" 是跨 worker 共享的关键，而 autouse=True 容易误触发——只应在明确需要的测试类或模块上显式 use。

fixture 返回的是锁的持有状态（如 fd），不是锁本身；释放必须在 fixture teardown 阶段完成
不能依赖 yield 自动 cleanup：worker 进程退出不一定走 yield 后逻辑，必须用 request.addfinalizer
如果锁获取失败（比如超时），建议抛出 pytest.skip() 而不是 raise，避免整个 session 失败

import pytest
import fcntl
import os

@pytest.fixture(scope="session")
def global_init_lock(request):
    lock_fd = None
    lock_path = "./.test-init-lock"
    
    try:
        lock_fd = os.open(lock_path, os.O_CREAT | os.O_RDWR)
        fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
    except (OSError, IOError):
        pytest.skip("Another worker holds the init lock")

    def _release():
        if lock_fd is not None:
            try:
                fcntl.flock(lock_fd, fcntl.LOCK_UN)
            finally:
                os.close(lock_fd)
    request.addfinalizer(_release)
    return lock_fd