pytest-xdist: 8x ускорение через параллельный запуск — Pytest: Профессиональные инструменты

У вас 500 тестов. Запуск занимает 10 минут. Код-ревью ждёт. CI блокирует. Как ускорить в 8 раз?

Цель: Научиться запускать тесты параллельно с pytest-xdist для драматического ускорения.

Вы точно готовы?

Убедитесь, что прошли Level 1-2:

# Умеете писать тесты с фикстурами
@pytest.fixture
def database():
    return connect_to_db()
 
def test_query(database):
    result = database.query("SELECT * FROM users")
    assert len(result) > 0

Если фикстуры или моки непонятны — вернитесь к pytest-junior.

Проблема: медленные тесты

Пример: 100 integration-тестов

# tests/test_api.py — 100 тестов
 
def test_api_endpoint_1():
    response = requests.get("https://api.example.com/users/1")
    assert response.status_code == 200
    # Каждый тест ~3 секунды
 
def test_api_endpoint_2():
    response = requests.get("https://api.example.com/users/2")
    assert response.status_code == 200
 
# ... ещё 98 тестов

Запуск:

pytest tests/

Результат:

========================= test session starts ==========================
collected 100 items
 
tests/test_api.py::test_api_endpoint_1 PASSED                    [  1%]
tests/test_api.py::test_api_endpoint_2 PASSED                    [  2%]
...
tests/test_api.py::test_api_endpoint_100 PASSED                  [100%]
 
==================== 100 passed in 300.00s (5:00) ====================

❌ 5 минут! Слишком долго для ожидания.

Проблемы:

❌ Тесты выполняются последовательно (один за другим)
❌ CPU простаивает (используется только 1 ядро из 8)
❌ Медленно на локальной машине и в CI

Решение: pytest-xdist

Установка

pip install pytest-xdist

Базовый запуск: -n auto

pytest -n auto

-n auto — автоматически определяет количество CPU и запускает столько параллельных процессов.

Результат:

========================= test session starts ==========================
8 workers [100 items]
 
test_api_endpoint_1 PASSED    test_api_endpoint_9 PASSED
test_api_endpoint_2 PASSED    test_api_endpoint_10 PASSED
test_api_endpoint_3 PASSED    test_api_endpoint_11 PASSED
...
 
==================== 100 passed in 40.00s (0:40) ====================

✅ 40 секунд вместо 300! Ускорение 7.5x

Как работает xdist

┌─────────────────────────────────────┐
│        Главный процесс (master)     │
│  - Собирает тесты                   │
│  - Распределяет по workers          │
│  - Собирает результаты              │
└──────────┬──────────────────────────┘
           │
     ┌─────┴─────┬───────┬───────────┐
     │           │       │           │
  Worker 1    Worker 2  ...     Worker  8
     │           │       │           │
  test 1-12  test 13-25 ...    test 88-100

Каждый worker:

Получает тесты от master
Выполняет их независимо
Отправляет результаты обратно

✅ Тесты выполняются параллельно на всех CPU!

Фиксированное количество workers

# 4 параллельных процесса
pytest -n 4
 
# 2 процесса (для слабых машин)
pytest -n 2
 
# 16 процессов (для мощных серверов)
pytest -n 16

Когда использовать:

pytest -n auto — для локальной разработки (автоматически)
pytest -n 4 — для CI с известными ресурсами
pytest -n 2 — если тесты требуют много памяти

Scope стратегии

Проблема: порядок выполнения

По умолчанию: тесты распределяются произвольно между workers.

Проблема: фикстуры с scope="module" создаются несколько раз!

# tests/test_database.py
 
@pytest.fixture(scope="module")
def database():
    print("🔌 Connecting to database...")  # Медленно!
    return connect_to_db()
 
def test_query_1(database):
    pass
 
def test_query_2(database):
    pass
 
# tests/test_cache.py
 
@pytest.fixture(scope="module")
def cache():
    print("🔌 Connecting to Redis...")  # Медленно!
    return RedisCache()
 
def test_cache_1(cache):
    pass

Без scope стратегии:

Worker 1: test_query_1 → 🔌 Connecting to database...
Worker 2: test_query_2 → 🔌 Connecting to database... (повторно!)
Worker 3: test_cache_1 → 🔌 Connecting to Redis...

❌ Database создаётся дважды!

--dist loadscope: группировка по scope

pytest -n auto --dist loadscope

Результат:

Worker 1: test_database.py::test_query_1, test_query_2 (одна БД!)
Worker 2: test_cache.py::test_cache_1 (один Redis!)

✅ Фикстуры с scope="module" создаются один раз на worker!

Dist стратегии

# По умолчанию: load (равномерное распределение)
pytest -n auto --dist load
 
# loadscope: группировка по scope фикстур
pytest -n auto --dist loadscope
 
# loadfile: все тесты из одного файла на одном worker
pytest -n auto --dist loadfile
 
# loadgroup: группировка по xdist_group маркеру (advanced)
pytest -n auto --dist loadgroup

Рекомендации:

--dist load — для независимых unit-тестов (default)
--dist loadscope — если есть scope="module" фикстуры ✅
--dist loadfile — если тесты зависят от порядка в файле

Подводные камни и решения

Проблема 1: Shared state в БД

# ❌ ПЛОХО — race condition!
 
def test_create_user():
    user = User.create(name="Alice")
    assert User.count() == 1  # Может быть 2, 3, ...!
 
def test_create_product():
    product = Product.create(name="Laptop")
    assert Product.count() == 1  # Может быть 2, 3, ...!

Проблема: Оба теста работают с одной БД параллельно!

Решение 1: Изоляция через transactional fixtures

@pytest.fixture(autouse=True)
def isolate_database():
    """Каждый тест в отдельной транзакции"""
    transaction = database.begin()
    yield
    transaction.rollback()  # Откатываем после теста

Решение 2: Уникальные данные для каждого worker

import pytest
 
@pytest.fixture
def worker_id(request):
    """ID текущего worker (gw0, gw1, gw2, ...)"""
    if hasattr(request.config, "workerinput"):
        return request.config.workerinput["workerid"]
    return "master"
 
def test_create_user(worker_id):
    # Уникальное имя для каждого worker
    user = User.create(name=f"Alice-{worker_id}")
    assert user.name.startswith("Alice-")

Проблема 2: Файлы и файловая система

# ❌ ПЛОХО — конфликт файлов!
 
def test_save_report():
    save_report("report.txt", data=[1, 2, 3])
    assert os.path.exists("report.txt")
 
def test_load_report():
    save_report("report.txt", data=[4, 5, 6])
    # Файл может быть перезаписан другим worker!

Решение: Уникальные имена файлов

def test_save_report(worker_id, tmp_path):
    """tmp_path автоматически уникальный!"""
    report_file = tmp_path / f"report-{worker_id}.txt"
    save_report(report_file, data=[1, 2, 3])
    assert report_file.exists()

Проблема 3: Глобальные переменные

# ❌ ПЛОХО — shared state!
 
counter = 0  # Глобальная переменная
 
def test_increment():
    global counter
    counter += 1
    assert counter == 1  # Может быть 2, 3, ...!

Решение: Избегайте глобального состояния

# ✅ ХОРОШО — состояние в фикстуре
 
@pytest.fixture
def counter():
    return {"value": 0}
 
def test_increment(counter):
    counter["value"] += 1
    assert counter["value"] == 1  # Всегда 1!

Проблема 4: Порядок тестов

# ❌ ПЛОХО — зависимость от порядка
 
def test_setup_cache():
    cache.set("key", "value")
 
def test_read_cache():
    # Предполагает что test_setup_cache уже выполнен!
    assert cache.get("key") == "value"

Решение: Независимые тесты

# ✅ ХОРОШО — каждый тест самодостаточен
 
def test_setup_cache():
    cache.set("key", "value")
    assert cache.get("key") == "value"
 
def test_read_cache():
    cache.set("key", "value")  # Setup внутри теста
    assert cache.get("key") == "value"

Измерение ускорения

Без xdist

time pytest tests/

Результат:

100 passed in 300.00s
 
real    5m0.123s
user    0m45.234s
sys     0m2.456s

С xdist

time pytest -n auto tests/

Результат:

100 passed in 40.00s
 
real    0m40.567s
user    3m12.345s  # Суммарное время всех workers
sys     0m15.678s

Ускорение:

300s / 40s = 7.5x

✅ 7.5x быстрее!

Замечания:

real time — реальное время (то что важно)
user time — суммарное CPU время (больше чем real при параллельности)

Настройка для CI

pytest.ini для команды

# pytest.ini
 
[pytest]
addopts =
    -v
    --tb=short
    # Параллельный запуск локально
    # -n auto  # Раскомментировать для локального использования
 
# Для CI переопределяем через переменную окружения

CI конфигурация

# .gitlab-ci.yml
 
test:
  script:
    # Используем фиксированное количество для предсказуемости
    - pytest -n 4 --dist loadscope
  parallel:
    matrix:
      - PYTHON_VERSION: ["3.9", "3.10", "3.11"]

GitHub Actions:

# .github/workflows/test.yml
 
- name: Run tests
  run: |
    pytest -n auto --dist loadscope

Что вы изучили

pytest-xdist — параллельный запуск тестов
-n auto — автоматическое определение CPU
--dist loadscope — оптимизация для module-scoped фикстур
Подводные камни — shared state, файлы, глобальные переменные
Решения — изоляция, уникальные данные, независимые тесты
Измерение — 7-8x ускорение в реальных проектах

Следующий урок

Отлично! Теперь тесты запускаются в 8 раз быстрее. Но как правильно организовать структуру проекта? Почему flat layout плох и что такое src layout?

Переходите к уроку 1: src layout: Правильная структура проектов

В следующем уроке вы узнаете:

Почему src layout лучше flat layout
pip install -e . для editable install
pyproject.toml для конфигурации
Правильные импорты без sys.path.append

Troubleshooting

Проблема: pytest: error: unrecognized arguments: -n Решение: pytest-xdist не установлен:

pip install pytest-xdist
pytest --version  # Должен показать pytest-xdist

Проблема: Тесты падают при параллельном запуске, но работают последовательно Решение: Есть shared state! Проверьте:

Глобальные переменные
Файлы (используйте tmp_path)
БД (используйте транзакции или unique данные с worker_id)

Проблема: Параллельный запуск медленнее чем обычный Решение:

Слишком быстрые тесты (overhead запуска workers больше чем выигрыш)
Используйте --dist loadscope для оптимизации фикстур
Проверьте что тесты CPU-bound, а не I/O-bound

Проблема: Хочу запретить параллельный запуск для некоторых тестов Решение: Используйте маркер:

@pytest.mark.xdist_group(name="serial")
def test_must_run_alone():
    pass
 
@pytest.mark.xdist_group(name="serial")
def test_must_run_alone_2():
    pass

Запуск:

pytest -n auto --dist loadgroup

Тесты с одинаковым xdist_group выполнятся на одном worker последовательно.

Проблема: Coverage показывает 0% при использовании xdist Решение: Используйте pytest-cov с правильными опциями:

pip install pytest-cov
pytest -n auto --cov=src --cov-report=html

pytest-cov автоматически объединяет coverage от всех workers.