pytest-xdist 插件的介绍和使用

简介：

pytest-xdist 是非常重要的插件。通过多进程方式，并行执行测试用例来显著提升测试效率（充分利用多核 CPU 资源，加快整体测试执行速度）。特别适合那些测试用例较多、耗时较长的项目。

准备：

安装插件：

pip install pytest-xdist

运行方式：

1、自动使用所有 CPU 核心

pytest -n auto

• -n auto：自动检测并使用与 CPU 核心数相同的进程数

2、指定使用固定数量的进程

pytest -n 4

• 使用 4 个进程并行执行测试

3、配置 pytest.ini 以默认启用 -n auto（推荐）

在 pytest.ini 中使用 addopts 选项来设置默认参数。

[pytest]
addopts = -v -n auto

注意事项：

✅ 测试之间必须独立！

• 如果多个测试共享资源（如数据库、文件、全局变量），可能会导致冲突
• 建议使用 fixture 创建隔离的测试环境

✅ 日志和输出可能乱序

• 并行执行时，不同进程的输出会混在一起
• 可使用 pytest --capture=no 或生成 HTML 报告查看详细结果

✅ 某些插件不兼容

• 有些插件（如 pytest-cov）需要特别处理才能正确统计覆盖率
• 使用 --no-cov 可临时禁用覆盖率统计

示例使用：

全局变量隔离：

使用全局变量时，需使用 @pytest.fixture避免用例间冲突

import pytest


@pytest.fixture
def isolated_data():
    return []


def test_1(isolated_data):
    isolated_data.append(1)
    assert isolated_data == [1]


def test_2(isolated_data):
    isolated_data.append(2)
    assert isolated_data == [2]

数据库隔离：

MySQL + pytest-xdist

1、使用 fixture 提供隔离的数据库连接。

# conftest.py

import pytest
import mysql.connector
from mysql.connector import errorcode
import uuid

@pytest.fixture(scope="session")
def mysql_db():
    # 连接 MySQL 服务器（注意不要连接具体数据库）
    conn = mysql.connector.connect(
        host="localhost",
        user="root",
        password="your_password"
    )
    cursor = conn.cursor()

    # 生成唯一数据库名（使用 UUID）
    db_name = f"testdb_{uuid.uuid4().hex}"
  
    try:
        cursor.execute(f"CREATE DATABASE {db_name}")
        conn.commit()
    finally:
        cursor.close()
        conn.close()

    # 新建连接到新数据库
    db_conn = mysql.connector.connect(
        host="localhost",
        user="root",
        password="your_password",
        database=db_name
    )

    yield db_conn

    # 清理：删除测试数据库
    db_conn.close()
    drop_conn = mysql.connector.connect(
        host="localhost",
        user="root",
        password="your_password"
    )
    drop_cursor = drop_conn.cursor()
    drop_cursor.execute(f"DROP DATABASE IF EXISTS {db_name}")
    drop_conn.commit()
    drop_cursor.close()
    drop_conn.close()

测试中使用隔离的数据库连接：

# test_mysql.py

def test_insert_data(mysql_db):
    cursor = mysql_db.cursor()
    cursor.execute("CREATE TABLE test_table (id INT PRIMARY KEY, name VARCHAR(255))")
    cursor.execute("INSERT INTO test_table VALUES (1, 'Alice')")
    mysql_db.commit()

    cursor.execute("SELECT * FROM test_table")
    result = cursor.fetchall()
    assert result == [(1, 'Alice')]

2、使用事务隔离（推荐）

在每个测试中使用事务，并在测试结束后回滚，避免污染数据。

使用 session 级数据库连接 + function 级事务，兼顾性能与隔离性。

# conftest.py

import pytest
import mysql.connector
from mysql.connector import errorcode
import uuid

@pytest.fixture(scope="session")
def mysql_db():
    # 连接 MySQL 服务器（注意不要连接具体数据库）
    conn = mysql.connector.connect(
        host="localhost",
        user="root",
        password="your_password"
    )
    cursor = conn.cursor()

    # 生成唯一数据库名（使用 UUID）
    db_name = f"testdb_{uuid.uuid4().hex}"
  
    try:
        cursor.execute(f"CREATE DATABASE {db_name}")
        conn.commit()
    finally:
        cursor.close()
        conn.close()

    # 新建连接到新数据库
    db_conn = mysql.connector.connect(
        host="localhost",
        user="root",
        password="your_password",
        database=db_name
    )

    yield db_conn

    # 清理：删除测试数据库
    db_conn.close()
    drop_conn = mysql.connector.connect(
        host="localhost",
        user="root",
        password="your_password"
    )
    drop_cursor = drop_conn.cursor()
    drop_cursor.execute(f"DROP DATABASE IF EXISTS {db_name}")
    drop_conn.commit()
    drop_cursor.close()
    drop_conn.close()


@pytest.fixture(scope="function")
def mysql_cursor(mysql_db):
    cursor = mysql_db.cursor()
    mysql_db.start_transaction()
    yield cursor
    mysql_db.rollback()
    cursor.close()

然后在测试中使用：

def test_with_transaction(mysql_cursor):
    mysql_cursor.execute("CREATE TABLE test_table (id INT PRIMARY KEY, name VARCHAR(255))")
    mysql_cursor.execute("INSERT INTO test_table VALUES (1, 'Alice')")

    mysql_cursor.execute("SELECT * FROM test_table")
    result = mysql_cursor.fetchall()
    assert result == [(1, 'Alice')]

所有操作在事务中进行，测试结束后自动回滚，数据库干净无污染。