1.并行运行测试

pytest-xdist 最常见的用途是并行运行测试用例,可以使用 -n 参数来指定并行执行的测试数。例如,如果希望将测试分配到 4 个 CPU 核心上并行运行,可以使用以下命令:

pytest -n 4

这会将测试分配到 4 个工作进程中,每个进程将并行执行测试。可以根据机器上 CPU 核心的数量来调整并行数。

2.分布式测试

pytest-xdist 也可以在多台机器上分布式地运行测试,这样可以进一步提高测试的执行速度。需要配置多个节点来运行测试并将测试分配给这些节点。首先,需要启动一个主节点,然后将其他机器作为工作节点加入。

pytest --dist=loadscope --tx ssh=remote_machine//python=python3 --maxfail=1 --disable-warnings

在这个命令中:

  • --dist=loadscope:指定测试分发的方式,可以按文件、类、作用域等进行分发。

  • --tx ssh=remote_machine//python=python3:指定远程节点运行测试。remote_machine 是远程机器的地址,python=python3 是远程机器上 Python 的路径。

  • --maxfail=1:在测试失败时中止测试。

  • --disable-warnings:禁用警告信息。

3.使用负载平衡

pytest-xdist 默认会对测试进行负载平衡,即它会尽可能平均地分配测试用例到不同的工作进程中。如果希望改变负载平衡的方式,可以使用 --dist 参数指定不同的策略:

  • --dist=loadscope:按照测试的作用域(例如类、方法)分配测试。

  • --dist=loadfile:按照测试文件分配测试。

  • --dist=none:不使用负载平衡,直接将测试用例平分到工作进程。

4.实际场景

如何使用 scope=sessionfixture 来确保在整个测试会话中只执行一次某个操作(比如获取并共享数据)

test.demo.py:

class TestCase():
    def test_01(self,login):
        with open('./text.txt','w') as f:
            f.write(login)
        print("--1--")
    def test_02(self,login):
        with open('./text2.txt','w') as f:
            f.write(login)
        print("--2--")
    def test_03(self,login):
        with open('./text3.txt','w') as f:
            f.write(login)
        print("--3--")

conftest.py:

import json
import os
import uuid
import pytest
from filelock import FileLock

@pytest.fixture(scope='session')
def login(tmp_path_factory, worker_id):
    if worker_id == 'master':
        # 【伪代码:获取登录id】
        uuid_value = str(uuid.uuid4())
        print("fixture: 请求登录接口,获取token", uuid_value)
        os.environ['token'] = uuid_value
        return uuid_value

    root_tmp_dir = tmp_path_factory.getbasetemp().parent
    fn = root_tmp_dir / "data.json"
    with FileLock(str(fn) + '.lock'):
        if fn.is_file():
            token = json.loads(fn.read_text())
            print("读取文件的token:{}".format(token))
        else:
            # 【伪代码:获取登录id】
            token = str(uuid.uuid4())
            print("fixture: 请求登录接口,获取token", token)
            fn.write_text(json.dumps(token))
            print("首次运行,token:{}".format(token))
        os.environ['token'] = token
    return token

这段代码解决什么样的问题?

scope='session' 的 fixture 在单进程下确实只会执行一次;

但用 pytest-xdist -n N 并行时,会有 N 个独立的 Python 进程,每个进程都会各自执行一次 session 级 fixture——这就“多次”了,造成获取到不同的 token。

这段代码用“共享文件 + 文件锁”把 token 存在公共位置,保证并行的多个进程拿到的是同一个 token,从效果上实现了“会话只执行一次”的目的。

执行流程(并行时):

  1. 第一个抢到锁的进程发现没有文件 → 生成 token → 写文件 → 解锁。

  2. 其他进程依次拿到锁时,文件已经存在 → 读取同一个 token。

  3. 每个进程都把这个 token 放入自己的 os.environ,并把它作为 fixture 返回值供测试使用。
    → 所有进程拿到的都是同一个 token