问题：并行测试共享数据库

pytest-xdist 用 -n 启动多个 worker，默认分发模式会按负载把用例分给空闲 worker。若多个用例都写同一个测试数据库，就可能出现两个 worker 同时执行 INSERT、UPDATE 或建表，导致锁等待、唯一键冲突，甚至读到中间状态。

只在用例上加 @pytest.mark.xdist_group 还不够：调度模式必须切换为 --dist loadgroup，否则该标记不会改变分发策略。

机制：按组调度

--dist loadgroup 会读取 xdist_group 标记，把组名相同的用例放到同一个 worker 中执行。组内用例在该 worker 内顺序运行，不同组之间仍然可以并行。因此，把“会写同一个数据库”的用例统一打成一个组，就能避免它们跨 worker 并发写。

需要明确：它只约束 pytest 用例的调度，不是数据库事务隔离方案。组内用例仍要负责数据清理、唯一键设计和顺序假设。

最小可运行示例

目录结构如下：

pytest-xdist-demo/
  pytest.ini
  test_shared_db.py

pytest.ini 注册标记，避免 pytest 给出未知标记警告：

[pytest]
markers =
    xdist_group: 将共享同一测试数据库的用例固定到同一个 worker

test_shared_db.py 用 SQLite 文件模拟共享测试库。前两个用例写入当前 worker 的进程号，第三个用例读取两条记录并检查是否来自同一个 worker：

import os
import sqlite3
from pathlib import Path

import pytest

DB = Path(__file__).with_name('shared_test.db')
GROUP = 'shared_db'


def write_case(case_id: str) -> None:
    con = sqlite3.connect(DB)
    try:
        con.execute(
            'CREATE TABLE IF NOT EXISTS case_worker ('
            'case_id TEXT PRIMARY KEY, '
            'worker_pid INTEGER NOT NULL)'
        )
        con.execute(
            'INSERT OR REPLACE INTO case_worker(case_id, worker_pid) '
            'VALUES (?, ?)',
            (case_id, os.getpid()),
        )
        con.commit()
    finally:
        con.close()


@pytest.mark.xdist_group(GROUP)
def test_write_case_a():
    write_case('a')


@pytest.mark.xdist_group(GROUP)
def test_write_case_b():
    write_case('b')


@pytest.mark.xdist_group(GROUP)
def test_same_worker_for_shared_db():
    con = sqlite3.connect(DB)
    try:
        rows = con.execute(
            'SELECT DISTINCT worker_pid FROM case_worker '
            'WHERE case_id IN (?, ?)',
            ('a', 'b'),
        ).fetchall()
    finally:
        con.close()

    assert len(rows) == 1, f'共享库用例被分配到多个 worker: {rows}'

三个用例都访问同一个 SQLite 文件。 test_write_case_a 和 test_write_case_b 把当前进程号写入表，test_same_worker_for_shared_db 查询 a、b 两条记录的进程号并去重。若调度正确，应该只有一个进程号；若前两个用例被分到不同 worker，断言会失败。这个检查用于演示分组效果，真实项目可以换成 fixture 汇总或日志核对。

运行方法

先安装依赖：

python -m pip install pytest pytest-xdist

使用 -n auto 和 --dist loadgroup 运行：

python -m pytest -n auto --dist loadgroup -q

如果想更直观地观察 worker 进程，可显式启动两个 worker，并保留测试输出：

python -m pytest -n 2 --dist loadgroup -s -q

-n 2 比 -n auto 更适合在单核或资源受限环境验证并行行为，因为 -n auto 可能只启动一个 worker。若组内断言通过，说明共享库用例被固定到了同一个 worker。不要用普通 -n auto 且不带 --dist loadgroup 的方式期待 xdist_group 生效；默认分发模式不会按该标记分组。

整个测试模块都访问同一个数据库时，可以用模块级标记减少重复：

pytestmark = pytest.mark.xdist_group('db_main')

组名要统一，并且大小写敏感。不同模块使用同一个组名，也会被视为同一组。

边界与限制

• 只解决调度隔离：同组用例在同一个 worker 内串行，但不同组仍可能同时访问同一个数据库。如果共享范围扩大，应合并组名或拆分测试库。
• 不提供事务隔离：组内用例仍要自己清理数据、使用唯一键、避免依赖执行顺序。示例的第三个用例依赖 pytest 默认收集顺序，仅用于演示；生产测试可用 session 级 fixture 或收集阶段钩子汇总。
• session fixture 会在每个 worker 中各初始化一次。不要在 fixture 中创建跨 worker 共享的连接对象；共享文件路径可以，共享连接不行。
• -n auto 在单核环境可能只启动一个 worker，隔离效果不易观察；验证时优先使用 -n 2。
• SQLite 的文件锁和数据库连接超时仍需按实际数据库配置处理。loadgroup 不能替代连接池、事务和重试设计。示例生成的 shared_test.db 应加入 .gitignore，并在 CI 中清理。
• 如果 pytest-xdist 版本不支持 --dist loadgroup，命令会提示未知分发模式，此时应升级依赖，不要依赖未经验证的参数组合。

小结

把共享数据库的测试用例打上同一个 xdist_group，再用 --dist loadgroup 运行，是 pytest-xdist 下成本较低的防跨 worker 数据竞争方式。它适合少量用例共享同一测试库的场景；若每个 worker 都能拥有独立库或 schema，优先做物理隔离，调度分组只作为补充。