Установка pytest-xdist, pytest -n auto, scope стратегии, подводные камни с shared state
У вас 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.
# 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 минут! Слишком долго для ожидания.
Проблемы:
pip install pytest-xdistpytest -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
┌─────────────────────────────────────┐
│ Главный процесс (master) │
│ - Собирает тесты │
│ - Распределяет по workers │
│ - Собирает результаты │
└──────────┬──────────────────────────┘
│
┌─────┴─────┬───────┬───────────┐
│ │ │ │
Worker 1 Worker 2 ... Worker 8
│ │ │ │
test 1-12 test 13-25 ... test 88-100
Каждый worker:
✅ Тесты выполняются параллельно на всех CPU!
# 4 параллельных процесса
pytest -n 4
# 2 процесса (для слабых машин)
pytest -n 2
# 16 процессов (для мощных серверов)
pytest -n 16Когда использовать:
pytest -n auto — для локальной разработки (автоматически)pytest -n 4 — для CI с известными ресурсамиpytest -n 2 — если тесты требуют много памятиПо умолчанию: тесты распределяются произвольно между 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 создаётся дважды!
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!
# По умолчанию: 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 — если тесты зависят от порядка в файле# ❌ ПЛОХО — 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-")# ❌ ПЛОХО — конфликт файлов!
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()# ❌ ПЛОХО — 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!# ❌ ПЛОХО — зависимость от порядка
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"time pytest tests/Результат:
100 passed in 300.00s
real 5m0.123s
user 0m45.234s
sys 0m2.456s
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 при параллельности)# pytest.ini
[pytest]
addopts =
-v
--tb=short
# Параллельный запуск локально
# -n auto # Раскомментировать для локального использования
# Для 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Вопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.
Далее: src layout: Правильная структура проектов