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=session 的 fixture 来确保在整个测试会话中只执行一次某个操作(比如获取并共享数据)
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,从效果上实现了“会话只执行一次”的目的。
执行流程(并行时):
第一个抢到锁的进程发现没有文件 → 生成 token → 写文件 → 解锁。
其他进程依次拿到锁时,文件已经存在 → 读取同一个 token。
每个进程都把这个 token 放入自己的
os.environ,并把它作为 fixture 返回值供测试使用。
→ 所有进程拿到的都是同一个 token。